delta_arrow_reader/reader/
options.rs

1//! Parquet backend, snapshot, storage, and execution options.
2
3use std::collections::BTreeMap;
4
5use tokio::sync::Semaphore;
6
7use crate::{DeltaReaderError, error::InvalidConfigurationSnafu};
8
9const DEFAULT_MAX_CONCURRENT_FILE_READS_PER_PARTITION: usize = 3;
10const DEFAULT_OUTPUT_BUFFER_BATCHES_PER_PARTITION: usize = 1;
11const DEFAULT_PREFETCH_FILES_PER_PARTITION: usize = 2;
12const DEFAULT_PARQUET_METADATA_SIZE_HINT_BYTES: usize = 64 * 1024;
13pub(crate) const MAX_CONCURRENT_PARQUET_RANGE_READS: usize = 10;
14
15/// Storage options forwarded to Delta object-store construction.
16pub type DeltaStorageOptions = BTreeMap<String, String>;
17
18/// Delta snapshot selected for a table load.
19#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
20pub enum DeltaSnapshotSelection {
21    /// Load the latest available snapshot.
22    #[default]
23    Latest,
24    /// Load one exact Delta log version.
25    Version(u64),
26}
27
28/// Backend used to read Parquet data files.
29#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
30pub enum ParquetReaderBackend {
31    /// Delegate data-file reads to Delta Kernel's Parquet handler.
32    DeltaKernel,
33    /// Read data files directly through the asynchronous Parquet API.
34    #[default]
35    Direct,
36}
37
38/// Parquet range-read policy used by internal diagnostics and benchmarks.
39#[doc(hidden)]
40#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
41pub enum ParquetRangeReadPolicy {
42    /// Choose automatically for built-in remote stores and preserve other store implementations.
43    #[default]
44    Automatic,
45    /// Read only the normalized ranges requested by Parquet.
46    ExactRanges,
47    /// Merge ranges separated by at most one MiB before reading them.
48    MergeRangesWithinOneMegabyte,
49    /// Pass the requested ranges to the object store's own multi-range implementation.
50    StoreImplementation,
51}
52
53/// Bounded execution settings for one Delta scan.
54#[must_use = "execution options do nothing unless passed to a table or scan"]
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct DeltaScanExecutionOptions {
57    parquet_backend: ParquetReaderBackend,
58    max_concurrent_file_reads_per_scan: Option<usize>,
59    max_concurrent_file_reads_per_partition: usize,
60    output_buffer_batches_per_partition: usize,
61    prefetch_files_per_partition: usize,
62    parquet_metadata_size_hint_bytes: Option<usize>,
63    parquet_full_file_read_threshold_bytes: Option<usize>,
64    parquet_range_read_policy: ParquetRangeReadPolicy,
65}
66
67impl DeltaScanExecutionOptions {
68    /// Returns the baseline execution settings.
69    pub const fn new() -> Self {
70        Self {
71            parquet_backend: ParquetReaderBackend::Direct,
72            max_concurrent_file_reads_per_scan: None,
73            max_concurrent_file_reads_per_partition:
74                DEFAULT_MAX_CONCURRENT_FILE_READS_PER_PARTITION,
75            output_buffer_batches_per_partition: DEFAULT_OUTPUT_BUFFER_BATCHES_PER_PARTITION,
76            prefetch_files_per_partition: DEFAULT_PREFETCH_FILES_PER_PARTITION,
77            parquet_metadata_size_hint_bytes: Some(DEFAULT_PARQUET_METADATA_SIZE_HINT_BYTES),
78            parquet_full_file_read_threshold_bytes: None,
79            parquet_range_read_policy: ParquetRangeReadPolicy::Automatic,
80        }
81    }
82
83    /// Returns the selected Parquet reader backend.
84    pub const fn parquet_backend(&self) -> ParquetReaderBackend {
85        self.parquet_backend
86    }
87
88    /// Returns the optional scan-wide file-read limit.
89    pub const fn max_concurrent_file_reads_per_scan(&self) -> Option<usize> {
90        self.max_concurrent_file_reads_per_scan
91    }
92
93    /// Returns the per-partition file-read limit.
94    pub const fn max_concurrent_file_reads_per_partition(&self) -> usize {
95        self.max_concurrent_file_reads_per_partition
96    }
97
98    /// Returns the number of output batches buffered per partition.
99    pub const fn output_buffer_batches_per_partition(&self) -> usize {
100        self.output_buffer_batches_per_partition
101    }
102
103    /// Returns the number of future direct Parquet files prepared per partition.
104    pub const fn prefetch_files_per_partition(&self) -> usize {
105        self.prefetch_files_per_partition
106    }
107
108    /// Returns the direct Parquet reader's metadata size hint in bytes.
109    pub const fn parquet_metadata_size_hint_bytes(&self) -> Option<usize> {
110        self.parquet_metadata_size_hint_bytes
111    }
112
113    /// Returns the direct Parquet reader's full-file read threshold in bytes.
114    pub const fn parquet_full_file_read_threshold_bytes(&self) -> Option<usize> {
115        self.parquet_full_file_read_threshold_bytes
116    }
117
118    pub(crate) const fn parquet_range_read_policy(&self) -> ParquetRangeReadPolicy {
119        self.parquet_range_read_policy
120    }
121
122    /// Selects a Parquet reader backend.
123    pub const fn with_parquet_backend(mut self, parquet_backend: ParquetReaderBackend) -> Self {
124        self.parquet_backend = parquet_backend;
125        self
126    }
127
128    /// Selects a Parquet range-read policy for diagnostics and benchmarks.
129    #[doc(hidden)]
130    pub const fn with_parquet_range_read_policy(mut self, policy: ParquetRangeReadPolicy) -> Self {
131        self.parquet_range_read_policy = policy;
132        self
133    }
134
135    /// Sets or clears the scan-wide file-read limit.
136    ///
137    /// Explicit limits must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
138    /// `None` derives the limit from the partition target and per-partition limit,
139    /// capped at Tokio's maximum.
140    pub fn with_max_concurrent_file_reads_per_scan(
141        mut self,
142        max_concurrent_file_reads: Option<usize>,
143    ) -> Result<Self, DeltaReaderError> {
144        if let Some(capacity) = max_concurrent_file_reads {
145            validate_capacity(
146                capacity,
147                "max_concurrent_file_reads_per_scan_must_be_positive",
148                "max_concurrent_file_reads_per_scan_exceeds_tokio_max_permits",
149            )?;
150        }
151        self.max_concurrent_file_reads_per_scan = max_concurrent_file_reads;
152        Ok(self)
153    }
154
155    /// Sets the per-partition file-read limit.
156    ///
157    /// The limit must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
158    pub fn with_max_concurrent_file_reads_per_partition(
159        mut self,
160        max_concurrent_file_reads: usize,
161    ) -> Result<Self, DeltaReaderError> {
162        validate_capacity(
163            max_concurrent_file_reads,
164            "max_concurrent_file_reads_per_partition_must_be_positive",
165            "max_concurrent_file_reads_per_partition_exceeds_tokio_max_permits",
166        )?;
167        self.max_concurrent_file_reads_per_partition = max_concurrent_file_reads;
168        Ok(self)
169    }
170
171    /// Sets the number of output batches buffered per partition.
172    ///
173    /// The capacity must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
174    pub fn with_output_buffer_batches_per_partition(
175        mut self,
176        output_buffer_batches: usize,
177    ) -> Result<Self, DeltaReaderError> {
178        validate_capacity(
179            output_buffer_batches,
180            "output_buffer_batches_per_partition_must_be_positive",
181            "output_buffer_batches_per_partition_exceeds_tokio_max_permits",
182        )?;
183        self.output_buffer_batches_per_partition = output_buffer_batches;
184        Ok(self)
185    }
186
187    /// Sets the number of future direct Parquet files prepared per partition.
188    pub const fn with_prefetch_files_per_partition(mut self, prefetch_files: usize) -> Self {
189        self.prefetch_files_per_partition = prefetch_files;
190        self
191    }
192
193    /// Sets or clears the direct Parquet reader's metadata size hint in bytes.
194    pub fn with_parquet_metadata_size_hint_bytes(
195        mut self,
196        metadata_size_hint_bytes: Option<usize>,
197    ) -> Result<Self, DeltaReaderError> {
198        validate_optional_positive(
199            metadata_size_hint_bytes,
200            "parquet_metadata_size_hint_bytes_must_be_positive",
201        )?;
202        self.parquet_metadata_size_hint_bytes = metadata_size_hint_bytes;
203        Ok(self)
204    }
205
206    /// Sets or clears the direct Parquet reader's full-file read threshold in bytes.
207    pub fn with_parquet_full_file_read_threshold_bytes(
208        mut self,
209        full_file_read_threshold_bytes: Option<usize>,
210    ) -> Result<Self, DeltaReaderError> {
211        validate_optional_positive(
212            full_file_read_threshold_bytes,
213            "parquet_full_file_read_threshold_bytes_must_be_positive",
214        )?;
215        self.parquet_full_file_read_threshold_bytes = full_file_read_threshold_bytes;
216        Ok(self)
217    }
218
219    pub(crate) fn resolved_max_concurrent_file_reads_per_scan(
220        &self,
221        target_partitions: usize,
222    ) -> usize {
223        self.max_concurrent_file_reads_per_scan.unwrap_or_else(|| {
224            target_partitions
225                .saturating_mul(self.max_concurrent_file_reads_per_partition)
226                .clamp(1, Semaphore::MAX_PERMITS)
227        })
228    }
229}
230
231impl Default for DeltaScanExecutionOptions {
232    fn default() -> Self {
233        Self::new()
234    }
235}
236
237fn validate_capacity(
238    value: usize,
239    zero_reason: &'static str,
240    overflow_reason: &'static str,
241) -> Result<(), DeltaReaderError> {
242    if value == 0 {
243        return InvalidConfigurationSnafu {
244            reason: zero_reason,
245        }
246        .fail();
247    }
248    if value > Semaphore::MAX_PERMITS {
249        return InvalidConfigurationSnafu {
250            reason: overflow_reason,
251        }
252        .fail();
253    }
254    Ok(())
255}
256
257fn validate_optional_positive(
258    value: Option<usize>,
259    reason: &'static str,
260) -> Result<(), DeltaReaderError> {
261    if value == Some(0) {
262        return InvalidConfigurationSnafu { reason }.fail();
263    }
264    Ok(())
265}
266
267#[cfg(test)]
268mod tests {
269    use crate::DeltaReaderPhase;
270
271    use super::{
272        DeltaScanExecutionOptions, DeltaSnapshotSelection, DeltaStorageOptions,
273        ParquetReaderBackend,
274    };
275
276    #[test]
277    fn public_defaults_match_the_frozen_baseline() {
278        let options = DeltaScanExecutionOptions::new();
279
280        assert_eq!(
281            DeltaSnapshotSelection::default(),
282            DeltaSnapshotSelection::Latest
283        );
284        assert_eq!(
285            ParquetReaderBackend::default(),
286            ParquetReaderBackend::Direct
287        );
288        assert_eq!(DeltaScanExecutionOptions::default(), options);
289        assert_eq!(options.parquet_backend(), ParquetReaderBackend::Direct);
290        assert_eq!(options.max_concurrent_file_reads_per_scan(), None);
291        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
292        assert_eq!(options.output_buffer_batches_per_partition(), 1);
293        assert_eq!(options.prefetch_files_per_partition(), 2);
294        assert_eq!(options.parquet_metadata_size_hint_bytes(), Some(65_536));
295        assert_eq!(options.parquet_full_file_read_threshold_bytes(), None);
296        assert_eq!(DeltaStorageOptions::default(), DeltaStorageOptions::new());
297        assert_eq!(
298            DeltaSnapshotSelection::Version(7),
299            DeltaSnapshotSelection::Version(7)
300        );
301    }
302
303    #[test]
304    fn builders_set_every_public_option() -> Result<(), Box<dyn std::error::Error>> {
305        let options = DeltaScanExecutionOptions::new()
306            .with_parquet_backend(ParquetReaderBackend::DeltaKernel)
307            .with_max_concurrent_file_reads_per_scan(Some(8))?
308            .with_max_concurrent_file_reads_per_partition(4)?
309            .with_output_buffer_batches_per_partition(2)?
310            .with_prefetch_files_per_partition(0)
311            .with_parquet_metadata_size_hint_bytes(None)?
312            .with_parquet_full_file_read_threshold_bytes(Some(1024))?;
313
314        assert_eq!(options.parquet_backend(), ParquetReaderBackend::DeltaKernel);
315        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(8));
316        assert_eq!(options.max_concurrent_file_reads_per_partition(), 4);
317        assert_eq!(options.output_buffer_batches_per_partition(), 2);
318        assert_eq!(options.prefetch_files_per_partition(), 0);
319        assert_eq!(options.parquet_metadata_size_hint_bytes(), None);
320        assert_eq!(options.parquet_full_file_read_threshold_bytes(), Some(1024));
321        Ok(())
322    }
323
324    #[test]
325    fn invalid_bounds_return_redacted_configuration_errors() {
326        let invalid = [
327            DeltaScanExecutionOptions::new().with_max_concurrent_file_reads_per_scan(Some(0)),
328            DeltaScanExecutionOptions::new().with_max_concurrent_file_reads_per_partition(0),
329            DeltaScanExecutionOptions::new().with_output_buffer_batches_per_partition(0),
330            DeltaScanExecutionOptions::new().with_parquet_metadata_size_hint_bytes(Some(0)),
331            DeltaScanExecutionOptions::new().with_parquet_full_file_read_threshold_bytes(Some(0)),
332        ];
333
334        for result in invalid {
335            let error = result.expect_err("invalid execution options must fail");
336            assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
337            assert_eq!(error.code(), "invalid_configuration");
338        }
339    }
340
341    #[test]
342    fn independent_bounds_preserve_the_frozen_reader_behavior()
343    -> Result<(), Box<dyn std::error::Error>> {
344        let options = DeltaScanExecutionOptions::new()
345            .with_max_concurrent_file_reads_per_scan(Some(2))?
346            .with_prefetch_files_per_partition(4);
347
348        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(2));
349        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
350        assert_eq!(options.prefetch_files_per_partition(), 4);
351        Ok(())
352    }
353
354    #[test]
355    fn scan_capacity_resolves_once_from_the_fixed_partition_target()
356    -> Result<(), Box<dyn std::error::Error>> {
357        let max = tokio::sync::Semaphore::MAX_PERMITS;
358        for partition_capacity in [1, 3, max] {
359            let options = DeltaScanExecutionOptions::new()
360                .with_max_concurrent_file_reads_per_partition(partition_capacity)?;
361            for target in [0, 1, 4, max / 3, max / 3 + 1, max, max + 1, usize::MAX] {
362                // Widen the product to check both capacity overflow and usize overflow.
363                let product = (target as u128) * (partition_capacity as u128);
364                let expected = usize::try_from(product.clamp(1, max as u128))?;
365                assert_eq!(
366                    options.resolved_max_concurrent_file_reads_per_scan(target),
367                    expected
368                );
369                for explicit in [1, 7, max] {
370                    let options =
371                        options.with_max_concurrent_file_reads_per_scan(Some(explicit))?;
372                    assert_eq!(
373                        options.resolved_max_concurrent_file_reads_per_scan(target),
374                        explicit
375                    );
376                    assert_eq!(
377                        options
378                            .with_max_concurrent_file_reads_per_scan(None)?
379                            .resolved_max_concurrent_file_reads_per_scan(target),
380                        expected,
381                    );
382                }
383            }
384        }
385        Ok(())
386    }
387}
delta_arrow_reader/reader/options.rs

delta_arrow_reader/reader/
options.rs
