algeon_datafusion/native_exec/config/
chunking.rs

1//! Source chunking settings of native reads.
2
3use std::num::{NonZeroU64, NonZeroUsize};
4
5use query_planning::execution_contracts::MemoryControlPolicy;
6
7use crate::error::{Error, ErrorCode, Result};
8
9/// Source chunking of native reads. Device capacity always comes from the
10/// admitted grant, so a chunking carries no device budget.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub struct SourceChunking {
13    /// Source bytes planned into one chunk; a planning target.
14    /// `u64::MAX` derives the target from the immutable admitted grant.
15    pub max_source_chunk_bytes: NonZeroU64,
16    /// Row groups planned into one chunk.
17    pub max_row_groups_per_chunk: NonZeroUsize,
18    /// Attempts a failed chunk read retries with smaller chunks.
19    pub max_retry_attempts: NonZeroUsize,
20    /// Row-group floor of a retried chunk.
21    pub min_retry_chunk_row_groups: NonZeroUsize,
22}
23
24impl SourceChunking {
25    /// The configured source-byte ceiling, or `None` when the default leaves
26    /// the target to the admitted grant.
27    pub fn explicit_max_source_chunk_bytes(&self) -> Option<NonZeroU64> {
28        (self.max_source_chunk_bytes != Self::DEFAULT_MAX_SOURCE_CHUNK_BYTES).then_some(self.max_source_chunk_bytes)
29    }
30
31    /// Default source-byte setting, without a configured ceiling.
32    pub const DEFAULT_MAX_SOURCE_CHUNK_BYTES: NonZeroU64 =
33        NonZeroU64::new(MemoryControlPolicy::DEFAULT_MAX_SOURCE_CHUNK_BYTES).unwrap();
34    /// Default row groups planned into one chunk.
35    pub const DEFAULT_MAX_ROW_GROUPS_PER_CHUNK: NonZeroUsize =
36        NonZeroUsize::new(MemoryControlPolicy::DEFAULT_MAX_ROW_GROUPS_PER_CHUNK).unwrap();
37    /// Default retry attempts of a failed chunk read.
38    pub const DEFAULT_MAX_RETRY_ATTEMPTS: NonZeroUsize =
39        NonZeroUsize::new(MemoryControlPolicy::DEFAULT_MAX_RETRY_ATTEMPTS).unwrap();
40    /// Default row-group floor of a retried chunk.
41    pub const DEFAULT_MIN_RETRY_CHUNK_ROW_GROUPS: NonZeroUsize =
42        NonZeroUsize::new(MemoryControlPolicy::DEFAULT_MIN_RETRY_CHUNK_ROW_GROUPS).unwrap();
43
44    /// Chunking from plain counts; each must be greater than zero.
45    pub fn try_new(
46        max_source_chunk_bytes: u64,
47        max_row_groups_per_chunk: usize,
48        max_retry_attempts: usize,
49        min_retry_chunk_row_groups: usize,
50    ) -> Result<Self> {
51        Ok(Self {
52            max_source_chunk_bytes: NonZeroU64::new(max_source_chunk_bytes)
53                .ok_or_else(|| zero_source_chunking_field("max_source_chunk_bytes"))?,
54            max_row_groups_per_chunk: NonZeroUsize::new(max_row_groups_per_chunk)
55                .ok_or_else(|| zero_source_chunking_field("max_row_groups_per_chunk"))?,
56            max_retry_attempts: NonZeroUsize::new(max_retry_attempts)
57                .ok_or_else(|| zero_source_chunking_field("max_retry_attempts"))?,
58            min_retry_chunk_row_groups: NonZeroUsize::new(min_retry_chunk_row_groups)
59                .ok_or_else(|| zero_source_chunking_field("min_retry_chunk_row_groups"))?,
60        })
61    }
62}
63
64impl Default for SourceChunking {
65    fn default() -> Self {
66        Self {
67            max_source_chunk_bytes: Self::DEFAULT_MAX_SOURCE_CHUNK_BYTES,
68            max_row_groups_per_chunk: Self::DEFAULT_MAX_ROW_GROUPS_PER_CHUNK,
69            max_retry_attempts: Self::DEFAULT_MAX_RETRY_ATTEMPTS,
70            min_retry_chunk_row_groups: Self::DEFAULT_MIN_RETRY_CHUNK_ROW_GROUPS,
71        }
72    }
73}
74
75fn zero_source_chunking_field(field: &'static str) -> Error {
76    Error::new(ErrorCode::InvalidRequest, "algeon_datafusion.native_exec.source_chunking", "must be greater than zero")
77        .with_fact("field", field)
78}
algeon_datafusion/native_exec/config/chunking.rs

algeon_datafusion/native_exec/config/
chunking.rs
