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

delta_arrow_reader/reader/
options.rs
