delta_arrow_reader/reader/
options.rs

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

delta_arrow_reader/reader/
options.rs
