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        Ok(())
190    }
191
192    pub(crate) fn resolved_max_concurrent_file_reads_per_scan(
193        &self,
194        target_partitions: usize,
195    ) -> usize {
196        self.max_concurrent_file_reads_per_scan.unwrap_or_else(|| {
197            target_partitions
198                .saturating_mul(self.max_concurrent_file_reads_per_partition)
199                .max(1)
200        })
201    }
202}
203
204impl Default for DeltaReaderExecutionOptions {
205    fn default() -> Self {
206        Self::new()
207    }
208}
209
210fn validate_positive(value: usize, reason: &'static str) -> Result<(), DeltaReaderError> {
211    if value == 0 {
212        return InvalidConfigurationSnafu { reason }.fail();
213    }
214    Ok(())
215}
216
217fn validate_optional_positive(
218    value: Option<usize>,
219    reason: &'static str,
220) -> Result<(), DeltaReaderError> {
221    if value == Some(0) {
222        return InvalidConfigurationSnafu { reason }.fail();
223    }
224    Ok(())
225}
226
227#[cfg(test)]
228mod tests {
229    use crate::DeltaReaderPhase;
230
231    use super::{
232        DeltaReaderBackend, DeltaReaderExecutionOptions, DeltaSnapshotSelection,
233        DeltaStorageOptions,
234    };
235
236    #[test]
237    fn public_defaults_match_the_frozen_baseline() {
238        let options = DeltaReaderExecutionOptions::new();
239
240        assert_eq!(
241            DeltaSnapshotSelection::default(),
242            DeltaSnapshotSelection::Latest
243        );
244        assert_eq!(
245            DeltaReaderBackend::default(),
246            DeltaReaderBackend::NativeAsync
247        );
248        assert_eq!(DeltaReaderExecutionOptions::default(), options);
249        assert_eq!(options.reader_backend(), DeltaReaderBackend::NativeAsync);
250        assert_eq!(options.max_concurrent_file_reads_per_scan(), None);
251        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
252        assert_eq!(options.output_buffer_capacity_per_partition(), 1);
253        assert_eq!(options.native_async_prefetch_file_count_per_partition(), 2);
254        assert_eq!(options.parquet_metadata_size_hint(), Some(65_536));
255        assert_eq!(options.parquet_full_file_read_threshold(), None);
256        assert_eq!(DeltaStorageOptions::default(), DeltaStorageOptions::new());
257        assert_eq!(
258            DeltaSnapshotSelection::Version(7),
259            DeltaSnapshotSelection::Version(7)
260        );
261    }
262
263    #[test]
264    fn builders_set_every_public_option() -> Result<(), Box<dyn std::error::Error>> {
265        let options = DeltaReaderExecutionOptions::new()
266            .with_reader_backend(DeltaReaderBackend::OfficialKernel)?
267            .with_max_concurrent_file_reads_per_scan(Some(8))?
268            .with_max_concurrent_file_reads_per_partition(4)?
269            .with_output_buffer_capacity_per_partition(2)?
270            .with_native_async_prefetch_file_count_per_partition(0)?
271            .with_parquet_metadata_size_hint(None)?
272            .with_parquet_full_file_read_threshold(Some(1024))?;
273
274        assert_eq!(options.reader_backend(), DeltaReaderBackend::OfficialKernel);
275        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(8));
276        assert_eq!(options.max_concurrent_file_reads_per_partition(), 4);
277        assert_eq!(options.output_buffer_capacity_per_partition(), 2);
278        assert_eq!(options.native_async_prefetch_file_count_per_partition(), 0);
279        assert_eq!(options.parquet_metadata_size_hint(), None);
280        assert_eq!(options.parquet_full_file_read_threshold(), Some(1024));
281        Ok(())
282    }
283
284    #[test]
285    fn invalid_bounds_return_redacted_configuration_errors() {
286        let invalid = [
287            DeltaReaderExecutionOptions::new().with_max_concurrent_file_reads_per_scan(Some(0)),
288            DeltaReaderExecutionOptions::new().with_max_concurrent_file_reads_per_partition(0),
289            DeltaReaderExecutionOptions::new().with_output_buffer_capacity_per_partition(0),
290            DeltaReaderExecutionOptions::new().with_parquet_metadata_size_hint(Some(0)),
291            DeltaReaderExecutionOptions::new().with_parquet_full_file_read_threshold(Some(0)),
292        ];
293
294        for result in invalid {
295            let error = result.expect_err("invalid execution options must fail");
296            assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
297            assert_eq!(error.as_str(), "invalid_configuration");
298        }
299    }
300
301    #[test]
302    fn independent_bounds_preserve_the_frozen_reader_behavior()
303    -> Result<(), Box<dyn std::error::Error>> {
304        let options = DeltaReaderExecutionOptions::new()
305            .with_max_concurrent_file_reads_per_scan(Some(2))?
306            .with_native_async_prefetch_file_count_per_partition(4)?;
307
308        assert_eq!(options.max_concurrent_file_reads_per_scan(), Some(2));
309        assert_eq!(options.max_concurrent_file_reads_per_partition(), 3);
310        assert_eq!(options.native_async_prefetch_file_count_per_partition(), 4);
311        Ok(())
312    }
313
314    #[test]
315    fn scan_capacity_resolves_once_from_the_fixed_partition_target()
316    -> Result<(), Box<dyn std::error::Error>> {
317        let defaults = DeltaReaderExecutionOptions::new();
318        assert_eq!(defaults.resolved_max_concurrent_file_reads_per_scan(4), 12);
319        assert_eq!(
320            defaults.resolved_max_concurrent_file_reads_per_scan(usize::MAX),
321            usize::MAX
322        );
323
324        let explicit = defaults.with_max_concurrent_file_reads_per_scan(Some(7))?;
325        assert_eq!(explicit.resolved_max_concurrent_file_reads_per_scan(4), 7);
326        Ok(())
327    }
328}
delta_arrow_reader/config.rs

delta_arrow_reader/
config.rs
