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// A process-wide ceiling, not a concurrency target for each file or scan.
15pub(crate) const MAX_SHARED_RANGE_READS: usize = 512;
16// Keep the estimator and partial-read execution within the same byte bound.
17pub(crate) const MAX_INTRA_PAGE_READ_BYTES: u128 = 128 * 1024 * 1024;
18
19/// Storage options forwarded to Delta object-store construction.
20pub type DeltaStorageOptions = BTreeMap<String, String>;
21
22/// Delta snapshot selected for a table load.
23#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
24pub enum DeltaSnapshotSelection {
25    /// Load the latest available snapshot.
26    #[default]
27    Latest,
28    /// Load one exact Delta log version.
29    Version(u64),
30}
31
32/// Backend used to read Parquet data files.
33#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
34pub enum ParquetReaderBackend {
35    /// Delegate data-file reads to Delta Kernel's Parquet handler.
36    DeltaKernel,
37    /// Read data files directly through the asynchronous Parquet API.
38    #[default]
39    Direct,
40}
41
42/// Parquet range-read policy used by internal diagnostics and benchmarks.
43#[doc(hidden)]
44#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
45pub enum ParquetRangeReadPolicy {
46    /// Choose automatically for built-in remote stores and preserve other store implementations.
47    #[default]
48    Automatic,
49    /// Read only the normalized ranges requested by Parquet.
50    ExactRanges,
51    /// Merge ranges separated by at most one MiB before reading them.
52    MergeRangesWithinOneMegabyte,
53    /// Pass the requested ranges to the object store's own multi-range implementation.
54    StoreImplementation,
55}
56
57/// Bounded execution settings for one Delta scan.
58#[must_use = "execution options do nothing unless passed to a table or scan"]
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub struct DeltaScanExecutionOptions {
61    parquet_backend: ParquetReaderBackend,
62    max_concurrent_file_reads_per_scan: Option<usize>,
63    max_concurrent_file_reads_per_partition: usize,
64    output_buffer_batches_per_partition: usize,
65    prefetch_files_per_partition: usize,
66    parquet_metadata_size_hint_bytes: Option<usize>,
67    parquet_full_file_read_threshold_bytes: Option<usize>,
68    parquet_range_read_policy: ParquetRangeReadPolicy,
69    experimental_intra_page_reads: bool,
70}
71
72impl DeltaScanExecutionOptions {
73    /// Returns the baseline execution settings.
74    pub const fn new() -> Self {
75        Self {
76            parquet_backend: ParquetReaderBackend::Direct,
77            max_concurrent_file_reads_per_scan: None,
78            max_concurrent_file_reads_per_partition:
79                DEFAULT_MAX_CONCURRENT_FILE_READS_PER_PARTITION,
80            output_buffer_batches_per_partition: DEFAULT_OUTPUT_BUFFER_BATCHES_PER_PARTITION,
81            prefetch_files_per_partition: DEFAULT_PREFETCH_FILES_PER_PARTITION,
82            parquet_metadata_size_hint_bytes: Some(DEFAULT_PARQUET_METADATA_SIZE_HINT_BYTES),
83            parquet_full_file_read_threshold_bytes: None,
84            parquet_range_read_policy: ParquetRangeReadPolicy::Automatic,
85            experimental_intra_page_reads: true,
86        }
87    }
88
89    /// Returns the selected Parquet reader backend.
90    pub const fn parquet_backend(&self) -> ParquetReaderBackend {
91        self.parquet_backend
92    }
93
94    /// Returns the optional scan-wide file-read limit.
95    pub const fn max_concurrent_file_reads_per_scan(&self) -> Option<usize> {
96        self.max_concurrent_file_reads_per_scan
97    }
98
99    /// Returns the per-partition file-read limit.
100    pub const fn max_concurrent_file_reads_per_partition(&self) -> usize {
101        self.max_concurrent_file_reads_per_partition
102    }
103
104    /// Returns the number of output batches buffered per partition.
105    pub const fn output_buffer_batches_per_partition(&self) -> usize {
106        self.output_buffer_batches_per_partition
107    }
108
109    /// Returns the number of future direct Parquet files prepared per partition.
110    pub const fn prefetch_files_per_partition(&self) -> usize {
111        self.prefetch_files_per_partition
112    }
113
114    /// Returns the direct Parquet reader's metadata size hint in bytes.
115    pub const fn parquet_metadata_size_hint_bytes(&self) -> Option<usize> {
116        self.parquet_metadata_size_hint_bytes
117    }
118
119    /// Returns the direct Parquet reader's full-file read threshold in bytes.
120    pub const fn parquet_full_file_read_threshold_bytes(&self) -> Option<usize> {
121        self.parquet_full_file_read_threshold_bytes
122    }
123
124    pub(crate) const fn parquet_range_read_policy(&self) -> ParquetRangeReadPolicy {
125        self.parquet_range_read_policy
126    }
127
128    /// Whether experimental partial-page reads are enabled for the direct backend.
129    pub const fn experimental_intra_page_reads(&self) -> bool {
130        self.experimental_intra_page_reads
131    }
132
133    /// Enables experimental partial reads of supported flat Parquet pages.
134    ///
135    /// Enabled by default. The direct reader uses predicate-derived row numbers to
136    /// read selected values from PLAIN numeric pages that are uncompressed or
137    /// held in Zstd raw blocks without checksums. Supports nullable and required
138    /// INT32, INT64, FLOAT, DOUBLE and FIXED_LEN_BYTE_ARRAY physical columns,
139    /// including existing decimal conversions. Fixed-length widths come from the
140    /// Parquet schema. This adds no new Delta type mappings, such as fixed-size
141    /// binary to Delta binary.
142    /// UTF-8 and binary columns also support uncompressed DELTA_LENGTH_BYTE_ARRAY
143    /// pages. All supported pages require V1 layout, offset indexes and no page checksum.
144    /// Variable-length discovery must fit a modeled fallback budget and shared memory
145    /// limits; fallback reuses downloaded prefixes.
146    /// Unsupported layouts and bounded-work limits use ordinary reads.
147    /// Explicit range-read policies other than `Automatic` disable this optimization.
148    /// The Delta Kernel backend ignores this direct-reader option.
149    ///
150    /// Recent query traffic supplies latency and throughput estimates. Partial
151    /// reads must beat the ordinary plan by more than ten percent, including
152    /// probes and dependent request rounds. Missing evidence, dense selections,
153    /// an unmeasured small-request cost, or insufficient shared request capacity
154    /// retain ordinary reads. Automatic table warmup supplies the full profile for
155    /// supported S3 stores. If that warmup did not finish, an eligible filtered scan
156    /// retries it once while building its plan.
157    /// Passing `false` restores ordinary reads and skips automatic table warmup when
158    /// supplied to the table builder. It also skips pending recovery for this scan.
159    /// Explicit warmup settings remain in effect during table loading.
160    pub const fn with_experimental_intra_page_reads(mut self, enabled: bool) -> Self {
161        self.experimental_intra_page_reads = enabled;
162        self
163    }
164
165    /// Selects a Parquet reader backend.
166    pub const fn with_parquet_backend(mut self, parquet_backend: ParquetReaderBackend) -> Self {
167        self.parquet_backend = parquet_backend;
168        self
169    }
170
171    /// Selects a Parquet range-read policy for diagnostics and benchmarks.
172    #[doc(hidden)]
173    pub const fn with_parquet_range_read_policy(mut self, policy: ParquetRangeReadPolicy) -> Self {
174        self.parquet_range_read_policy = policy;
175        self
176    }
177
178    /// Sets or clears the scan-wide file-read limit.
179    ///
180    /// Explicit limits must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
181    /// `None` derives the limit from the partition target and per-partition limit,
182    /// capped at Tokio's maximum.
183    pub fn with_max_concurrent_file_reads_per_scan(
184        mut self,
185        max_concurrent_file_reads: Option<usize>,
186    ) -> Result<Self, DeltaReaderError> {
187        if let Some(capacity) = max_concurrent_file_reads {
188            validate_capacity(
189                capacity,
190                "max_concurrent_file_reads_per_scan_must_be_positive",
191                "max_concurrent_file_reads_per_scan_exceeds_tokio_max_permits",
192            )?;
193        }
194        self.max_concurrent_file_reads_per_scan = max_concurrent_file_reads;
195        Ok(self)
196    }
197
198    /// Sets the per-partition file-read limit.
199    ///
200    /// The limit must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
201    pub fn with_max_concurrent_file_reads_per_partition(
202        mut self,
203        max_concurrent_file_reads: usize,
204    ) -> Result<Self, DeltaReaderError> {
205        validate_capacity(
206            max_concurrent_file_reads,
207            "max_concurrent_file_reads_per_partition_must_be_positive",
208            "max_concurrent_file_reads_per_partition_exceeds_tokio_max_permits",
209        )?;
210        self.max_concurrent_file_reads_per_partition = max_concurrent_file_reads;
211        Ok(self)
212    }
213
214    /// Sets the number of output batches buffered per partition.
215    ///
216    /// The capacity must be in `1..=tokio::sync::Semaphore::MAX_PERMITS`.
217    pub fn with_output_buffer_batches_per_partition(
218        mut self,
219        output_buffer_batches: usize,
220    ) -> Result<Self, DeltaReaderError> {
221        validate_capacity(
222            output_buffer_batches,
223            "output_buffer_batches_per_partition_must_be_positive",
224            "output_buffer_batches_per_partition_exceeds_tokio_max_permits",
225        )?;
226        self.output_buffer_batches_per_partition = output_buffer_batches;
227        Ok(self)
228    }
229
230    /// Sets the number of future direct Parquet files prepared per partition.
231    pub const fn with_prefetch_files_per_partition(mut self, prefetch_files: usize) -> Self {
232        self.prefetch_files_per_partition = prefetch_files;
233        self
234    }
235
236    /// Sets or clears the direct Parquet reader's metadata size hint in bytes.
237    pub fn with_parquet_metadata_size_hint_bytes(
238        mut self,
239        metadata_size_hint_bytes: Option<usize>,
240    ) -> Result<Self, DeltaReaderError> {
241        validate_optional_positive(
242            metadata_size_hint_bytes,
243            "parquet_metadata_size_hint_bytes_must_be_positive",
244        )?;
245        self.parquet_metadata_size_hint_bytes = metadata_size_hint_bytes;
246        Ok(self)
247    }
248
249    /// Sets or clears the direct Parquet reader's full-file read threshold in bytes.
250    pub fn with_parquet_full_file_read_threshold_bytes(
251        mut self,
252        full_file_read_threshold_bytes: Option<usize>,
253    ) -> Result<Self, DeltaReaderError> {
254        validate_optional_positive(
255            full_file_read_threshold_bytes,
256            "parquet_full_file_read_threshold_bytes_must_be_positive",
257        )?;
258        self.parquet_full_file_read_threshold_bytes = full_file_read_threshold_bytes;
259        Ok(self)
260    }
261
262    pub(crate) fn resolved_max_concurrent_file_reads_per_scan(
263        &self,
264        target_partitions: usize,
265    ) -> usize {
266        self.max_concurrent_file_reads_per_scan.unwrap_or_else(|| {
267            target_partitions
268                .saturating_mul(self.max_concurrent_file_reads_per_partition)
269                .clamp(1, Semaphore::MAX_PERMITS)
270        })
271    }
272}
273
274impl Default for DeltaScanExecutionOptions {
275    fn default() -> Self {
276        Self::new()
277    }
278}
279
280fn validate_capacity(
281    value: usize,
282    zero_reason: &'static str,
283    overflow_reason: &'static str,
284) -> Result<(), DeltaReaderError> {
285    if value == 0 {
286        return InvalidConfigurationSnafu {
287            reason: zero_reason,
288        }
289        .fail();
290    }
291    if value > Semaphore::MAX_PERMITS {
292        return InvalidConfigurationSnafu {
293            reason: overflow_reason,
294        }
295        .fail();
296    }
297    Ok(())
298}
299
300fn validate_optional_positive(
301    value: Option<usize>,
302    reason: &'static str,
303) -> Result<(), DeltaReaderError> {
304    if value == Some(0) {
305        return InvalidConfigurationSnafu { reason }.fail();
306    }
307    Ok(())
308}
309
310#[cfg(test)]
311mod tests {
312    use crate::DeltaReaderPhase;
313
314    use super::{
315        DeltaScanExecutionOptions, DeltaSnapshotSelection, DeltaStorageOptions,
316        ParquetReaderBackend,
317    };
318
319    #[test]
320    fn public_defaults_enable_conditional_partial_reads() {
321        let options = DeltaScanExecutionOptions::new();
322
323        assert_eq!(
324            DeltaSnapshotSelection::default(),
325            DeltaSnapshotSelection::Latest
326        );
327        assert_eq!(
328            ParquetReaderBackend::default(),
329            ParquetReaderBackend::Direct
330        );
331        assert_eq!(DeltaScanExecutionOptions::default(), options);
332        assert_eq!(options.parquet_backend(), ParquetReaderBackend::Direct);
333        assert!(options.experimental_intra_page_reads());
334        assert_eq!(options.max_concurrent_file_reads_per_scan(), None);
335        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
336        assert_eq!(options.output_buffer_batches_per_partition(), 1);
337        assert_eq!(options.prefetch_files_per_partition(), 2);
338        assert_eq!(options.parquet_metadata_size_hint_bytes(), Some(65_536));
339        assert_eq!(options.parquet_full_file_read_threshold_bytes(), None);
340        assert_eq!(DeltaStorageOptions::default(), DeltaStorageOptions::new());
341        assert_eq!(
342            DeltaSnapshotSelection::Version(7),
343            DeltaSnapshotSelection::Version(7)
344        );
345    }
346
347    #[test]
348    fn builders_set_every_public_option() -> Result<(), Box<dyn std::error::Error>> {
349        let options = DeltaScanExecutionOptions::new()
350            .with_parquet_backend(ParquetReaderBackend::DeltaKernel)
351            .with_experimental_intra_page_reads(false)
352            .with_max_concurrent_file_reads_per_scan(Some(8))?
353            .with_max_concurrent_file_reads_per_partition(4)?
354            .with_output_buffer_batches_per_partition(2)?
355            .with_prefetch_files_per_partition(0)
356            .with_parquet_metadata_size_hint_bytes(None)?
357            .with_parquet_full_file_read_threshold_bytes(Some(1024))?;
358
359        assert_eq!(options.parquet_backend(), ParquetReaderBackend::DeltaKernel);
360        assert!(!options.experimental_intra_page_reads());
361        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(8));
362        assert_eq!(options.max_concurrent_file_reads_per_partition(), 4);
363        assert_eq!(options.output_buffer_batches_per_partition(), 2);
364        assert_eq!(options.prefetch_files_per_partition(), 0);
365        assert_eq!(options.parquet_metadata_size_hint_bytes(), None);
366        assert_eq!(options.parquet_full_file_read_threshold_bytes(), Some(1024));
367        Ok(())
368    }
369
370    #[test]
371    fn invalid_bounds_return_redacted_configuration_errors() {
372        let invalid = [
373            DeltaScanExecutionOptions::new().with_max_concurrent_file_reads_per_scan(Some(0)),
374            DeltaScanExecutionOptions::new().with_max_concurrent_file_reads_per_partition(0),
375            DeltaScanExecutionOptions::new().with_output_buffer_batches_per_partition(0),
376            DeltaScanExecutionOptions::new().with_parquet_metadata_size_hint_bytes(Some(0)),
377            DeltaScanExecutionOptions::new().with_parquet_full_file_read_threshold_bytes(Some(0)),
378        ];
379
380        for result in invalid {
381            let error = result.expect_err("invalid execution options must fail");
382            assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
383            assert_eq!(error.code(), "invalid_configuration");
384        }
385    }
386
387    #[test]
388    fn independent_bounds_preserve_the_frozen_reader_behavior()
389    -> Result<(), Box<dyn std::error::Error>> {
390        let options = DeltaScanExecutionOptions::new()
391            .with_max_concurrent_file_reads_per_scan(Some(2))?
392            .with_prefetch_files_per_partition(4);
393
394        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(2));
395        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
396        assert_eq!(options.prefetch_files_per_partition(), 4);
397        Ok(())
398    }
399
400    #[test]
401    fn scan_capacity_resolves_once_from_the_fixed_partition_target()
402    -> Result<(), Box<dyn std::error::Error>> {
403        let max = tokio::sync::Semaphore::MAX_PERMITS;
404        for partition_capacity in [1, 3, max] {
405            let options = DeltaScanExecutionOptions::new()
406                .with_max_concurrent_file_reads_per_partition(partition_capacity)?;
407            for target in [0, 1, 4, max / 3, max / 3 + 1, max, max + 1, usize::MAX] {
408                // Widen the product to check both capacity overflow and usize overflow.
409                let product = (target as u128) * (partition_capacity as u128);
410                let expected = usize::try_from(product.clamp(1, max as u128))?;
411                assert_eq!(
412                    options.resolved_max_concurrent_file_reads_per_scan(target),
413                    expected
414                );
415                for explicit in [1, 7, max] {
416                    let options =
417                        options.with_max_concurrent_file_reads_per_scan(Some(explicit))?;
418                    assert_eq!(
419                        options.resolved_max_concurrent_file_reads_per_scan(target),
420                        explicit
421                    );
422                    assert_eq!(
423                        options
424                            .with_max_concurrent_file_reads_per_scan(None)?
425                            .resolved_max_concurrent_file_reads_per_scan(target),
426                        expected,
427                    );
428                }
429            }
430        }
431        Ok(())
432    }
433}
delta_arrow_reader/reader/options.rs

delta_arrow_reader/reader/
options.rs
