query_runtime/runtime/query_service/
config.rs

1//! Queue bounds and the execution policy copied into admitted contexts.
2//!
3//! Both values are validated once, by `GpuBackendConfig::validate`, before a
4//! backend claims any process-wide state. Device-local capacity lives in the
5//! device profile (see `website/docs/design/admission.mdx#device-profiles-and-grants`).
6
7use std::time::{Duration, Instant};
8
9use query_planning::{
10    capability::AggregateStrategyExecutionOptions,
11    chunk::SourceChunkPlanningOptions,
12    error::{QueryError, QueryErrorCode, Result},
13    execution_contracts::{MemoryControlPolicy, SourceChunkPlanningMode},
14    mode::StatisticalAggregateExecutionMode,
15    policy::PlanningPolicy,
16};
17
18use crate::runtime::{
19    LakehouseCacheFillConfig, SourceSchedulerPolicy, SourceStorageProfile, object_store::SourceObjectStoreRegistry,
20};
21
22/// Bounded admission-queue policy.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct QueueBounds {
25    /// Maximum queued waiters. Must be nonzero. Pending submissions that have
26    /// not yet been queued count toward it, so a burst of submissions is
27    /// rejected with `admission_queue_full` before it can overfill the queue.
28    pub max_waiters: usize,
29    /// Maximum time one waiter may remain queued. Must be nonzero.
30    pub max_wait: Duration,
31}
32
33impl QueueBounds {
34    /// Canonical bounded-admission queue defaults shared by every frontend.
35    ///
36    /// Frontends may project these values into their own configuration
37    /// surfaces, but must not restate them. Keeping the construction here
38    /// makes the runtime controller the sole owner of queue defaults.
39    pub(crate) const fn canonical_default() -> Self {
40        Self { max_waiters: 64, max_wait: Duration::from_secs(30) }
41    }
42
43    /// The latest instant a wait that starts at `start` may last: `max_wait`
44    /// after `start`, or the caller's `requested` deadline when that is
45    /// earlier. A `max_wait` too long to add to `start` sets no limit of its
46    /// own.
47    pub(crate) fn wait_deadline(&self, start: Instant, requested: Option<Instant>) -> Option<Instant> {
48        let configured = start.checked_add(self.max_wait);
49        match (requested, configured) {
50            (Some(requested), Some(configured)) => Some(requested.min(configured)),
51            (requested, configured) => requested.or(configured),
52        }
53    }
54
55    pub(in crate::runtime) fn validate(&self) -> Result<()> {
56        validate_nonzero_config_field("queue_bounds.max_waiters", self.max_waiters == 0)?;
57        validate_nonzero_config_field("queue_bounds.max_wait", self.max_wait.is_zero())
58    }
59}
60
61impl Default for QueueBounds {
62    fn default() -> Self {
63        Self::canonical_default()
64    }
65}
66
67/// Execution policy copied into admitted native query contexts.
68#[derive(Debug, Clone, PartialEq, Eq)]
69pub struct QueryExecutionConfig {
70    /// Lakehouse cache-fill policy and shared resource limits.
71    pub lakehouse_cache_fill: LakehouseCacheFillConfig,
72    /// Source object-store profiles available to service-owned execution contexts.
73    pub source_storage_profiles: Vec<SourceStorageProfile>,
74    /// Bounded source scheduling policy copied into admitted contexts.
75    pub source_scheduler_policy: SourceSchedulerPolicy,
76    /// Source chunk planning mode copied into admitted contexts.
77    pub source_chunk_planning_mode: SourceChunkPlanningMode,
78    /// Floating statistical aggregate execution contract copied into admitted
79    /// contexts.
80    pub statistical_aggregate_execution_mode: StatisticalAggregateExecutionMode,
81    /// Join and dynamic-filter thresholds copied into admitted contexts; a
82    /// plan decided without an admission artifact is admitted under it.
83    pub planning_policy: PlanningPolicy,
84    /// Grouped aggregate strategy execution policy copied into admitted contexts.
85    pub aggregate_strategy_execution_options: AggregateStrategyExecutionOptions,
86    /// Expert local-throughput source chunk planning policy copied into admitted contexts.
87    pub source_chunk_planning_options: SourceChunkPlanningOptions,
88    /// Source chunk byte target copied into admitted bounded contexts.
89    /// `u64::MAX` derives a finite target from the immutable grant.
90    pub max_source_chunk_bytes: u64,
91    /// Source row-group target copied into admitted bounded contexts.
92    pub max_row_groups_per_chunk: usize,
93    /// Bounded retry-attempt cap copied into admitted bounded contexts.
94    pub max_retry_attempts: usize,
95    /// Bounded retry row-group floor copied into admitted bounded contexts.
96    pub min_retry_chunk_row_groups: usize,
97}
98
99impl QueryExecutionConfig {
100    pub(in crate::runtime) fn validate(&self) -> Result<()> {
101        self.lakehouse_cache_fill.validate()?;
102        self.source_scheduler_policy.validate()?;
103        self.planning_policy.validate()?;
104        self.aggregate_strategy_execution_options.validate()?;
105        self.source_chunk_planning_options.validate()?;
106        validate_nonzero_config_field("max_source_chunk_bytes", self.max_source_chunk_bytes == 0)?;
107        validate_nonzero_config_field("max_row_groups_per_chunk", self.max_row_groups_per_chunk == 0)?;
108        validate_nonzero_config_field("max_retry_attempts", self.max_retry_attempts == 0)?;
109        validate_nonzero_config_field("min_retry_chunk_row_groups", self.min_retry_chunk_row_groups == 0)?;
110        SourceObjectStoreRegistry::validate_profiles(&self.source_storage_profiles)
111    }
112
113    pub(in crate::runtime) fn memory_policy_for_admitted_query(&self, device_budget_bytes: u64) -> MemoryControlPolicy {
114        MemoryControlPolicy {
115            device_budget_bytes,
116            max_source_chunk_bytes: self.max_source_chunk_bytes,
117            max_row_groups_per_chunk: self.max_row_groups_per_chunk,
118            max_retry_attempts: self.max_retry_attempts,
119            min_retry_chunk_row_groups: self.min_retry_chunk_row_groups,
120        }
121    }
122}
123
124impl Default for QueryExecutionConfig {
125    fn default() -> Self {
126        Self {
127            lakehouse_cache_fill: LakehouseCacheFillConfig::default(),
128            source_storage_profiles: Vec::new(),
129            source_scheduler_policy: SourceSchedulerPolicy::default(),
130            source_chunk_planning_mode: SourceChunkPlanningMode::default(),
131            statistical_aggregate_execution_mode: StatisticalAggregateExecutionMode::default(),
132            planning_policy: PlanningPolicy::default(),
133            aggregate_strategy_execution_options: AggregateStrategyExecutionOptions::default(),
134            source_chunk_planning_options: SourceChunkPlanningOptions::default(),
135            max_source_chunk_bytes: MemoryControlPolicy::DEFAULT_MAX_SOURCE_CHUNK_BYTES,
136            max_row_groups_per_chunk: MemoryControlPolicy::DEFAULT_MAX_ROW_GROUPS_PER_CHUNK,
137            max_retry_attempts: MemoryControlPolicy::DEFAULT_MAX_RETRY_ATTEMPTS,
138            min_retry_chunk_row_groups: MemoryControlPolicy::DEFAULT_MIN_RETRY_CHUNK_ROW_GROUPS,
139        }
140    }
141}
142
143fn validate_nonzero_config_field(field: &'static str, is_zero: bool) -> Result<()> {
144    if is_zero {
145        return Err(QueryError::new(
146            QueryErrorCode::InvalidConfiguration,
147            "query_engine.query_service.config.validate",
148            "query service config field must be greater than zero",
149        )
150        .with_fact("field", field)
151        .with_fact("reason", "zero"));
152    }
153    Ok(())
154}
query_runtime/runtime/query_service/config.rs

query_runtime/runtime/query_service/
config.rs
