delta_arrow_reader/
config.rs

1use std::collections::BTreeMap;
2
3use crate::{DeltaReaderError, error::InvalidConfigurationSnafu};
4
5const DEFAULT_MAX_CONCURRENT_FILE_READS_PER_PARTITION: usize = 3;
6const DEFAULT_OUTPUT_BUFFER_CAPACITY_PER_PARTITION: usize = 1;
7const DEFAULT_NATIVE_ASYNC_PREFETCH_FILE_COUNT_PER_PARTITION: usize = 2;
8const DEFAULT_PARQUET_METADATA_SIZE_HINT: usize = 64 * 1024;
9
10/// Storage options forwarded to Delta object-store construction.
11pub type DeltaStorageOptions = BTreeMap<String, String>;
12
13/// Delta snapshot selected for a table load.
14#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
15pub enum DeltaSnapshotSelection {
16    /// Load the latest available snapshot.
17    #[default]
18    Latest,
19    /// Load one exact Delta log version.
20    Version(u64),
21}
22
23/// Backend used to read Delta data files.
24#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
25pub enum DeltaReaderBackend {
26    /// Use the official Delta Kernel data-file reader.
27    OfficialKernel,
28    /// Use the native asynchronous Parquet reader.
29    #[default]
30    NativeAsync,
31}
32
33/// Bounded execution settings for one Delta scan.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct DeltaReaderExecutionOptions {
36    reader_backend: DeltaReaderBackend,
37    max_concurrent_file_reads_per_scan: Option<usize>,
38    max_concurrent_file_reads_per_partition: usize,
39    output_buffer_capacity_per_partition: usize,
40    native_async_prefetch_file_count_per_partition: usize,
41    parquet_metadata_size_hint: Option<usize>,
42    parquet_full_file_read_threshold: Option<usize>,
43}
44
45impl DeltaReaderExecutionOptions {
46    /// Returns the baseline execution settings.
47    pub const fn new() -> Self {
48        Self {
49            reader_backend: DeltaReaderBackend::NativeAsync,
50            max_concurrent_file_reads_per_scan: None,
51            max_concurrent_file_reads_per_partition:
52                DEFAULT_MAX_CONCURRENT_FILE_READS_PER_PARTITION,
53            output_buffer_capacity_per_partition: DEFAULT_OUTPUT_BUFFER_CAPACITY_PER_PARTITION,
54            native_async_prefetch_file_count_per_partition:
55                DEFAULT_NATIVE_ASYNC_PREFETCH_FILE_COUNT_PER_PARTITION,
56            parquet_metadata_size_hint: Some(DEFAULT_PARQUET_METADATA_SIZE_HINT),
57            parquet_full_file_read_threshold: None,
58        }
59    }
60
61    /// Returns the selected data-file reader backend.
62    pub const fn reader_backend(&self) -> DeltaReaderBackend {
63        self.reader_backend
64    }
65
66    /// Returns the optional scan-wide file-read limit.
67    pub const fn max_concurrent_file_reads_per_scan(&self) -> Option<usize> {
68        self.max_concurrent_file_reads_per_scan
69    }
70
71    /// Returns the per-partition file-read limit.
72    pub const fn max_concurrent_file_reads_per_partition(&self) -> usize {
73        self.max_concurrent_file_reads_per_partition
74    }
75
76    /// Returns the per-partition output buffer capacity.
77    pub const fn output_buffer_capacity_per_partition(&self) -> usize {
78        self.output_buffer_capacity_per_partition
79    }
80
81    /// Returns the NativeAsync per-partition file prefetch depth.
82    pub const fn native_async_prefetch_file_count_per_partition(&self) -> usize {
83        self.native_async_prefetch_file_count_per_partition
84    }
85
86    /// Returns the NativeAsync Parquet metadata size hint.
87    pub const fn parquet_metadata_size_hint(&self) -> Option<usize> {
88        self.parquet_metadata_size_hint
89    }
90
91    /// Returns the NativeAsync full-file read threshold.
92    pub const fn parquet_full_file_read_threshold(&self) -> Option<usize> {
93        self.parquet_full_file_read_threshold
94    }
95
96    /// Selects a data-file reader backend.
97    pub fn with_reader_backend(
98        mut self,
99        value: DeltaReaderBackend,
100    ) -> Result<Self, DeltaReaderError> {
101        self.reader_backend = value;
102        self.validate()?;
103        Ok(self)
104    }
105
106    /// Sets or clears the scan-wide file-read limit.
107    pub fn with_max_concurrent_file_reads_per_scan(
108        mut self,
109        value: Option<usize>,
110    ) -> Result<Self, DeltaReaderError> {
111        self.max_concurrent_file_reads_per_scan = value;
112        self.validate()?;
113        Ok(self)
114    }
115
116    /// Sets the per-partition file-read limit.
117    pub fn with_max_concurrent_file_reads_per_partition(
118        mut self,
119        value: usize,
120    ) -> Result<Self, DeltaReaderError> {
121        self.max_concurrent_file_reads_per_partition = value;
122        self.validate()?;
123        Ok(self)
124    }
125
126    /// Sets the per-partition output buffer capacity.
127    pub fn with_output_buffer_capacity_per_partition(
128        mut self,
129        value: usize,
130    ) -> Result<Self, DeltaReaderError> {
131        self.output_buffer_capacity_per_partition = value;
132        self.validate()?;
133        Ok(self)
134    }
135
136    /// Sets the NativeAsync per-partition file prefetch depth.
137    pub fn with_native_async_prefetch_file_count_per_partition(
138        mut self,
139        value: usize,
140    ) -> Result<Self, DeltaReaderError> {
141        self.native_async_prefetch_file_count_per_partition = value;
142        self.validate()?;
143        Ok(self)
144    }
145
146    /// Sets or clears the NativeAsync Parquet metadata size hint.
147    pub fn with_parquet_metadata_size_hint(
148        mut self,
149        value: Option<usize>,
150    ) -> Result<Self, DeltaReaderError> {
151        self.parquet_metadata_size_hint = value;
152        self.validate()?;
153        Ok(self)
154    }
155
156    /// Sets or clears the NativeAsync full-file read threshold.
157    pub fn with_parquet_full_file_read_threshold(
158        mut self,
159        value: Option<usize>,
160    ) -> Result<Self, DeltaReaderError> {
161        self.parquet_full_file_read_threshold = value;
162        self.validate()?;
163        Ok(self)
164    }
165
166    /// Validates all execution bounds.
167    pub fn validate(&self) -> Result<(), DeltaReaderError> {
168        validate_optional_positive(
169            self.max_concurrent_file_reads_per_scan,
170            "max_concurrent_file_reads_per_scan_must_be_positive",
171        )?;
172        validate_positive(
173            self.max_concurrent_file_reads_per_partition,
174            "max_concurrent_file_reads_per_partition_must_be_positive",
175        )?;
176        validate_positive(
177            self.output_buffer_capacity_per_partition,
178            "output_buffer_capacity_per_partition_must_be_positive",
179        )?;
180        validate_optional_positive(
181            self.parquet_metadata_size_hint,
182            "parquet_metadata_size_hint_must_be_positive",
183        )?;
184        validate_optional_positive(
185            self.parquet_full_file_read_threshold,
186            "parquet_full_file_read_threshold_must_be_positive",
187        )?;
188
189        if self
190            .max_concurrent_file_reads_per_scan
191            .is_some_and(|scan_limit| self.max_concurrent_file_reads_per_partition > scan_limit)
192        {
193            return InvalidConfigurationSnafu {
194                reason: "partition_file_read_limit_exceeds_scan_limit",
195            }
196            .fail();
197        }
198
199        if self.native_async_prefetch_file_count_per_partition
200            > self.max_concurrent_file_reads_per_partition
201        {
202            return InvalidConfigurationSnafu {
203                reason: "native_async_prefetch_exceeds_partition_file_read_limit",
204            }
205            .fail();
206        }
207
208        Ok(())
209    }
210
211    pub(crate) fn resolved_max_concurrent_file_reads_per_scan(
212        &self,
213        target_partitions: usize,
214    ) -> usize {
215        self.max_concurrent_file_reads_per_scan.unwrap_or_else(|| {
216            target_partitions
217                .saturating_mul(self.max_concurrent_file_reads_per_partition)
218                .max(1)
219        })
220    }
221}
222
223impl Default for DeltaReaderExecutionOptions {
224    fn default() -> Self {
225        Self::new()
226    }
227}
228
229fn validate_positive(value: usize, reason: &'static str) -> Result<(), DeltaReaderError> {
230    if value == 0 {
231        return InvalidConfigurationSnafu { reason }.fail();
232    }
233    Ok(())
234}
235
236fn validate_optional_positive(
237    value: Option<usize>,
238    reason: &'static str,
239) -> Result<(), DeltaReaderError> {
240    if value == Some(0) {
241        return InvalidConfigurationSnafu { reason }.fail();
242    }
243    Ok(())
244}
245
246#[cfg(test)]
247mod tests {
248    use crate::DeltaReaderPhase;
249
250    use super::{
251        DeltaReaderBackend, DeltaReaderExecutionOptions, DeltaSnapshotSelection,
252        DeltaStorageOptions,
253    };
254
255    #[test]
256    fn public_defaults_match_the_frozen_baseline() {
257        let options = DeltaReaderExecutionOptions::new();
258
259        assert_eq!(
260            DeltaSnapshotSelection::default(),
261            DeltaSnapshotSelection::Latest
262        );
263        assert_eq!(
264            DeltaReaderBackend::default(),
265            DeltaReaderBackend::NativeAsync
266        );
267        assert_eq!(DeltaReaderExecutionOptions::default(), options);
268        assert_eq!(options.reader_backend(), DeltaReaderBackend::NativeAsync);
269        assert_eq!(options.max_concurrent_file_reads_per_scan(), None);
270        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
271        assert_eq!(options.output_buffer_capacity_per_partition(), 1);
272        assert_eq!(options.native_async_prefetch_file_count_per_partition(), 2);
273        assert_eq!(options.parquet_metadata_size_hint(), Some(65_536));
274        assert_eq!(options.parquet_full_file_read_threshold(), None);
275        assert_eq!(DeltaStorageOptions::default(), DeltaStorageOptions::new());
276        assert_eq!(
277            DeltaSnapshotSelection::Version(7),
278            DeltaSnapshotSelection::Version(7)
279        );
280    }
281
282    #[test]
283    fn builders_set_every_public_option() -> Result<(), Box<dyn std::error::Error>> {
284        let options = DeltaReaderExecutionOptions::new()
285            .with_reader_backend(DeltaReaderBackend::OfficialKernel)?
286            .with_max_concurrent_file_reads_per_scan(Some(8))?
287            .with_max_concurrent_file_reads_per_partition(4)?
288            .with_output_buffer_capacity_per_partition(2)?
289            .with_native_async_prefetch_file_count_per_partition(0)?
290            .with_parquet_metadata_size_hint(None)?
291            .with_parquet_full_file_read_threshold(Some(1024))?;
292
293        assert_eq!(options.reader_backend(), DeltaReaderBackend::OfficialKernel);
294        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(8));
295        assert_eq!(options.max_concurrent_file_reads_per_partition(), 4);
296        assert_eq!(options.output_buffer_capacity_per_partition(), 2);
297        assert_eq!(options.native_async_prefetch_file_count_per_partition(), 0);
298        assert_eq!(options.parquet_metadata_size_hint(), None);
299        assert_eq!(options.parquet_full_file_read_threshold(), Some(1024));
300        Ok(())
301    }
302
303    #[test]
304    fn invalid_bounds_return_redacted_configuration_errors() {
305        let invalid = [
306            DeltaReaderExecutionOptions::new().with_max_concurrent_file_reads_per_scan(Some(0)),
307            DeltaReaderExecutionOptions::new().with_max_concurrent_file_reads_per_partition(0),
308            DeltaReaderExecutionOptions::new().with_output_buffer_capacity_per_partition(0),
309            DeltaReaderExecutionOptions::new().with_parquet_metadata_size_hint(Some(0)),
310            DeltaReaderExecutionOptions::new().with_parquet_full_file_read_threshold(Some(0)),
311            DeltaReaderExecutionOptions::new().with_max_concurrent_file_reads_per_scan(Some(2)),
312            DeltaReaderExecutionOptions::new()
313                .with_native_async_prefetch_file_count_per_partition(4),
314        ];
315
316        for result in invalid {
317            let error = result.expect_err("invalid execution options must fail");
318            assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
319            assert_eq!(error.as_str(), "invalid_configuration");
320        }
321    }
322
323    #[test]
324    fn scan_capacity_resolves_once_from_the_fixed_partition_target()
325    -> Result<(), Box<dyn std::error::Error>> {
326        let defaults = DeltaReaderExecutionOptions::new();
327        assert_eq!(defaults.resolved_max_concurrent_file_reads_per_scan(4), 12);
328        assert_eq!(
329            defaults.resolved_max_concurrent_file_reads_per_scan(usize::MAX),
330            usize::MAX
331        );
332
333        let explicit = defaults.with_max_concurrent_file_reads_per_scan(Some(7))?;
334        assert_eq!(explicit.resolved_max_concurrent_file_reads_per_scan(4), 7);
335        Ok(())
336    }
337}
delta_arrow_reader/config.rs

delta_arrow_reader/
config.rs
