1use std::collections::{BTreeMap, HashMap, HashSet};
4
5use crate::{DeltaFunnelError, error::DeltaSourceSchemaSnafu, observability};
6use delta_kernel::object_store::aws::AmazonS3ConfigKey;
7
8mod deletion_vector;
9mod kernel;
10mod protocol;
11mod read;
12mod snapshot;
13#[cfg(test)]
14pub(crate) mod test_support;
15mod uri;
16
17use super::validate_table_source_names;
18#[allow(unused_imports)]
19pub(crate) use deletion_vector::{
20 KernelDeletionVectorReadRequest, KernelDeletionVectorReader, KernelDeletionVectorReaderConfig,
21 ProviderDeletionVectorSelection, ProviderDeletionVectorSelectionContext,
22};
23#[cfg(test)]
24pub(crate) use kernel::KernelStructType;
25use kernel::{ArrowSchemaRef, Version, snapshot_arrow_schema};
26pub(crate) use kernel::{
27 ColumnName as KernelColumnName, DecimalData as KernelDecimalData,
28 DeltaKernelPartitionScalarAdapterError, DeltaKernelPredicate, KernelColumnMetadataKey,
29 KernelDataType, KernelMetadataColumnSpec, KernelMetadataValue, KernelPrimitiveType,
30 KernelSchemaRef, KernelStructField, Scalar as KernelScalar,
31 arrow_partition_type_to_kernel_primitive, datafusion_expr_to_kernel_predicate,
32 kernel_partition_scalar_to_datafusion_scalar,
33};
34pub(crate) use protocol::preflight_delta_protocol_with_tracing;
35pub use protocol::{ProtocolPreflight, preflight_delta_protocol, preflight_delta_sources};
36pub(crate) use read::{
37 KernelDataFilePredicateEvalRequest, KernelDataFileReadRequest, KernelDataFileReader,
38 KernelDataFileReaderConfig, KernelDataFileTransformRequest, KernelScanReadSchema,
39};
40use snapshot::{LoadedDeltaTableSnapshot, load_delta_table_snapshot, s3_auth_mode_hint_for_source};
41
42#[allow(dead_code)]
49pub(crate) struct KernelScanMetadataExpansion {
50 pub(crate) files: Vec<KernelScanFileMetadata>,
52 pub(crate) scan_metadata_exhausted: bool,
58}
59
60#[allow(dead_code)]
62pub(crate) struct KernelScanFileMetadata {
63 pub(crate) path: String,
68 pub(crate) size: i64,
70 pub(crate) modification_time: i64,
75 pub(crate) stats: Option<KernelScanFileStats>,
77 pub(crate) deletion_vector: KernelScanDeletionVectorMetadata,
83 pub(crate) physical_to_logical_transform: KernelPhysicalToLogicalTransform,
89 pub(crate) partition_values: HashMap<String, String>,
93}
94
95#[allow(dead_code)]
97pub(crate) struct KernelScanFileStats {
98 pub(crate) num_records: u64,
100}
101
102#[allow(dead_code)]
107#[derive(Clone)]
108pub(crate) enum KernelScanDeletionVectorMetadata {
109 NotPresent,
111 Present(KernelScanDeletionVectorHandle),
113}
114
115#[allow(dead_code)]
121#[derive(Clone)]
122pub(crate) struct KernelScanDeletionVectorHandle {
123 dv_info: kernel::DvInfo,
125}
126
127impl KernelScanDeletionVectorHandle {
128 #[cfg(test)]
129 pub(crate) fn from_test_dv_info(dv_info: kernel::DvInfo) -> Self {
130 Self { dv_info }
131 }
132}
133
134impl KernelScanDeletionVectorMetadata {
135 #[must_use]
136 pub(crate) fn is_present(&self) -> bool {
137 matches!(self, Self::Present(_))
138 }
139
140 #[cfg(test)]
141 #[must_use]
142 pub(crate) fn test_present_from_descriptor(
143 descriptor: kernel::DeletionVectorDescriptor,
144 ) -> Self {
145 Self::Present(KernelScanDeletionVectorHandle {
146 dv_info: descriptor.into(),
147 })
148 }
149}
150
151#[allow(dead_code)]
156#[derive(Clone)]
157pub(crate) enum KernelPhysicalToLogicalTransform {
158 NotRequired,
160 Required(KernelPhysicalToLogicalTransformHandle),
162}
163
164#[allow(dead_code)]
169#[derive(Clone)]
170pub(crate) struct KernelPhysicalToLogicalTransformHandle {
171 transform: kernel::ExpressionRef,
173}
174
175pub type DeltaStorageOptions = BTreeMap<String, String>;
177
178pub struct DeltaSourceConfig {
180 pub name: String,
182 pub table_uri: String,
184 pub version: Option<Version>,
186 pub storage_options: DeltaStorageOptions,
188}
189
190impl DeltaSourceConfig {
191 #[must_use]
193 pub fn new(name: impl Into<String>, table_uri: impl Into<String>) -> Self {
194 Self {
195 name: name.into(),
196 table_uri: table_uri.into(),
197 version: None,
198 storage_options: DeltaStorageOptions::default(),
199 }
200 }
201
202 #[must_use]
204 pub fn with_version(mut self, version: Option<Version>) -> Self {
205 self.version = version;
206 self
207 }
208
209 #[must_use]
211 pub fn with_storage_options(mut self, storage_options: DeltaStorageOptions) -> Self {
212 self.storage_options = storage_options;
213 self
214 }
215}
216
217pub struct PlannedDeltaSource {
219 name: String,
220 requested_table_uri: String,
221 storage_options: DeltaStorageOptions,
222 snapshot: LoadedDeltaTableSnapshot,
223}
224
225#[allow(dead_code)]
227pub(crate) struct ProjectedDeltaScan {
228 scan: kernel::Scan,
229 kernel_schema: kernel::KernelSchemaRef,
230}
231
232impl ProjectedDeltaScan {
233 #[allow(dead_code)]
235 #[must_use]
236 pub(crate) fn kernel_schema(&self) -> &kernel::KernelSchemaRef {
237 &self.kernel_schema
238 }
239
240 #[allow(dead_code)]
242 #[must_use]
243 pub(crate) fn kernel_scan(&self) -> &kernel::Scan {
244 &self.scan
245 }
246
247 #[allow(dead_code)]
249 #[must_use]
250 pub(crate) fn read_schema(&self) -> KernelScanReadSchema {
251 KernelScanReadSchema::new(
252 self.scan.physical_schema().clone(),
253 self.scan.logical_schema().clone(),
254 self.scan.physical_predicate().clone(),
255 )
256 }
257
258 #[allow(dead_code)]
259 #[must_use]
260 pub(crate) fn physical_predicate(&self) -> Option<DeltaKernelPredicate> {
261 self.scan
262 .physical_predicate()
263 .clone()
264 .map(DeltaKernelPredicate::from_ref)
265 }
266
267 #[allow(dead_code)]
273 pub(crate) fn expand_kernel_scan_metadata(
274 &self,
275 table_uri: &str,
276 storage_options: &DeltaStorageOptions,
277 ) -> Result<KernelScanMetadataExpansion, delta_kernel::Error> {
278 fn collect_scan_file(files: &mut Vec<KernelScanFileMetadata>, file: kernel::ScanFile) {
279 files.push(KernelScanFileMetadata::from_kernel_scan_file(file));
280 }
281
282 let table_url = kernel::try_parse_uri(table_uri)?;
283 let store = kernel::store_from_url_opts(
284 &table_url,
285 storage_options
286 .iter()
287 .map(|(key, value)| (key.as_str(), value.as_str())),
288 )?;
289 let engine = kernel::DefaultEngineBuilder::new(store).build();
290 let mut files = Vec::new();
291
292 for scan_metadata in self.scan.scan_metadata(&engine)? {
293 files = scan_metadata?.visit_scan_files(files, collect_scan_file)?;
294 }
295
296 Ok(KernelScanMetadataExpansion {
297 files,
298 scan_metadata_exhausted: true,
299 })
300 }
301
302 #[cfg(test)]
303 pub(crate) fn scan_file_paths(
305 &self,
306 table_uri: &str,
307 storage_options: &DeltaStorageOptions,
308 ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
309 let mut paths = self
310 .expand_kernel_scan_metadata(table_uri, storage_options)?
311 .files
312 .into_iter()
313 .map(|file| file.path)
314 .collect::<Vec<_>>();
315 paths.sort();
316 Ok(paths)
317 }
318}
319
320impl KernelScanFileMetadata {
321 fn from_kernel_scan_file(file: kernel::ScanFile) -> Self {
322 let kernel::ScanFile {
323 path,
324 size,
325 modification_time,
326 stats,
327 dv_info,
328 transform,
329 partition_values,
330 } = file;
331
332 Self {
333 path,
334 size,
335 modification_time,
336 stats: stats.map(|stats| KernelScanFileStats {
337 num_records: stats.num_records,
338 }),
339 deletion_vector: KernelScanDeletionVectorMetadata::from_dv_info(dv_info),
340 physical_to_logical_transform: KernelPhysicalToLogicalTransform::from_transform(
341 transform,
342 ),
343 partition_values,
344 }
345 }
346}
347
348impl KernelScanDeletionVectorMetadata {
349 fn from_dv_info(dv_info: kernel::DvInfo) -> Self {
350 if dv_info.has_vector() {
351 Self::Present(KernelScanDeletionVectorHandle { dv_info })
352 } else {
353 Self::NotPresent
354 }
355 }
356}
357
358impl KernelPhysicalToLogicalTransform {
359 #[cfg(test)]
360 #[must_use]
361 pub(crate) fn is_required(&self) -> bool {
362 matches!(self, Self::Required(_))
363 }
364
365 #[cfg(test)]
366 pub(crate) fn test_required_column_transform(column_name: &str) -> Self {
367 let transform = std::sync::Arc::new(kernel::Expression::Column(kernel::ColumnName::new([
368 column_name,
369 ])));
370
371 Self::Required(KernelPhysicalToLogicalTransformHandle { transform })
372 }
373
374 #[cfg(test)]
375 pub(crate) fn test_required_orders_customer_name_replacement_transform(
376 replacement: &str,
377 ) -> Self {
378 let transform = std::sync::Arc::new(kernel::Expression::struct_from([
379 kernel::Expression::Column(kernel::ColumnName::new(["id"])),
380 kernel::Expression::Literal(kernel::Scalar::String(replacement.to_owned())),
381 ]));
382
383 Self::Required(KernelPhysicalToLogicalTransformHandle { transform })
384 }
385
386 fn from_transform(transform: Option<kernel::ExpressionRef>) -> Self {
387 match transform {
388 Some(transform) => Self::Required(KernelPhysicalToLogicalTransformHandle { transform }),
389 None => Self::NotRequired,
390 }
391 }
392}
393
394impl PlannedDeltaSource {
395 #[must_use]
397 pub fn name(&self) -> &str {
398 &self.name
399 }
400
401 #[must_use]
403 pub fn requested_table_uri(&self) -> &str {
404 &self.requested_table_uri
405 }
406
407 #[must_use]
409 pub fn storage_options(&self) -> &DeltaStorageOptions {
410 &self.storage_options
411 }
412
413 #[must_use]
415 pub fn table_uri(&self) -> &str {
416 self.loaded_snapshot().table_uri()
417 }
418
419 #[must_use]
421 pub fn version(&self) -> Version {
422 self.loaded_snapshot().version()
423 }
424
425 pub(crate) fn loaded_snapshot(&self) -> &LoadedDeltaTableSnapshot {
426 &self.snapshot
427 }
428}
429
430pub(crate) fn delta_source_arrow_schema(
431 source: &PlannedDeltaSource,
432) -> Result<ArrowSchemaRef, DeltaFunnelError> {
433 match snapshot_arrow_schema(source.loaded_snapshot().kernel_snapshot()) {
434 Ok(schema) => Ok(schema),
435 Err(error) => DeltaSourceSchemaSnafu {
436 source_name: source.name().to_owned(),
437 table_uri: source.table_uri().to_owned(),
438 reason: error.to_string(),
439 }
440 .fail(),
441 }
442}
443
444#[allow(dead_code)]
446pub(crate) fn build_projected_delta_scan(
447 source: &PlannedDeltaSource,
448 projected_column_names: Option<&[String]>,
449) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
450 build_projected_predicated_delta_scan(source, projected_column_names, None)
451}
452
453#[allow(dead_code)]
459pub(crate) fn build_projected_predicated_delta_scan(
460 source: &PlannedDeltaSource,
461 projected_column_names: Option<&[String]>,
462 predicate: Option<DeltaKernelPredicate>,
463) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
464 let (scan, kernel_schema) = kernel::build_projected_predicated_scan(
465 source.loaded_snapshot().kernel_snapshot(),
466 projected_column_names,
467 predicate,
468 )?;
469
470 Ok(ProjectedDeltaScan {
471 scan,
472 kernel_schema,
473 })
474}
475
476#[allow(dead_code)]
478pub(crate) fn build_projected_predicated_stats_delta_scan(
479 source: &PlannedDeltaSource,
480 projected_column_names: Option<&[String]>,
481 predicate: Option<DeltaKernelPredicate>,
482) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
483 let (scan, kernel_schema) = kernel::build_projected_predicated_stats_scan(
484 source.loaded_snapshot().kernel_snapshot(),
485 projected_column_names,
486 predicate,
487 )?;
488
489 Ok(ProjectedDeltaScan {
490 scan,
491 kernel_schema,
492 })
493}
494
495pub fn load_delta_source(
508 config: DeltaSourceConfig,
509) -> Result<PlannedDeltaSource, DeltaFunnelError> {
510 validate_table_source_names([config.name.as_str()])?;
511
512 load_delta_source_after_name_validation(config, false)
513}
514
515pub(crate) fn load_delta_source_with_tracing(
516 config: DeltaSourceConfig,
517) -> Result<PlannedDeltaSource, DeltaFunnelError> {
518 validate_table_source_names([config.name.as_str()])?;
519
520 load_delta_source_after_name_validation(config, true)
521}
522
523pub fn load_delta_sources<I>(configs: I) -> Result<Vec<PlannedDeltaSource>, DeltaFunnelError>
535where
536 I: IntoIterator<Item = DeltaSourceConfig>,
537{
538 let configs: Vec<_> = configs.into_iter().collect();
539
540 validate_table_source_names(configs.iter().map(|config| config.name.as_str()))?;
541
542 configs
543 .into_iter()
544 .map(|config| load_delta_source_after_name_validation(config, false))
545 .collect()
546}
547
548fn load_delta_source_after_name_validation(
549 mut config: DeltaSourceConfig,
550 emit_tracing: bool,
551) -> Result<PlannedDeltaSource, DeltaFunnelError> {
552 config.storage_options = effective_storage_options_for_source_from_env(
553 &config.table_uri,
554 config.storage_options,
555 std::env::vars(),
556 );
557 let source_name = config.name.clone();
558 let s3_auth_mode_hint =
559 s3_auth_mode_hint_for_source(&config.table_uri, &config.storage_options);
560 if emit_tracing {
561 observability::source_loading_started(
562 &source_name,
563 s3_auth_mode_hint.map(|hint| hint.as_str()),
564 );
565 }
566
567 let DeltaSourceConfig {
568 name,
569 table_uri,
570 version,
571 storage_options,
572 } = config;
573
574 let result = load_delta_table_snapshot(&table_uri, version, &storage_options).map(|snapshot| {
575 PlannedDeltaSource {
576 name,
577 requested_table_uri: table_uri,
578 storage_options,
579 snapshot,
580 }
581 });
582
583 if emit_tracing {
584 match &result {
585 Ok(source) => observability::source_loading_completed(
586 source.name(),
587 source.version(),
588 s3_auth_mode_hint.map(|hint| hint.as_str()),
589 ),
590 Err(error) => observability::source_loading_failed(
591 &source_name,
592 error,
593 s3_auth_mode_hint.map(|hint| hint.as_str()),
594 ),
595 }
596 }
597
598 result
599}
600
601fn effective_storage_options_for_source_from_env<I>(
602 table_uri: &str,
603 storage_options: DeltaStorageOptions,
604 env: I,
605) -> DeltaStorageOptions
606where
607 I: IntoIterator<Item = (String, String)>,
608{
609 if s3_auth_mode_hint_for_source(table_uri, &storage_options).is_none() {
610 return storage_options;
611 }
612
613 let caller_s3_keys = storage_options
614 .keys()
615 .filter_map(|key| s3_option_precedence_key(key))
616 .collect::<HashSet<_>>();
617 let mut effective = env
618 .into_iter()
619 .filter(|(key, _)| key.starts_with("AWS_"))
620 .filter(|(key, _)| match s3_option_precedence_key(key) {
621 Some(key) => !caller_s3_keys.contains(&key),
622 None => true,
623 })
624 .collect::<DeltaStorageOptions>();
625 effective.extend(storage_options);
626 effective
627}
628
629fn s3_option_precedence_key(key: &str) -> Option<String> {
630 let key = key.to_ascii_lowercase().parse::<AmazonS3ConfigKey>().ok()?;
631 Some(match key {
632 AmazonS3ConfigKey::DefaultRegion | AmazonS3ConfigKey::Region => "aws_region".to_owned(),
633 _ => key.as_ref().to_owned(),
634 })
635}
636
637#[cfg(test)]
638mod tests {
639 use std::collections::{BTreeMap, HashMap};
640 use std::fs;
641 use std::path::{Path, PathBuf};
642 use std::sync::{Arc, Mutex};
643 use std::time::{SystemTime, UNIX_EPOCH};
644
645 use datafusion::common::ScalarValue;
646 use datafusion::logical_expr::{Expr, col, lit};
647 use delta_kernel::arrow::array::{Array, StructArray};
648 use delta_kernel::arrow::record_batch::RecordBatch;
649 use delta_kernel::object_store::{
650 aws::{AmazonS3Builder, AmazonS3ConfigKey},
651 memory::InMemory,
652 path::Path as ObjectStorePath,
653 };
654
655 use super::kernel::{ColumnName, Expression, Predicate, Scalar};
656 use super::{
657 DeltaKernelPredicate, DeltaSourceConfig, DeltaStorageOptions, KernelDataFileReader,
658 KernelDataFileReaderConfig, KernelDeletionVectorReader, KernelDeletionVectorReaderConfig,
659 ProjectedDeltaScan, build_projected_delta_scan, build_projected_predicated_delta_scan,
660 datafusion_expr_to_kernel_predicate, effective_storage_options_for_source_from_env,
661 load_delta_source, load_delta_sources, load_delta_table_snapshot,
662 };
663 use crate::DeltaFunnelError;
664
665 struct DeltaLogTable {
666 path: PathBuf,
667 }
668
669 impl Drop for DeltaLogTable {
670 fn drop(&mut self) {
671 let _ = fs::remove_dir_all(&self.path);
672 }
673 }
674
675 impl DeltaLogTable {
676 fn new(name: &str) -> Result<Self, Box<dyn std::error::Error>> {
677 Self::new_with_metadata_and_adds(name, METADATA_JSON, &[add_json("part-00001.parquet")])
678 }
679
680 fn new_with_metadata_and_adds(
681 name: &str,
682 metadata_json: &str,
683 add_jsons: &[String],
684 ) -> Result<Self, Box<dyn std::error::Error>> {
685 Self::new_with_protocol_metadata_and_adds(name, PROTOCOL_JSON, metadata_json, add_jsons)
686 }
687
688 fn new_with_protocol_metadata_and_adds(
689 name: &str,
690 protocol_json: &str,
691 metadata_json: &str,
692 add_jsons: &[String],
693 ) -> Result<Self, Box<dyn std::error::Error>> {
694 let path = Path::new("target")
695 .join("delta-funnel-named-source-tests")
696 .join(unique_name(name)?);
697 let log_path = path.join("_delta_log");
698 fs::create_dir_all(&log_path)?;
699 fs::write(
700 log_path.join("00000000000000000000.json"),
701 format!("{protocol_json}\n{metadata_json}\n"),
702 )?;
703 fs::write(log_path.join("00000000000000000001.json"), {
704 let mut actions = add_jsons.join("\n");
705 actions.push('\n');
706 actions
707 })?;
708
709 Ok(Self { path })
710 }
711 }
712
713 const PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}"#;
714 const METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
715 const BOOLEAN_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"is_current\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
716 const BINARY_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"payload\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
717 const DATE_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
718 const DECIMAL_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"amount\",\"type\":\"decimal(10,2)\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
719 const FLOATING_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"float_score\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"double_score\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
720 const STRING_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"customer_name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
721 const TIMESTAMP_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_ts\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
722 const TIMESTAMP_NTZ_DATA_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"event_ts_ntz\",\"type\":\"timestamp_ntz\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
723 const PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"region\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["region"],"configuration":{},"createdTime":1587968585495}}"#;
724 const INTEGER_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"byte_part\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"short_part\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int_part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"long_part\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["byte_part","short_part","int_part","long_part"],"configuration":{},"createdTime":1587968585495}}"#;
725 const BOOLEAN_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"is_current\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["is_current"],"configuration":{},"createdTime":1587968585495}}"#;
726 const DATE_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_date"],"configuration":{},"createdTime":1587968585495}}"#;
727 const DECIMAL_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"amount\",\"type\":\"decimal(10,2)\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["amount"],"configuration":{},"createdTime":1587968585495}}"#;
728 const FLOATING_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float_part\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"double_part\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["float_part","double_part"],"configuration":{},"createdTime":1587968585495}}"#;
729 const BINARY_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"payload\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["payload"],"configuration":{},"createdTime":1587968585495}}"#;
730 const TIMESTAMP_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_ts\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_ts"],"configuration":{},"createdTime":1587968585495}}"#;
731 const TIMESTAMP_NTZ_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["timestampNtz"],"writerFeatures":["timestampNtz"]}}"#;
732 const TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"event_ts_ntz\",\"type\":\"timestamp_ntz\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["event_ts_ntz"],"configuration":{},"createdTime":1587968585495}}"#;
733 const DELETION_VECTOR_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#;
734
735 fn add_json(path: &str) -> String {
736 format!(
737 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true}}}}"#
738 )
739 }
740
741 fn partitioned_add_json(path: &str, partition_values_json: &str) -> String {
742 format!(
743 r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":0,"modificationTime":1587968586000,"dataChange":true}}}}"#
744 )
745 }
746
747 fn add_json_with_id_stats(
748 path: &str,
749 num_records: i64,
750 min_value: i32,
751 max_value: i32,
752 null_count: i64,
753 ) -> String {
754 format!(
755 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}"}}}}"#
756 )
757 }
758
759 fn add_json_with_boolean_stats(
760 path: &str,
761 num_records: i64,
762 min_value: bool,
763 max_value: bool,
764 null_count: i64,
765 ) -> String {
766 format!(
767 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"is_current\":{min_value}}},\"maxValues\":{{\"is_current\":{max_value}}},\"nullCount\":{{\"is_current\":{null_count}}}}}"}}}}"#
768 )
769 }
770
771 fn add_json_with_partial_boolean_stats(
772 path: &str,
773 num_records: i64,
774 min_value: Option<bool>,
775 max_value: Option<bool>,
776 null_count: Option<i64>,
777 ) -> String {
778 let min_values = min_value
779 .map(|value| format!(r#"\"is_current\":{value}"#))
780 .unwrap_or_default();
781 let max_values = max_value
782 .map(|value| format!(r#"\"is_current\":{value}"#))
783 .unwrap_or_default();
784 let null_count = null_count
785 .map(|value| format!(r#"\"is_current\":{value}"#))
786 .unwrap_or_default();
787
788 format!(
789 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
790 )
791 }
792
793 fn add_json_with_date_stats(
794 path: &str,
795 num_records: i64,
796 min_value: &str,
797 max_value: &str,
798 null_count: i64,
799 ) -> String {
800 format!(
801 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"event_date\":\"{min_value}\"}},\"maxValues\":{{\"event_date\":\"{max_value}\"}},\"nullCount\":{{\"event_date\":{null_count}}}}}"}}}}"#
802 )
803 }
804
805 fn add_json_with_partial_date_stats(
806 path: &str,
807 num_records: i64,
808 min_value: Option<&str>,
809 max_value: Option<&str>,
810 null_count: Option<i64>,
811 ) -> String {
812 let min_values = min_value
813 .map(|value| format!(r#"\"event_date\":\"{value}\""#))
814 .unwrap_or_default();
815 let max_values = max_value
816 .map(|value| format!(r#"\"event_date\":\"{value}\""#))
817 .unwrap_or_default();
818 let null_count = null_count
819 .map(|value| format!(r#"\"event_date\":{value}"#))
820 .unwrap_or_default();
821
822 format!(
823 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
824 )
825 }
826
827 fn add_json_with_decimal_stats(
828 path: &str,
829 num_records: i64,
830 min_value: &str,
831 max_value: &str,
832 null_count: i64,
833 ) -> String {
834 format!(
835 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"amount\":\"{min_value}\"}},\"maxValues\":{{\"amount\":\"{max_value}\"}},\"nullCount\":{{\"amount\":{null_count}}}}}"}}}}"#
836 )
837 }
838
839 fn add_json_with_partial_decimal_stats(
840 path: &str,
841 num_records: i64,
842 min_value: Option<&str>,
843 max_value: Option<&str>,
844 null_count: Option<i64>,
845 ) -> String {
846 let min_values = min_value
847 .map(|value| format!(r#"\"amount\":\"{value}\""#))
848 .unwrap_or_default();
849 let max_values = max_value
850 .map(|value| format!(r#"\"amount\":\"{value}\""#))
851 .unwrap_or_default();
852 let null_count = null_count
853 .map(|value| format!(r#"\"amount\":{value}"#))
854 .unwrap_or_default();
855
856 format!(
857 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
858 )
859 }
860
861 fn add_json_with_binary_stats(
862 path: &str,
863 num_records: i64,
864 min_value: &str,
865 max_value: &str,
866 null_count: i64,
867 ) -> String {
868 format!(
869 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"payload\":\"{min_value}\"}},\"maxValues\":{{\"payload\":\"{max_value}\"}},\"nullCount\":{{\"payload\":{null_count}}}}}"}}}}"#
870 )
871 }
872
873 fn add_json_with_partial_binary_stats(
874 path: &str,
875 num_records: i64,
876 min_value: Option<&str>,
877 max_value: Option<&str>,
878 null_count: Option<i64>,
879 ) -> String {
880 let min_values = min_value
881 .map(|value| format!(r#"\"payload\":\"{value}\""#))
882 .unwrap_or_default();
883 let max_values = max_value
884 .map(|value| format!(r#"\"payload\":\"{value}\""#))
885 .unwrap_or_default();
886 let null_count = null_count
887 .map(|value| format!(r#"\"payload\":{value}"#))
888 .unwrap_or_default();
889
890 format!(
891 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
892 )
893 }
894
895 fn add_json_with_floating_stats(
896 path: &str,
897 num_records: i64,
898 float_min_value: &str,
899 float_max_value: &str,
900 double_min_value: &str,
901 double_max_value: &str,
902 null_count: i64,
903 ) -> String {
904 format!(
905 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"float_score\":{float_min_value},\"double_score\":{double_min_value}}},\"maxValues\":{{\"float_score\":{float_max_value},\"double_score\":{double_max_value}}},\"nullCount\":{{\"float_score\":{null_count},\"double_score\":{null_count}}}}}"}}}}"#
906 )
907 }
908
909 fn add_json_with_partial_floating_stats(
910 path: &str,
911 num_records: i64,
912 float_min_value: Option<&str>,
913 float_max_value: Option<&str>,
914 double_min_value: Option<&str>,
915 double_max_value: Option<&str>,
916 null_count: Option<i64>,
917 ) -> String {
918 let mut min_values = Vec::new();
919 if let Some(value) = float_min_value {
920 min_values.push(format!(r#"\"float_score\":{value}"#));
921 }
922 if let Some(value) = double_min_value {
923 min_values.push(format!(r#"\"double_score\":{value}"#));
924 }
925
926 let mut max_values = Vec::new();
927 if let Some(value) = float_max_value {
928 max_values.push(format!(r#"\"float_score\":{value}"#));
929 }
930 if let Some(value) = double_max_value {
931 max_values.push(format!(r#"\"double_score\":{value}"#));
932 }
933
934 let null_count = null_count
935 .map(|value| format!(r#"\"float_score\":{value},\"double_score\":{value}"#))
936 .unwrap_or_default();
937
938 format!(
939 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{}}},\"maxValues\":{{{}}},\"nullCount\":{{{null_count}}}}}"}}}}"#,
940 min_values.join(","),
941 max_values.join(",")
942 )
943 }
944
945 fn add_json_with_string_stats(
946 path: &str,
947 num_records: i64,
948 min_value: &str,
949 max_value: &str,
950 null_count: i64,
951 ) -> String {
952 format!(
953 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"customer_name\":\"{min_value}\"}},\"maxValues\":{{\"customer_name\":\"{max_value}\"}},\"nullCount\":{{\"customer_name\":{null_count}}}}}"}}}}"#
954 )
955 }
956
957 fn add_json_with_partial_string_stats(
958 path: &str,
959 num_records: i64,
960 min_value: Option<&str>,
961 max_value: Option<&str>,
962 null_count: Option<i64>,
963 ) -> String {
964 let min_values = min_value
965 .map(|value| format!(r#"\"customer_name\":\"{value}\""#))
966 .unwrap_or_default();
967 let max_values = max_value
968 .map(|value| format!(r#"\"customer_name\":\"{value}\""#))
969 .unwrap_or_default();
970 let null_count = null_count
971 .map(|value| format!(r#"\"customer_name\":{value}"#))
972 .unwrap_or_default();
973
974 format!(
975 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
976 )
977 }
978
979 fn add_json_with_timestamp_stats(
980 path: &str,
981 column_name: &str,
982 num_records: i64,
983 min_value: &str,
984 max_value: &str,
985 null_count: i64,
986 ) -> String {
987 format!(
988 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"{column_name}\":\"{min_value}\"}},\"maxValues\":{{\"{column_name}\":\"{max_value}\"}},\"nullCount\":{{\"{column_name}\":{null_count}}}}}"}}}}"#
989 )
990 }
991
992 fn add_json_with_partial_timestamp_stats(
993 path: &str,
994 column_name: &str,
995 num_records: i64,
996 min_value: Option<&str>,
997 max_value: Option<&str>,
998 null_count: Option<i64>,
999 ) -> String {
1000 let min_values = min_value
1001 .map(|value| format!(r#"\"{column_name}\":\"{value}\""#))
1002 .unwrap_or_default();
1003 let max_values = max_value
1004 .map(|value| format!(r#"\"{column_name}\":\"{value}\""#))
1005 .unwrap_or_default();
1006 let null_count = null_count
1007 .map(|value| format!(r#"\"{column_name}\":{value}"#))
1008 .unwrap_or_default();
1009
1010 format!(
1011 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{{min_values}}},\"maxValues\":{{{max_values}}},\"nullCount\":{{{null_count}}}}}"}}}}"#
1012 )
1013 }
1014
1015 fn dv_add_json_with_id_stats(
1016 path: &str,
1017 num_records: i64,
1018 min_value: i32,
1019 max_value: i32,
1020 null_count: i64,
1021 ) -> String {
1022 format!(
1023 r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}","deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
1024 )
1025 }
1026
1027 fn partitioned_dv_add_json(path: &str, partition_values_json: &str) -> String {
1028 format!(
1029 r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":0,"modificationTime":1587968586000,"dataChange":true,"deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
1030 )
1031 }
1032
1033 fn partitioned_dv_add_json_with_id_stats_and_size(
1034 path: &str,
1035 partition_values_json: &str,
1036 size: i64,
1037 num_records: i64,
1038 min_value: i32,
1039 max_value: i32,
1040 null_count: i64,
1041 ) -> String {
1042 format!(
1043 r#"{{"add":{{"path":"{path}","partitionValues":{partition_values_json},"size":{size},"modificationTime":1587968586000,"dataChange":true,"stats":"{{\"numRecords\":{num_records},\"minValues\":{{\"id\":{min_value}}},\"maxValues\":{{\"id\":{max_value}}},\"nullCount\":{{\"id\":{null_count}}}}}","deletionVector":{{"storageType":"u","pathOrInlineDv":"vBn[lx{{q8@P<9BNH/isA","offset":1,"sizeInBytes":36,"cardinality":2}}}}}}"#
1044 )
1045 }
1046
1047 fn integer_partitioned_add_json(path: &str, value: &str) -> String {
1048 partitioned_add_json(
1049 path,
1050 &format!(
1051 r#"{{"byte_part":"{value}","short_part":"{value}","int_part":"{value}","long_part":"{value}"}}"#
1052 ),
1053 )
1054 }
1055
1056 fn unique_name(name: &str) -> Result<String, Box<dyn std::error::Error>> {
1057 let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
1058
1059 Ok(format!("{}-{}-{nanos}", std::process::id(), name))
1060 }
1061
1062 type CapturedStorageOptions = Arc<Mutex<Vec<DeltaStorageOptions>>>;
1063
1064 fn storage_options(entries: &[(&str, &str)]) -> DeltaStorageOptions {
1065 entries
1066 .iter()
1067 .map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
1068 .collect()
1069 }
1070
1071 fn env_vars(entries: &[(&str, &str)]) -> Vec<(String, String)> {
1072 entries
1073 .iter()
1074 .map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
1075 .collect()
1076 }
1077
1078 fn s3_builder_from_storage_options(options: &DeltaStorageOptions) -> AmazonS3Builder {
1079 options.iter().fold(
1080 AmazonS3Builder::new().with_url("s3://bucket/root"),
1081 |builder, (key, value)| match key.to_ascii_lowercase().parse::<AmazonS3ConfigKey>() {
1082 Ok(key) => builder.with_config(key, value.as_str()),
1083 Err(_) => builder,
1084 },
1085 )
1086 }
1087
1088 fn unique_storage_scheme(name: &str) -> Result<String, Box<dyn std::error::Error>> {
1089 let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
1090 let sanitized_name = name
1091 .chars()
1092 .filter(|character| character.is_ascii_alphanumeric())
1093 .collect::<String>();
1094
1095 Ok(format!("df{sanitized_name}{}{}", std::process::id(), nanos))
1096 }
1097
1098 fn register_capturing_storage_handler(
1099 scheme: &str,
1100 captured: CapturedStorageOptions,
1101 ) -> Result<(), Box<dyn std::error::Error>> {
1102 super::kernel::insert_url_handler(
1103 scheme,
1104 Arc::new(move |_url, options| {
1105 let options = options.into_iter().collect::<BTreeMap<_, _>>();
1106 captured
1107 .lock()
1108 .map_err(|_| delta_kernel::object_store::Error::Generic {
1109 store: "capture",
1110 source: std::io::Error::other("captured storage options lock poisoned")
1111 .into(),
1112 })?
1113 .push(options);
1114
1115 Ok((Box::new(InMemory::new()), ObjectStorePath::from("")))
1116 }),
1117 )?;
1118
1119 Ok(())
1120 }
1121
1122 fn register_failing_storage_handler(
1123 scheme: &str,
1124 captured: CapturedStorageOptions,
1125 secret_value: &'static str,
1126 ) -> Result<(), Box<dyn std::error::Error>> {
1127 super::kernel::insert_url_handler(
1128 scheme,
1129 Arc::new(move |_url, options| {
1130 let options = options.into_iter().collect::<BTreeMap<_, _>>();
1131 captured
1132 .lock()
1133 .map_err(|_| delta_kernel::object_store::Error::Generic {
1134 store: "capture",
1135 source: std::io::Error::other("captured storage options lock poisoned")
1136 .into(),
1137 })?
1138 .push(options);
1139
1140 Err(delta_kernel::object_store::Error::Generic {
1141 store: "capture",
1142 source: std::io::Error::other(secret_value).into(),
1143 })
1144 }),
1145 )?;
1146
1147 Ok(())
1148 }
1149
1150 fn captured_storage_options(captured: &CapturedStorageOptions) -> Vec<DeltaStorageOptions> {
1151 captured
1152 .lock()
1153 .map(|options| options.clone())
1154 .unwrap_or_default()
1155 }
1156
1157 fn kernel_scan_file_paths(
1158 scan: &ProjectedDeltaScan,
1159 table_uri: &str,
1160 ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
1161 let storage_options = DeltaStorageOptions::default();
1162 let mut paths = scan
1163 .expand_kernel_scan_metadata(table_uri, &storage_options)?
1164 .files
1165 .into_iter()
1166 .map(|file| file.path)
1167 .collect::<Vec<_>>();
1168 paths.sort();
1169 Ok(paths)
1170 }
1171
1172 #[derive(Debug, PartialEq, Eq)]
1173 struct KernelScanFileBoundary {
1174 path: String,
1175 has_deletion_vector: bool,
1176 }
1177
1178 fn kernel_scan_file_boundaries(
1179 scan: &ProjectedDeltaScan,
1180 table_uri: &str,
1181 ) -> Result<Vec<KernelScanFileBoundary>, Box<dyn std::error::Error>> {
1182 let storage_options = DeltaStorageOptions::default();
1183 let mut boundaries = scan
1184 .expand_kernel_scan_metadata(table_uri, &storage_options)?
1185 .files
1186 .into_iter()
1187 .map(|file| KernelScanFileBoundary {
1188 path: file.path,
1189 has_deletion_vector: file.deletion_vector.is_present(),
1190 })
1191 .collect::<Vec<_>>();
1192 boundaries.sort_by(|left, right| left.path.cmp(&right.path));
1193 Ok(boundaries)
1194 }
1195
1196 #[test]
1197 fn kernel_scan_metadata_expansion_preserves_file_metadata_without_data_reads()
1198 -> Result<(), Box<dyn std::error::Error>> {
1199 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
1200 "kernel-scan-metadata-expansion",
1201 DELETION_VECTOR_PROTOCOL_JSON,
1202 PARTITIONED_METADATA_JSON,
1203 &[partitioned_dv_add_json_with_id_stats_and_size(
1204 "region-west-dv.parquet",
1205 r#"{"region":"us-west"}"#,
1206 123,
1207 7,
1208 10,
1209 20,
1210 0,
1211 )],
1212 )?;
1213 let source = load_delta_source(DeltaSourceConfig {
1214 name: "orders".to_owned(),
1215 table_uri: table.path.to_string_lossy().to_string(),
1216 version: None,
1217 storage_options: Default::default(),
1218 })?;
1219 let scan = build_projected_predicated_stats_delta_scan(&source, None)?;
1220
1221 let expansion =
1222 scan.expand_kernel_scan_metadata(source.table_uri(), source.storage_options())?;
1223
1224 assert!(expansion.scan_metadata_exhausted);
1225 assert_eq!(expansion.files.len(), 1);
1226 let file = &expansion.files[0];
1227 assert_eq!(file.path, "region-west-dv.parquet");
1228 assert_eq!(file.size, 123);
1229 assert_eq!(file.modification_time, 1587968586000);
1230 assert_eq!(file.stats.as_ref().map(|stats| stats.num_records), Some(7));
1231 assert!(file.deletion_vector.is_present());
1232 assert_eq!(
1233 file.partition_values.get("region").map(String::as_str),
1234 Some("us-west")
1235 );
1236
1237 Ok(())
1238 }
1239
1240 #[test]
1241 fn kernel_scan_metadata_expansion_reports_empty_scan_as_exhausted()
1242 -> Result<(), Box<dyn std::error::Error>> {
1243 let table = DeltaLogTable::new_with_metadata_and_adds(
1244 "kernel-scan-metadata-empty",
1245 METADATA_JSON,
1246 &[],
1247 )?;
1248 let source = load_delta_source(DeltaSourceConfig {
1249 name: "orders".to_owned(),
1250 table_uri: table.path.to_string_lossy().to_string(),
1251 version: None,
1252 storage_options: Default::default(),
1253 })?;
1254 let scan = build_projected_predicated_stats_delta_scan(&source, None)?;
1255
1256 let expansion =
1257 scan.expand_kernel_scan_metadata(source.table_uri(), source.storage_options())?;
1258
1259 assert!(expansion.scan_metadata_exhausted);
1260 assert!(expansion.files.is_empty());
1261
1262 Ok(())
1263 }
1264
1265 #[test]
1266 fn kernel_scan_file_metadata_preserves_transform_handle_without_execution()
1267 -> Result<(), Box<dyn std::error::Error>> {
1268 let transform = Arc::new(Expression::Column(ColumnName::new(["physical_id"])));
1269 let partition_values = HashMap::from([("region".to_owned(), "us-west".to_owned())]);
1270
1271 let file = super::kernel::ScanFile {
1272 path: "part-with-transform.parquet".to_owned(),
1273 size: 456,
1274 modification_time: 1587968587000,
1275 stats: None,
1276 dv_info: super::kernel::DvInfo::default(),
1277 transform: Some(Arc::clone(&transform)),
1278 partition_values,
1279 };
1280
1281 let metadata = super::KernelScanFileMetadata::from_kernel_scan_file(file);
1282
1283 assert_eq!(metadata.path, "part-with-transform.parquet");
1284 assert_eq!(metadata.size, 456);
1285 assert_eq!(metadata.modification_time, 1587968587000);
1286 assert!(metadata.stats.is_none());
1287 assert!(!metadata.deletion_vector.is_present());
1288 assert!(metadata.physical_to_logical_transform.is_required());
1289 assert_eq!(
1290 metadata.partition_values.get("region").map(String::as_str),
1291 Some("us-west")
1292 );
1293 match metadata.physical_to_logical_transform {
1294 super::KernelPhysicalToLogicalTransform::Required(handle) => {
1295 assert!(Arc::ptr_eq(&handle.transform, &transform));
1296 }
1297 super::KernelPhysicalToLogicalTransform::NotRequired => {
1298 return Err("transform handle should be preserved".into());
1299 }
1300 }
1301
1302 Ok(())
1303 }
1304
1305 fn kernel_partition_characterization_source()
1306 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1307 let table = DeltaLogTable::new_with_metadata_and_adds(
1308 "kernel-partition-characterization",
1309 PARTITIONED_METADATA_JSON,
1310 &[
1311 partitioned_add_json("region-us-west.parquet", r#"{"region":"us-west"}"#),
1312 partitioned_add_json("region-us-east.parquet", r#"{"region":"us-east"}"#),
1313 partitioned_add_json("region-null.parquet", r#"{"region":null}"#),
1314 partitioned_add_json("region-missing.parquet", r#"{}"#),
1315 partitioned_add_json("region-empty-string.parquet", r#"{"region":""}"#),
1316 ],
1317 )?;
1318 let source = load_delta_source(DeltaSourceConfig {
1319 name: "orders".to_owned(),
1320 table_uri: table.path.to_string_lossy().to_string(),
1321 version: None,
1322 storage_options: Default::default(),
1323 })?;
1324
1325 Ok((table, source))
1326 }
1327
1328 fn kernel_integer_partition_characterization_source()
1329 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1330 let table = DeltaLogTable::new_with_metadata_and_adds(
1331 "kernel-integer-partition-characterization",
1332 INTEGER_PARTITIONED_METADATA_JSON,
1333 &[
1334 integer_partitioned_add_json("integer--1.parquet", "-1"),
1335 integer_partitioned_add_json("integer-2.parquet", "2"),
1336 integer_partitioned_add_json("integer-10.parquet", "10"),
1337 partitioned_add_json(
1338 "integer-null.parquet",
1339 r#"{"byte_part":null,"short_part":null,"int_part":null,"long_part":null}"#,
1340 ),
1341 integer_partitioned_add_json("integer-empty.parquet", ""),
1342 partitioned_add_json("integer-missing.parquet", r#"{}"#),
1343 ],
1344 )?;
1345 let source = load_delta_source(DeltaSourceConfig {
1346 name: "orders".to_owned(),
1347 table_uri: table.path.to_string_lossy().to_string(),
1348 version: None,
1349 storage_options: Default::default(),
1350 })?;
1351
1352 Ok((table, source))
1353 }
1354
1355 fn kernel_boolean_partition_characterization_source()
1356 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1357 let table = DeltaLogTable::new_with_metadata_and_adds(
1358 "kernel-boolean-partition-characterization",
1359 BOOLEAN_PARTITIONED_METADATA_JSON,
1360 &[
1361 partitioned_add_json("boolean-true.parquet", r#"{"is_current":"true"}"#),
1362 partitioned_add_json("boolean-false.parquet", r#"{"is_current":"false"}"#),
1363 partitioned_add_json("boolean-null.parquet", r#"{"is_current":null}"#),
1364 partitioned_add_json("boolean-empty.parquet", r#"{"is_current":""}"#),
1365 partitioned_add_json("boolean-missing.parquet", r#"{}"#),
1366 ],
1367 )?;
1368 let source = load_delta_source(DeltaSourceConfig {
1369 name: "orders".to_owned(),
1370 table_uri: table.path.to_string_lossy().to_string(),
1371 version: None,
1372 storage_options: Default::default(),
1373 })?;
1374
1375 Ok((table, source))
1376 }
1377
1378 fn kernel_date_partition_characterization_source()
1379 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1380 let table = DeltaLogTable::new_with_metadata_and_adds(
1381 "kernel-date-partition-characterization",
1382 DATE_PARTITIONED_METADATA_JSON,
1383 &[
1384 partitioned_add_json("date-pre-epoch.parquet", r#"{"event_date":"1969-12-31"}"#),
1385 partitioned_add_json("date-epoch.parquet", r#"{"event_date":"1970-01-01"}"#),
1386 partitioned_add_json("date-leap-day.parquet", r#"{"event_date":"2024-02-29"}"#),
1387 partitioned_add_json("date-new-year.parquet", r#"{"event_date":"2026-01-01"}"#),
1388 partitioned_add_json("date-null.parquet", r#"{"event_date":null}"#),
1389 partitioned_add_json("date-empty.parquet", r#"{"event_date":""}"#),
1390 partitioned_add_json("date-missing.parquet", r#"{}"#),
1391 ],
1392 )?;
1393 let source = load_delta_source(DeltaSourceConfig {
1394 name: "orders".to_owned(),
1395 table_uri: table.path.to_string_lossy().to_string(),
1396 version: None,
1397 storage_options: Default::default(),
1398 })?;
1399
1400 Ok((table, source))
1401 }
1402
1403 fn kernel_decimal_partition_characterization_source()
1404 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1405 let table = DeltaLogTable::new_with_metadata_and_adds(
1406 "kernel-decimal-partition-characterization",
1407 DECIMAL_PARTITIONED_METADATA_JSON,
1408 &[
1409 partitioned_add_json("decimal-negative.parquet", r#"{"amount":"-1.23"}"#),
1410 partitioned_add_json("decimal-zero.parquet", r#"{"amount":"0.00"}"#),
1411 partitioned_add_json("decimal-two.parquet", r#"{"amount":"2.00"}"#),
1412 partitioned_add_json("decimal-ten.parquet", r#"{"amount":"10.00"}"#),
1413 partitioned_add_json("decimal-large.parquet", r#"{"amount":"123.45"}"#),
1414 partitioned_add_json("decimal-null.parquet", r#"{"amount":null}"#),
1415 partitioned_add_json("decimal-empty.parquet", r#"{"amount":""}"#),
1416 partitioned_add_json("decimal-missing.parquet", r#"{}"#),
1417 ],
1418 )?;
1419 let source = load_delta_source(DeltaSourceConfig {
1420 name: "orders".to_owned(),
1421 table_uri: table.path.to_string_lossy().to_string(),
1422 version: None,
1423 storage_options: Default::default(),
1424 })?;
1425
1426 Ok((table, source))
1427 }
1428
1429 fn kernel_floating_partition_characterization_source()
1430 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1431 let table = DeltaLogTable::new_with_metadata_and_adds(
1432 "kernel-floating-partition-characterization",
1433 FLOATING_PARTITIONED_METADATA_JSON,
1434 &[
1435 partitioned_add_json(
1436 "floating-neg.parquet",
1437 r#"{"float_part":"-1.5","double_part":"-2.25"}"#,
1438 ),
1439 partitioned_add_json(
1440 "floating-neg-zero.parquet",
1441 r#"{"float_part":"-0.0","double_part":"-0.0"}"#,
1442 ),
1443 partitioned_add_json(
1444 "floating-pos-zero.parquet",
1445 r#"{"float_part":"0.0","double_part":"0.0"}"#,
1446 ),
1447 partitioned_add_json(
1448 "floating-one.parquet",
1449 r#"{"float_part":"1.5","double_part":"2.25"}"#,
1450 ),
1451 partitioned_add_json(
1452 "floating-ten.parquet",
1453 r#"{"float_part":"10.0","double_part":"10.0"}"#,
1454 ),
1455 partitioned_add_json(
1456 "floating-null.parquet",
1457 r#"{"float_part":null,"double_part":null}"#,
1458 ),
1459 partitioned_add_json(
1460 "floating-empty.parquet",
1461 r#"{"float_part":"","double_part":""}"#,
1462 ),
1463 partitioned_add_json("floating-missing.parquet", r#"{}"#),
1464 ],
1465 )?;
1466 let source = load_delta_source(DeltaSourceConfig {
1467 name: "orders".to_owned(),
1468 table_uri: table.path.to_string_lossy().to_string(),
1469 version: None,
1470 storage_options: Default::default(),
1471 })?;
1472
1473 Ok((table, source))
1474 }
1475
1476 fn kernel_binary_partition_characterization_source()
1477 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1478 let table = DeltaLogTable::new_with_metadata_and_adds(
1479 "kernel-binary-partition-characterization",
1480 BINARY_PARTITIONED_METADATA_JSON,
1481 &[
1482 partitioned_add_json("binary-HELLO.parquet", r#"{"payload":"HELLO"}"#),
1483 partitioned_add_json("binary-hello.parquet", r#"{"payload":"hello"}"#),
1484 partitioned_add_json("binary-special.parquet", r#"{"payload":"/=%"}"#),
1485 partitioned_add_json("binary-null.parquet", r#"{"payload":null}"#),
1486 partitioned_add_json("binary-empty.parquet", r#"{"payload":""}"#),
1487 partitioned_add_json("binary-missing.parquet", r#"{}"#),
1488 ],
1489 )?;
1490 let source = load_delta_source(DeltaSourceConfig {
1491 name: "orders".to_owned(),
1492 table_uri: table.path.to_string_lossy().to_string(),
1493 version: None,
1494 storage_options: Default::default(),
1495 })?;
1496
1497 Ok((table, source))
1498 }
1499
1500 fn kernel_timestamp_partition_characterization_source()
1501 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1502 let table = DeltaLogTable::new_with_metadata_and_adds(
1503 "kernel-timestamp-partition-characterization",
1504 TIMESTAMP_PARTITIONED_METADATA_JSON,
1505 &[
1506 partitioned_add_json(
1507 "timestamp-pre-epoch.parquet",
1508 r#"{"event_ts":"1969-12-31T23:59:59.999999Z"}"#,
1509 ),
1510 partitioned_add_json(
1511 "timestamp-low.parquet",
1512 r#"{"event_ts":"2025-12-31T23:59:59.999999Z"}"#,
1513 ),
1514 partitioned_add_json(
1515 "timestamp-target.parquet",
1516 r#"{"event_ts":"2026-01-01T00:00:00.123456Z"}"#,
1517 ),
1518 partitioned_add_json(
1519 "timestamp-high.parquet",
1520 r#"{"event_ts":"2026-01-01T00:00:00.123457Z"}"#,
1521 ),
1522 partitioned_add_json("timestamp-null.parquet", r#"{"event_ts":null}"#),
1523 partitioned_add_json("timestamp-empty.parquet", r#"{"event_ts":""}"#),
1524 partitioned_add_json("timestamp-missing.parquet", r#"{}"#),
1525 ],
1526 )?;
1527 let source = load_delta_source(DeltaSourceConfig {
1528 name: "orders".to_owned(),
1529 table_uri: table.path.to_string_lossy().to_string(),
1530 version: None,
1531 storage_options: Default::default(),
1532 })?;
1533
1534 Ok((table, source))
1535 }
1536
1537 fn kernel_timestamp_ntz_partition_characterization_source()
1538 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1539 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
1540 "kernel-timestamp-ntz-partition-characterization",
1541 TIMESTAMP_NTZ_PROTOCOL_JSON,
1542 TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
1543 &[
1544 partitioned_add_json(
1545 "timestamp-ntz-pre-epoch.parquet",
1546 r#"{"event_ts_ntz":"1969-12-31 23:59:59.999999"}"#,
1547 ),
1548 partitioned_add_json(
1549 "timestamp-ntz-low-space.parquet",
1550 r#"{"event_ts_ntz":"2025-12-31 23:59:59.999999"}"#,
1551 ),
1552 partitioned_add_json(
1553 "timestamp-ntz-target-space.parquet",
1554 r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
1555 ),
1556 partitioned_add_json(
1557 "timestamp-ntz-high.parquet",
1558 r#"{"event_ts_ntz":"2026-01-01 00:00:00.123457"}"#,
1559 ),
1560 partitioned_add_json("timestamp-ntz-null.parquet", r#"{"event_ts_ntz":null}"#),
1561 partitioned_add_json("timestamp-ntz-empty.parquet", r#"{"event_ts_ntz":""}"#),
1562 partitioned_add_json("timestamp-ntz-missing.parquet", r#"{}"#),
1563 ],
1564 )?;
1565 let source = load_delta_source(DeltaSourceConfig {
1566 name: "orders".to_owned(),
1567 table_uri: table.path.to_string_lossy().to_string(),
1568 version: None,
1569 storage_options: Default::default(),
1570 })?;
1571
1572 Ok((table, source))
1573 }
1574
1575 fn kernel_data_stats_characterization_source()
1576 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1577 let table = DeltaLogTable::new_with_metadata_and_adds(
1578 "kernel-data-stats-characterization",
1579 METADATA_JSON,
1580 &[
1581 add_json_with_id_stats("id-impossible.parquet", 10, 1, 50, 0),
1582 add_json_with_id_stats("id-possible.parquet", 10, 101, 150, 0),
1583 add_json("id-missing-stats.parquet"),
1584 ],
1585 )?;
1586 let source = load_delta_source(DeltaSourceConfig {
1587 name: "orders".to_owned(),
1588 table_uri: table.path.to_string_lossy().to_string(),
1589 version: None,
1590 storage_options: Default::default(),
1591 })?;
1592
1593 Ok((table, source))
1594 }
1595
1596 fn kernel_boolean_data_stats_characterization_source()
1597 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1598 let table = DeltaLogTable::new_with_metadata_and_adds(
1599 "kernel-boolean-data-stats-characterization",
1600 BOOLEAN_DATA_METADATA_JSON,
1601 &[
1602 add_json_with_boolean_stats("boolean-false-only.parquet", 10, false, false, 0),
1603 add_json_with_boolean_stats("boolean-true-only.parquet", 10, true, true, 0),
1604 add_json_with_boolean_stats("boolean-mixed.parquet", 10, false, true, 0),
1605 add_json_with_boolean_stats("boolean-false-with-null.parquet", 10, false, false, 2),
1606 add_json_with_boolean_stats("boolean-true-with-null.parquet", 10, true, true, 2),
1607 add_json_with_partial_boolean_stats(
1608 "boolean-all-null.parquet",
1609 10,
1610 None,
1611 None,
1612 Some(10),
1613 ),
1614 add_json("boolean-missing-stats.parquet"),
1615 ],
1616 )?;
1617 let source = load_delta_source(DeltaSourceConfig {
1618 name: "orders".to_owned(),
1619 table_uri: table.path.to_string_lossy().to_string(),
1620 version: None,
1621 storage_options: Default::default(),
1622 })?;
1623
1624 Ok((table, source))
1625 }
1626
1627 fn kernel_partial_boolean_data_stats_characterization_source()
1628 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1629 let table = DeltaLogTable::new_with_metadata_and_adds(
1630 "kernel-partial-boolean-data-stats-characterization",
1631 BOOLEAN_DATA_METADATA_JSON,
1632 &[
1633 add_json_with_partial_boolean_stats(
1634 "boolean-min-only-false.parquet",
1635 10,
1636 Some(false),
1637 None,
1638 Some(0),
1639 ),
1640 add_json_with_partial_boolean_stats(
1641 "boolean-max-only-true.parquet",
1642 10,
1643 None,
1644 Some(true),
1645 Some(0),
1646 ),
1647 add_json_with_partial_boolean_stats(
1648 "boolean-counts-only.parquet",
1649 10,
1650 None,
1651 None,
1652 Some(0),
1653 ),
1654 add_json_with_partial_boolean_stats(
1655 "boolean-missing-null-count.parquet",
1656 10,
1657 Some(false),
1658 Some(true),
1659 None,
1660 ),
1661 add_json("boolean-missing-stats.parquet"),
1662 ],
1663 )?;
1664 let source = load_delta_source(DeltaSourceConfig {
1665 name: "orders".to_owned(),
1666 table_uri: table.path.to_string_lossy().to_string(),
1667 version: None,
1668 storage_options: Default::default(),
1669 })?;
1670
1671 Ok((table, source))
1672 }
1673
1674 fn assert_boolean_stats_min_max_error(
1675 result: Result<Vec<String>, Box<dyn std::error::Error>>,
1676 ) -> Result<(), Box<dyn std::error::Error>> {
1677 let error = match result {
1678 Ok(paths) => {
1679 return Err(format!(
1680 "boolean min/max stats predicate should fail kernel scan: {paths:?}"
1681 )
1682 .into());
1683 }
1684 Err(error) => error,
1685 };
1686 let message = error.to_string();
1687 let debug_message = format!("{error:?}");
1688 assert!(
1689 message.contains("minValues") || debug_message.contains("minValues"),
1690 "{message}\n{debug_message}"
1691 );
1692
1693 Ok(())
1694 }
1695
1696 fn assert_binary_stats_min_max_error(
1697 result: Result<Vec<String>, Box<dyn std::error::Error>>,
1698 ) -> Result<(), Box<dyn std::error::Error>> {
1699 let error = match result {
1700 Ok(paths) => {
1701 return Err(format!(
1702 "binary min/max stats predicate should fail kernel scan: {paths:?}"
1703 )
1704 .into());
1705 }
1706 Err(error) => error,
1707 };
1708 let message = error.to_string();
1709 let debug_message = format!("{error:?}");
1710 assert!(
1711 message.contains("minValues")
1712 || message.contains("maxValues")
1713 || debug_message.contains("minValues")
1714 || debug_message.contains("maxValues"),
1715 "{message}\n{debug_message}"
1716 );
1717
1718 Ok(())
1719 }
1720
1721 fn assert_unsupported_literal_error(
1722 result: Result<Vec<String>, Box<dyn std::error::Error>>,
1723 ) -> Result<(), Box<dyn std::error::Error>> {
1724 let error = match result {
1725 Ok(paths) => {
1726 return Err(format!(
1727 "unsupported literal predicate should fail before kernel scan: {paths:?}"
1728 )
1729 .into());
1730 }
1731 Err(error) => error,
1732 };
1733 let message = error.to_string();
1734 let debug_message = format!("{error:?}");
1735 assert!(
1736 message.contains("unsupported DataFusion literal")
1737 || debug_message.contains("UnsupportedLiteral"),
1738 "{message}\n{debug_message}"
1739 );
1740
1741 Ok(())
1742 }
1743
1744 fn kernel_date_data_stats_characterization_source()
1745 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1746 let table = DeltaLogTable::new_with_metadata_and_adds(
1747 "kernel-date-data-stats-characterization",
1748 DATE_DATA_METADATA_JSON,
1749 &[
1750 add_json_with_date_stats(
1751 "date-pre-epoch-only.parquet",
1752 10,
1753 "1969-12-31",
1754 "1969-12-31",
1755 0,
1756 ),
1757 add_json_with_date_stats(
1758 "date-leap-only.parquet",
1759 10,
1760 "2024-02-29",
1761 "2024-02-29",
1762 0,
1763 ),
1764 add_json_with_date_stats(
1765 "date-new-year-only.parquet",
1766 10,
1767 "2026-01-01",
1768 "2026-01-01",
1769 0,
1770 ),
1771 add_json_with_date_stats("date-range.parquet", 10, "2024-02-29", "2026-01-01", 0),
1772 add_json_with_date_stats(
1773 "date-new-year-with-null.parquet",
1774 10,
1775 "2026-01-01",
1776 "2026-01-01",
1777 2,
1778 ),
1779 add_json_with_partial_date_stats("date-all-null.parquet", 10, None, None, Some(10)),
1780 add_json("date-missing-stats.parquet"),
1781 ],
1782 )?;
1783 let source = load_delta_source(DeltaSourceConfig {
1784 name: "orders".to_owned(),
1785 table_uri: table.path.to_string_lossy().to_string(),
1786 version: None,
1787 storage_options: Default::default(),
1788 })?;
1789
1790 Ok((table, source))
1791 }
1792
1793 fn kernel_partial_date_data_stats_characterization_source()
1794 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1795 let table = DeltaLogTable::new_with_metadata_and_adds(
1796 "kernel-partial-date-data-stats-characterization",
1797 DATE_DATA_METADATA_JSON,
1798 &[
1799 add_json_with_partial_date_stats(
1800 "date-min-only-high.parquet",
1801 10,
1802 Some("2026-01-01"),
1803 None,
1804 Some(0),
1805 ),
1806 add_json_with_partial_date_stats(
1807 "date-max-only-low.parquet",
1808 10,
1809 None,
1810 Some("2024-02-29"),
1811 Some(0),
1812 ),
1813 add_json_with_partial_date_stats(
1814 "date-counts-only.parquet",
1815 10,
1816 None,
1817 None,
1818 Some(0),
1819 ),
1820 add_json_with_partial_date_stats(
1821 "date-missing-null-count.parquet",
1822 10,
1823 Some("2024-02-29"),
1824 Some("2026-01-01"),
1825 None,
1826 ),
1827 add_json("date-missing-stats.parquet"),
1828 ],
1829 )?;
1830 let source = load_delta_source(DeltaSourceConfig {
1831 name: "orders".to_owned(),
1832 table_uri: table.path.to_string_lossy().to_string(),
1833 version: None,
1834 storage_options: Default::default(),
1835 })?;
1836
1837 Ok((table, source))
1838 }
1839
1840 fn kernel_decimal_data_stats_characterization_source()
1841 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1842 let table = DeltaLogTable::new_with_metadata_and_adds(
1843 "kernel-decimal-data-stats-characterization",
1844 DECIMAL_DATA_METADATA_JSON,
1845 &[
1846 add_json_with_decimal_stats(
1847 "decimal-negative-only.parquet",
1848 10,
1849 "-1.23",
1850 "-1.23",
1851 0,
1852 ),
1853 add_json_with_decimal_stats("decimal-zero-only.parquet", 10, "0.00", "0.00", 0),
1854 add_json_with_decimal_stats("decimal-two-only.parquet", 10, "2.00", "2.00", 0),
1855 add_json_with_decimal_stats("decimal-ten-only.parquet", 10, "10.00", "10.00", 0),
1856 add_json_with_decimal_stats(
1857 "decimal-large-only.parquet",
1858 10,
1859 "123.45",
1860 "123.45",
1861 0,
1862 ),
1863 add_json_with_decimal_stats("decimal-range.parquet", 10, "0.00", "10.00", 0),
1864 add_json_with_decimal_stats("decimal-two-with-null.parquet", 10, "2.00", "2.00", 2),
1865 add_json_with_partial_decimal_stats(
1866 "decimal-all-null.parquet",
1867 10,
1868 None,
1869 None,
1870 Some(10),
1871 ),
1872 add_json("decimal-missing-stats.parquet"),
1873 ],
1874 )?;
1875 let source = load_delta_source(DeltaSourceConfig {
1876 name: "orders".to_owned(),
1877 table_uri: table.path.to_string_lossy().to_string(),
1878 version: None,
1879 storage_options: Default::default(),
1880 })?;
1881
1882 Ok((table, source))
1883 }
1884
1885 fn kernel_partial_decimal_data_stats_characterization_source()
1886 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1887 let table = DeltaLogTable::new_with_metadata_and_adds(
1888 "kernel-partial-decimal-data-stats-characterization",
1889 DECIMAL_DATA_METADATA_JSON,
1890 &[
1891 add_json_with_partial_decimal_stats(
1892 "decimal-min-only-high.parquet",
1893 10,
1894 Some("10.00"),
1895 None,
1896 Some(0),
1897 ),
1898 add_json_with_partial_decimal_stats(
1899 "decimal-max-only-low.parquet",
1900 10,
1901 None,
1902 Some("0.00"),
1903 Some(0),
1904 ),
1905 add_json_with_partial_decimal_stats(
1906 "decimal-counts-only.parquet",
1907 10,
1908 None,
1909 None,
1910 Some(0),
1911 ),
1912 add_json_with_partial_decimal_stats(
1913 "decimal-missing-null-count.parquet",
1914 10,
1915 Some("0.00"),
1916 Some("10.00"),
1917 None,
1918 ),
1919 add_json("decimal-missing-stats.parquet"),
1920 ],
1921 )?;
1922 let source = load_delta_source(DeltaSourceConfig {
1923 name: "orders".to_owned(),
1924 table_uri: table.path.to_string_lossy().to_string(),
1925 version: None,
1926 storage_options: Default::default(),
1927 })?;
1928
1929 Ok((table, source))
1930 }
1931
1932 fn kernel_binary_data_stats_characterization_source()
1933 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1934 let table = DeltaLogTable::new_with_metadata_and_adds(
1935 "kernel-binary-data-stats-characterization",
1936 BINARY_DATA_METADATA_JSON,
1937 &[
1938 add_json_with_binary_stats("binary-HELLO.parquet", 10, "HELLO", "HELLO", 0),
1939 add_json_with_binary_stats("binary-empty.parquet", 10, "", "", 0),
1940 add_json_with_binary_stats("binary-hello.parquet", 10, "hello", "hello", 0),
1941 add_json_with_binary_stats("binary-range.parquet", 10, "a", "z", 0),
1942 add_json_with_binary_stats("binary-special.parquet", 10, "/=%", "/=%", 0),
1943 add_json_with_binary_stats("binary-with-null.parquet", 10, "hello", "hello", 2),
1944 add_json_with_partial_binary_stats(
1945 "binary-all-null.parquet",
1946 10,
1947 None,
1948 None,
1949 Some(10),
1950 ),
1951 add_json("binary-missing-stats.parquet"),
1952 ],
1953 )?;
1954 let source = load_delta_source(DeltaSourceConfig {
1955 name: "orders".to_owned(),
1956 table_uri: table.path.to_string_lossy().to_string(),
1957 version: None,
1958 storage_options: Default::default(),
1959 })?;
1960
1961 Ok((table, source))
1962 }
1963
1964 fn kernel_partial_binary_data_stats_characterization_source()
1965 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
1966 let table = DeltaLogTable::new_with_metadata_and_adds(
1967 "kernel-partial-binary-data-stats-characterization",
1968 BINARY_DATA_METADATA_JSON,
1969 &[
1970 add_json_with_partial_binary_stats(
1971 "binary-min-only-high.parquet",
1972 10,
1973 Some("m"),
1974 None,
1975 Some(0),
1976 ),
1977 add_json_with_partial_binary_stats(
1978 "binary-max-only-low.parquet",
1979 10,
1980 None,
1981 Some("a"),
1982 Some(0),
1983 ),
1984 add_json_with_partial_binary_stats(
1985 "binary-counts-only.parquet",
1986 10,
1987 None,
1988 None,
1989 Some(0),
1990 ),
1991 add_json_with_partial_binary_stats(
1992 "binary-missing-null-count.parquet",
1993 10,
1994 Some("a"),
1995 Some("z"),
1996 None,
1997 ),
1998 add_json("binary-missing-stats.parquet"),
1999 ],
2000 )?;
2001 let source = load_delta_source(DeltaSourceConfig {
2002 name: "orders".to_owned(),
2003 table_uri: table.path.to_string_lossy().to_string(),
2004 version: None,
2005 storage_options: Default::default(),
2006 })?;
2007
2008 Ok((table, source))
2009 }
2010
2011 fn kernel_floating_data_stats_characterization_source()
2012 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2013 let table = DeltaLogTable::new_with_metadata_and_adds(
2014 "kernel-floating-data-stats-characterization",
2015 FLOATING_DATA_METADATA_JSON,
2016 &[
2017 add_json_with_floating_stats(
2018 "floating-neg.parquet",
2019 10,
2020 "-1.5",
2021 "-1.5",
2022 "-2.25",
2023 "-2.25",
2024 0,
2025 ),
2026 add_json_with_floating_stats(
2027 "floating-neg-zero.parquet",
2028 10,
2029 "-0.0",
2030 "-0.0",
2031 "-0.0",
2032 "-0.0",
2033 0,
2034 ),
2035 add_json_with_floating_stats(
2036 "floating-pos-zero.parquet",
2037 10,
2038 "0.0",
2039 "0.0",
2040 "0.0",
2041 "0.0",
2042 0,
2043 ),
2044 add_json_with_floating_stats(
2045 "floating-one.parquet",
2046 10,
2047 "1.5",
2048 "1.5",
2049 "2.25",
2050 "2.25",
2051 0,
2052 ),
2053 add_json_with_floating_stats(
2054 "floating-range.parquet",
2055 10,
2056 "-1.0",
2057 "2.0",
2058 "-2.0",
2059 "3.0",
2060 0,
2061 ),
2062 add_json_with_floating_stats(
2063 "floating-ten.parquet",
2064 10,
2065 "10.0",
2066 "10.0",
2067 "10.0",
2068 "10.0",
2069 0,
2070 ),
2071 add_json_with_floating_stats(
2072 "floating-one-with-null.parquet",
2073 10,
2074 "1.5",
2075 "1.5",
2076 "2.25",
2077 "2.25",
2078 2,
2079 ),
2080 add_json_with_partial_floating_stats(
2081 "floating-all-null.parquet",
2082 10,
2083 None,
2084 None,
2085 None,
2086 None,
2087 Some(10),
2088 ),
2089 add_json("floating-missing-stats.parquet"),
2090 ],
2091 )?;
2092 let source = load_delta_source(DeltaSourceConfig {
2093 name: "orders".to_owned(),
2094 table_uri: table.path.to_string_lossy().to_string(),
2095 version: None,
2096 storage_options: Default::default(),
2097 })?;
2098
2099 Ok((table, source))
2100 }
2101
2102 fn kernel_partial_floating_data_stats_characterization_source()
2103 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2104 let table = DeltaLogTable::new_with_metadata_and_adds(
2105 "kernel-partial-floating-data-stats-characterization",
2106 FLOATING_DATA_METADATA_JSON,
2107 &[
2108 add_json_with_partial_floating_stats(
2109 "floating-min-only-high.parquet",
2110 10,
2111 Some("2.0"),
2112 None,
2113 Some("2.0"),
2114 None,
2115 Some(0),
2116 ),
2117 add_json_with_partial_floating_stats(
2118 "floating-max-only-low.parquet",
2119 10,
2120 None,
2121 Some("0.0"),
2122 None,
2123 Some("0.0"),
2124 Some(0),
2125 ),
2126 add_json_with_partial_floating_stats(
2127 "floating-counts-only.parquet",
2128 10,
2129 None,
2130 None,
2131 None,
2132 None,
2133 Some(0),
2134 ),
2135 add_json_with_partial_floating_stats(
2136 "floating-missing-null-count.parquet",
2137 10,
2138 Some("-1.0"),
2139 Some("2.0"),
2140 Some("-1.0"),
2141 Some("2.0"),
2142 None,
2143 ),
2144 add_json("floating-missing-stats.parquet"),
2145 ],
2146 )?;
2147 let source = load_delta_source(DeltaSourceConfig {
2148 name: "orders".to_owned(),
2149 table_uri: table.path.to_string_lossy().to_string(),
2150 version: None,
2151 storage_options: Default::default(),
2152 })?;
2153
2154 Ok((table, source))
2155 }
2156
2157 fn kernel_nonfinite_floating_data_stats_characterization_source()
2158 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2159 let table = DeltaLogTable::new_with_metadata_and_adds(
2160 "kernel-nonfinite-floating-data-stats-characterization",
2161 FLOATING_DATA_METADATA_JSON,
2162 &[
2163 add_json_with_floating_stats(
2164 "floating-valid.parquet",
2165 10,
2166 "1.5",
2167 "1.5",
2168 "2.25",
2169 "2.25",
2170 0,
2171 ),
2172 add_json_with_floating_stats(
2173 "floating-nan.parquet",
2174 10,
2175 "\\\"NaN\\\"",
2176 "\\\"NaN\\\"",
2177 "\\\"NaN\\\"",
2178 "\\\"NaN\\\"",
2179 0,
2180 ),
2181 add_json_with_floating_stats(
2182 "floating-inf.parquet",
2183 10,
2184 "\\\"Infinity\\\"",
2185 "\\\"Infinity\\\"",
2186 "\\\"Infinity\\\"",
2187 "\\\"Infinity\\\"",
2188 0,
2189 ),
2190 add_json_with_floating_stats(
2191 "floating-neg-inf.parquet",
2192 10,
2193 "\\\"-Infinity\\\"",
2194 "\\\"-Infinity\\\"",
2195 "\\\"-Infinity\\\"",
2196 "\\\"-Infinity\\\"",
2197 0,
2198 ),
2199 add_json("floating-missing-stats.parquet"),
2200 ],
2201 )?;
2202 let source = load_delta_source(DeltaSourceConfig {
2203 name: "orders".to_owned(),
2204 table_uri: table.path.to_string_lossy().to_string(),
2205 version: None,
2206 storage_options: Default::default(),
2207 })?;
2208
2209 Ok((table, source))
2210 }
2211
2212 fn kernel_string_data_stats_characterization_source()
2213 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2214 let table = DeltaLogTable::new_with_metadata_and_adds(
2215 "kernel-string-data-stats-characterization",
2216 STRING_DATA_METADATA_JSON,
2217 &[
2218 add_json_with_string_stats("string-empty-only.parquet", 10, "", "", 0),
2219 add_json_with_string_stats(
2220 "string-mixed-case-only.parquet",
2221 10,
2222 "Alice",
2223 "Alice",
2224 0,
2225 ),
2226 add_json_with_string_stats("string-alice-only.parquet", 10, "alice", "alice", 0),
2227 add_json_with_string_stats("string-bob-only.parquet", 10, "bob", "bob", 0),
2228 add_json_with_string_stats("string-range.parquet", 10, "alice", "morgan", 0),
2229 add_json_with_string_stats("string-zed-only.parquet", 10, "zed", "zed", 0),
2230 add_json_with_string_stats(
2231 "string-alice-with-null.parquet",
2232 10,
2233 "alice",
2234 "alice",
2235 2,
2236 ),
2237 add_json_with_partial_string_stats(
2238 "string-all-null.parquet",
2239 10,
2240 None,
2241 None,
2242 Some(10),
2243 ),
2244 add_json("string-missing-stats.parquet"),
2245 ],
2246 )?;
2247 let source = load_delta_source(DeltaSourceConfig {
2248 name: "orders".to_owned(),
2249 table_uri: table.path.to_string_lossy().to_string(),
2250 version: None,
2251 storage_options: Default::default(),
2252 })?;
2253
2254 Ok((table, source))
2255 }
2256
2257 fn kernel_partial_string_data_stats_characterization_source()
2258 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2259 let table = DeltaLogTable::new_with_metadata_and_adds(
2260 "kernel-partial-string-data-stats-characterization",
2261 STRING_DATA_METADATA_JSON,
2262 &[
2263 add_json_with_partial_string_stats(
2264 "string-min-only-morgan.parquet",
2265 10,
2266 Some("morgan"),
2267 None,
2268 Some(0),
2269 ),
2270 add_json_with_partial_string_stats(
2271 "string-max-only-alice.parquet",
2272 10,
2273 None,
2274 Some("alice"),
2275 Some(0),
2276 ),
2277 add_json_with_partial_string_stats(
2278 "string-counts-only.parquet",
2279 10,
2280 None,
2281 None,
2282 Some(0),
2283 ),
2284 add_json_with_partial_string_stats(
2285 "string-missing-null-count.parquet",
2286 10,
2287 Some("alice"),
2288 Some("morgan"),
2289 None,
2290 ),
2291 add_json("string-missing-stats.parquet"),
2292 ],
2293 )?;
2294 let source = load_delta_source(DeltaSourceConfig {
2295 name: "orders".to_owned(),
2296 table_uri: table.path.to_string_lossy().to_string(),
2297 version: None,
2298 storage_options: Default::default(),
2299 })?;
2300
2301 Ok((table, source))
2302 }
2303
2304 fn kernel_non_ascii_string_data_stats_characterization_source()
2305 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2306 let table = DeltaLogTable::new_with_metadata_and_adds(
2307 "kernel-non-ascii-string-data-stats-characterization",
2308 STRING_DATA_METADATA_JSON,
2309 &[
2310 add_json_with_string_stats("string-ascii-cafe.parquet", 10, "cafe", "cafe", 0),
2311 add_json_with_string_stats("string-ascii-zulu.parquet", 10, "zulu", "zulu", 0),
2312 add_json_with_string_stats(
2313 "string-eclair.parquet",
2314 10,
2315 "\\u00e9clair",
2316 "\\u00e9clair",
2317 0,
2318 ),
2319 add_json_with_string_stats(
2320 "string-emile.parquet",
2321 10,
2322 "\\u00e9mile",
2323 "\\u00e9mile",
2324 0,
2325 ),
2326 add_json("string-missing-stats.parquet"),
2327 ],
2328 )?;
2329 let source = load_delta_source(DeltaSourceConfig {
2330 name: "orders".to_owned(),
2331 table_uri: table.path.to_string_lossy().to_string(),
2332 version: None,
2333 storage_options: Default::default(),
2334 })?;
2335
2336 Ok((table, source))
2337 }
2338
2339 fn kernel_timestamp_data_stats_characterization_source()
2340 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2341 let table = DeltaLogTable::new_with_metadata_and_adds(
2342 "kernel-timestamp-data-stats-characterization",
2343 TIMESTAMP_DATA_METADATA_JSON,
2344 &[
2345 add_json_with_timestamp_stats(
2346 "timestamp-pre-epoch-only.parquet",
2347 "event_ts",
2348 10,
2349 "1969-12-31T23:59:59.999999Z",
2350 "1969-12-31T23:59:59.999999Z",
2351 0,
2352 ),
2353 add_json_with_timestamp_stats(
2354 "timestamp-low-only.parquet",
2355 "event_ts",
2356 10,
2357 "2025-12-31T23:59:59.999999Z",
2358 "2025-12-31T23:59:59.999999Z",
2359 0,
2360 ),
2361 add_json_with_timestamp_stats(
2362 "timestamp-target-only.parquet",
2363 "event_ts",
2364 10,
2365 "2026-01-01T00:00:00.123456Z",
2366 "2026-01-01T00:00:00.123456Z",
2367 0,
2368 ),
2369 add_json_with_timestamp_stats(
2370 "timestamp-high-only.parquet",
2371 "event_ts",
2372 10,
2373 "2026-01-01T00:00:00.123457Z",
2374 "2026-01-01T00:00:00.123457Z",
2375 0,
2376 ),
2377 add_json_with_timestamp_stats(
2378 "timestamp-range.parquet",
2379 "event_ts",
2380 10,
2381 "2025-12-31T23:59:59.999999Z",
2382 "2026-01-01T00:00:00.123456Z",
2383 0,
2384 ),
2385 add_json_with_timestamp_stats(
2386 "timestamp-target-with-null.parquet",
2387 "event_ts",
2388 10,
2389 "2026-01-01T00:00:00.123456Z",
2390 "2026-01-01T00:00:00.123456Z",
2391 2,
2392 ),
2393 add_json_with_partial_timestamp_stats(
2394 "timestamp-all-null.parquet",
2395 "event_ts",
2396 10,
2397 None,
2398 None,
2399 Some(10),
2400 ),
2401 add_json("timestamp-missing-stats.parquet"),
2402 ],
2403 )?;
2404 let source = load_delta_source(DeltaSourceConfig {
2405 name: "orders".to_owned(),
2406 table_uri: table.path.to_string_lossy().to_string(),
2407 version: None,
2408 storage_options: Default::default(),
2409 })?;
2410
2411 Ok((table, source))
2412 }
2413
2414 fn kernel_timestamp_ntz_data_stats_characterization_source()
2415 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2416 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
2417 "kernel-timestamp-ntz-data-stats-characterization",
2418 TIMESTAMP_NTZ_PROTOCOL_JSON,
2419 TIMESTAMP_NTZ_DATA_METADATA_JSON,
2420 &[
2421 add_json_with_timestamp_stats(
2422 "timestamp-ntz-pre-epoch-only.parquet",
2423 "event_ts_ntz",
2424 10,
2425 "1969-12-31 23:59:59.999999",
2426 "1969-12-31 23:59:59.999999",
2427 0,
2428 ),
2429 add_json_with_timestamp_stats(
2430 "timestamp-ntz-low-only.parquet",
2431 "event_ts_ntz",
2432 10,
2433 "2025-12-31 23:59:59.999999",
2434 "2025-12-31 23:59:59.999999",
2435 0,
2436 ),
2437 add_json_with_timestamp_stats(
2438 "timestamp-ntz-target-only.parquet",
2439 "event_ts_ntz",
2440 10,
2441 "2026-01-01 00:00:00.123456",
2442 "2026-01-01 00:00:00.123456",
2443 0,
2444 ),
2445 add_json_with_timestamp_stats(
2446 "timestamp-ntz-high-only.parquet",
2447 "event_ts_ntz",
2448 10,
2449 "2026-01-01 00:00:00.123457",
2450 "2026-01-01 00:00:00.123457",
2451 0,
2452 ),
2453 add_json_with_timestamp_stats(
2454 "timestamp-ntz-range.parquet",
2455 "event_ts_ntz",
2456 10,
2457 "2025-12-31 23:59:59.999999",
2458 "2026-01-01 00:00:00.123456",
2459 0,
2460 ),
2461 add_json_with_timestamp_stats(
2462 "timestamp-ntz-target-with-null.parquet",
2463 "event_ts_ntz",
2464 10,
2465 "2026-01-01 00:00:00.123456",
2466 "2026-01-01 00:00:00.123456",
2467 2,
2468 ),
2469 add_json_with_partial_timestamp_stats(
2470 "timestamp-ntz-all-null.parquet",
2471 "event_ts_ntz",
2472 10,
2473 None,
2474 None,
2475 Some(10),
2476 ),
2477 add_json("timestamp-ntz-missing-stats.parquet"),
2478 ],
2479 )?;
2480 let source = load_delta_source(DeltaSourceConfig {
2481 name: "orders".to_owned(),
2482 table_uri: table.path.to_string_lossy().to_string(),
2483 version: None,
2484 storage_options: Default::default(),
2485 })?;
2486
2487 Ok((table, source))
2488 }
2489
2490 fn kernel_partial_timestamp_data_stats_characterization_source()
2491 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2492 let table = DeltaLogTable::new_with_metadata_and_adds(
2493 "kernel-partial-timestamp-data-stats-characterization",
2494 TIMESTAMP_DATA_METADATA_JSON,
2495 &[
2496 add_json_with_partial_timestamp_stats(
2497 "timestamp-min-only-target.parquet",
2498 "event_ts",
2499 10,
2500 Some("2026-01-01T00:00:00.123456Z"),
2501 None,
2502 Some(0),
2503 ),
2504 add_json_with_partial_timestamp_stats(
2505 "timestamp-max-only-low.parquet",
2506 "event_ts",
2507 10,
2508 None,
2509 Some("2025-12-31T23:59:59.999999Z"),
2510 Some(0),
2511 ),
2512 add_json_with_partial_timestamp_stats(
2513 "timestamp-counts-only.parquet",
2514 "event_ts",
2515 10,
2516 None,
2517 None,
2518 Some(0),
2519 ),
2520 add_json_with_partial_timestamp_stats(
2521 "timestamp-missing-null-count.parquet",
2522 "event_ts",
2523 10,
2524 Some("2025-12-31T23:59:59.999999Z"),
2525 Some("2026-01-01T00:00:00.123456Z"),
2526 None,
2527 ),
2528 add_json("timestamp-missing-stats.parquet"),
2529 ],
2530 )?;
2531 let source = load_delta_source(DeltaSourceConfig {
2532 name: "orders".to_owned(),
2533 table_uri: table.path.to_string_lossy().to_string(),
2534 version: None,
2535 storage_options: Default::default(),
2536 })?;
2537
2538 Ok((table, source))
2539 }
2540
2541 fn kernel_partial_timestamp_ntz_data_stats_characterization_source()
2542 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2543 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
2544 "kernel-partial-timestamp-ntz-data-stats-characterization",
2545 TIMESTAMP_NTZ_PROTOCOL_JSON,
2546 TIMESTAMP_NTZ_DATA_METADATA_JSON,
2547 &[
2548 add_json_with_partial_timestamp_stats(
2549 "timestamp-ntz-min-only-target.parquet",
2550 "event_ts_ntz",
2551 10,
2552 Some("2026-01-01 00:00:00.123456"),
2553 None,
2554 Some(0),
2555 ),
2556 add_json_with_partial_timestamp_stats(
2557 "timestamp-ntz-max-only-low.parquet",
2558 "event_ts_ntz",
2559 10,
2560 None,
2561 Some("2025-12-31 23:59:59.999999"),
2562 Some(0),
2563 ),
2564 add_json_with_partial_timestamp_stats(
2565 "timestamp-ntz-counts-only.parquet",
2566 "event_ts_ntz",
2567 10,
2568 None,
2569 None,
2570 Some(0),
2571 ),
2572 add_json_with_partial_timestamp_stats(
2573 "timestamp-ntz-missing-null-count.parquet",
2574 "event_ts_ntz",
2575 10,
2576 Some("2025-12-31 23:59:59.999999"),
2577 Some("2026-01-01 00:00:00.123456"),
2578 None,
2579 ),
2580 add_json("timestamp-ntz-missing-stats.parquet"),
2581 ],
2582 )?;
2583 let source = load_delta_source(DeltaSourceConfig {
2584 name: "orders".to_owned(),
2585 table_uri: table.path.to_string_lossy().to_string(),
2586 version: None,
2587 storage_options: Default::default(),
2588 })?;
2589
2590 Ok((table, source))
2591 }
2592
2593 fn kernel_all_sufficient_data_stats_source()
2594 -> Result<(DeltaLogTable, super::PlannedDeltaSource), Box<dyn std::error::Error>> {
2595 let table = DeltaLogTable::new_with_metadata_and_adds(
2596 "kernel-all-sufficient-data-stats",
2597 METADATA_JSON,
2598 &[
2599 add_json_with_id_stats("id-low-a.parquet", 10, 1, 50, 0),
2600 add_json_with_id_stats("id-low-b.parquet", 10, 51, 100, 0),
2601 ],
2602 )?;
2603 let source = load_delta_source(DeltaSourceConfig {
2604 name: "orders".to_owned(),
2605 table_uri: table.path.to_string_lossy().to_string(),
2606 version: None,
2607 storage_options: Default::default(),
2608 })?;
2609
2610 Ok((table, source))
2611 }
2612
2613 fn kernel_predicated_file_paths(
2614 source: &super::PlannedDeltaSource,
2615 filter: &datafusion::logical_expr::Expr,
2616 ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
2617 let predicate = datafusion_expr_to_kernel_predicate(filter)?;
2618 kernel_predicate_file_paths(source, predicate)
2619 }
2620
2621 fn kernel_predicate_file_paths(
2622 source: &super::PlannedDeltaSource,
2623 predicate: DeltaKernelPredicate,
2624 ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
2625 let scan = build_projected_predicated_delta_scan(source, None, Some(predicate))?;
2626
2627 kernel_scan_file_paths(&scan, source.table_uri())
2628 }
2629
2630 fn build_projected_predicated_stats_delta_scan(
2631 source: &super::PlannedDeltaSource,
2632 predicate: Option<DeltaKernelPredicate>,
2633 ) -> Result<ProjectedDeltaScan, delta_kernel::Error> {
2634 let (scan, kernel_schema) = super::kernel::build_projected_predicated_stats_scan(
2635 source.loaded_snapshot().kernel_snapshot(),
2636 None,
2637 predicate,
2638 )?;
2639
2640 Ok(ProjectedDeltaScan {
2641 scan,
2642 kernel_schema,
2643 })
2644 }
2645
2646 fn kernel_predicated_stats_file_paths(
2647 source: &super::PlannedDeltaSource,
2648 filter: &datafusion::logical_expr::Expr,
2649 ) -> Result<Vec<String>, Box<dyn std::error::Error>> {
2650 let predicate = datafusion_expr_to_kernel_predicate(filter)?;
2651 let scan = build_projected_predicated_stats_delta_scan(source, Some(predicate))?;
2652
2653 kernel_scan_file_paths(&scan, source.table_uri())
2654 }
2655
2656 #[derive(Debug, Default, PartialEq, Eq)]
2657 struct KernelStatsMetadataBoundary {
2658 selected_rows: usize,
2659 selected_stats_rows: usize,
2660 selected_missing_stats_rows: usize,
2661 has_id_min_values: bool,
2662 has_id_max_values: bool,
2663 has_id_null_count: bool,
2664 }
2665
2666 fn kernel_stats_metadata_boundary(
2667 scan: &ProjectedDeltaScan,
2668 table_uri: &str,
2669 ) -> Result<KernelStatsMetadataBoundary, Box<dyn std::error::Error>> {
2670 let storage_options = DeltaStorageOptions::default();
2671 let table_url = super::kernel::try_parse_uri(table_uri)?;
2672 let store = super::kernel::store_from_url_opts(
2673 &table_url,
2674 storage_options
2675 .iter()
2676 .map(|(key, value)| (key.as_str(), value.as_str())),
2677 )?;
2678 let engine = super::kernel::DefaultEngineBuilder::new(store).build();
2679 let mut boundary = KernelStatsMetadataBoundary::default();
2680
2681 for scan_metadata in scan.kernel_scan().scan_metadata(&engine)? {
2682 let (underlying_data, selection_vector) = scan_metadata?.scan_files.into_parts();
2683 let batch: RecordBatch =
2684 super::kernel::ArrowEngineData::try_from_engine_data(underlying_data)?.into();
2685 let stats_parsed = stats_parsed_column(&batch)?;
2686 let min_values = stats_struct_child(stats_parsed, "minValues")?;
2687 let max_values = stats_struct_child(stats_parsed, "maxValues")?;
2688 let null_count = stats_struct_child(stats_parsed, "nullCount")?;
2689 let min_id = stats_struct_child_column(min_values, "id")?;
2690 let max_id = stats_struct_child_column(max_values, "id")?;
2691 let null_count_id = stats_struct_child_column(null_count, "id")?;
2692
2693 boundary.has_id_min_values |= min_values.column_by_name("id").is_some();
2694 boundary.has_id_max_values |= max_values.column_by_name("id").is_some();
2695 boundary.has_id_null_count |= null_count.column_by_name("id").is_some();
2696
2697 for (row_index, selected) in selection_vector.iter().copied().enumerate() {
2698 if selected {
2699 boundary.selected_rows += 1;
2700 if !min_id.is_null(row_index)
2701 && !max_id.is_null(row_index)
2702 && !null_count_id.is_null(row_index)
2703 {
2704 boundary.selected_stats_rows += 1;
2705 } else {
2706 boundary.selected_missing_stats_rows += 1;
2707 }
2708 }
2709 }
2710 }
2711
2712 Ok(boundary)
2713 }
2714
2715 fn stats_parsed_column(
2716 batch: &RecordBatch,
2717 ) -> Result<&StructArray, Box<dyn std::error::Error>> {
2718 let Some(column) = batch.column_by_name("stats_parsed") else {
2719 return Err("scan metadata did not expose stats_parsed".into());
2720 };
2721 let Some(stats_parsed) = column.as_any().downcast_ref::<StructArray>() else {
2722 return Err("scan metadata stats_parsed was not a struct array".into());
2723 };
2724
2725 Ok(stats_parsed)
2726 }
2727
2728 fn stats_struct_child<'a>(
2729 stats_parsed: &'a StructArray,
2730 name: &str,
2731 ) -> Result<&'a StructArray, Box<dyn std::error::Error>> {
2732 let Some(column) = stats_parsed.column_by_name(name) else {
2733 return Err(format!("scan metadata stats_parsed did not expose {name}").into());
2734 };
2735 let Some(stats_child) = column.as_any().downcast_ref::<StructArray>() else {
2736 return Err(format!("scan metadata stats_parsed.{name} was not a struct array").into());
2737 };
2738
2739 Ok(stats_child)
2740 }
2741
2742 fn stats_struct_child_column<'a>(
2743 stats_child: &'a StructArray,
2744 name: &str,
2745 ) -> Result<&'a dyn Array, Box<dyn std::error::Error>> {
2746 let Some(column) = stats_child.column_by_name(name) else {
2747 return Err(format!("scan metadata stats_parsed child did not expose {name}").into());
2748 };
2749
2750 Ok(column.as_ref())
2751 }
2752
2753 fn int8_lit(value: i8) -> Expr {
2754 Expr::Literal(ScalarValue::Int8(Some(value)), None)
2755 }
2756
2757 fn int16_lit(value: i16) -> Expr {
2758 Expr::Literal(ScalarValue::Int16(Some(value)), None)
2759 }
2760
2761 fn int32_lit(value: i32) -> Expr {
2762 Expr::Literal(ScalarValue::Int32(Some(value)), None)
2763 }
2764
2765 fn int64_lit(value: i64) -> Expr {
2766 Expr::Literal(ScalarValue::Int64(Some(value)), None)
2767 }
2768
2769 fn bool_lit(value: bool) -> Expr {
2770 Expr::Literal(ScalarValue::Boolean(Some(value)), None)
2771 }
2772
2773 fn date_lit(value: i32) -> Expr {
2774 Expr::Literal(ScalarValue::Date32(Some(value)), None)
2775 }
2776
2777 fn timestamp_lit(value: i64, timezone: &str) -> Expr {
2778 Expr::Literal(
2779 ScalarValue::TimestampMicrosecond(Some(value), Some(timezone.into())),
2780 None,
2781 )
2782 }
2783
2784 fn timestamp_ntz_lit(value: i64) -> Expr {
2785 Expr::Literal(ScalarValue::TimestampMicrosecond(Some(value), None), None)
2786 }
2787
2788 fn decimal_lit(value: i128) -> Expr {
2789 Expr::Literal(ScalarValue::Decimal128(Some(value), 10, 2), None)
2790 }
2791
2792 fn decimal_lit_with_type(value: i128, precision: u8, scale: i8) -> Expr {
2793 Expr::Literal(ScalarValue::Decimal128(Some(value), precision, scale), None)
2794 }
2795
2796 fn string_lit(value: &str) -> Expr {
2797 Expr::Literal(ScalarValue::Utf8(Some(value.to_owned())), None)
2798 }
2799
2800 fn binary_lit(value: &[u8]) -> Expr {
2801 Expr::Literal(ScalarValue::Binary(Some(value.to_vec())), None)
2802 }
2803
2804 fn large_string_lit(value: &str) -> Expr {
2805 Expr::Literal(ScalarValue::LargeUtf8(Some(value.to_owned())), None)
2806 }
2807
2808 fn float32_lit(value: f32) -> Expr {
2809 Expr::Literal(ScalarValue::Float32(Some(value)), None)
2810 }
2811
2812 fn float64_lit(value: f64) -> Expr {
2813 Expr::Literal(ScalarValue::Float64(Some(value)), None)
2814 }
2815
2816 fn binary_partition_column() -> Expression {
2817 Expression::Column(ColumnName::new(["payload"]))
2818 }
2819
2820 fn binary_partition_value(value: &[u8]) -> Expression {
2821 Expression::Literal(Scalar::Binary(value.to_vec()))
2822 }
2823
2824 fn binary_partition_eq(value: &[u8]) -> DeltaKernelPredicate {
2825 DeltaKernelPredicate::new(Predicate::eq(
2826 binary_partition_column(),
2827 binary_partition_value(value),
2828 ))
2829 }
2830
2831 fn binary_partition_ne(value: &[u8]) -> DeltaKernelPredicate {
2832 DeltaKernelPredicate::new(Predicate::ne(
2833 binary_partition_column(),
2834 binary_partition_value(value),
2835 ))
2836 }
2837
2838 fn timestamp_partition_column() -> Expression {
2839 Expression::Column(ColumnName::new(["event_ts"]))
2840 }
2841
2842 fn timestamp_partition_value(value: i64) -> Expression {
2843 Expression::Literal(Scalar::Timestamp(value))
2844 }
2845
2846 fn timestamp_partition_eq(value: i64) -> DeltaKernelPredicate {
2847 DeltaKernelPredicate::new(Predicate::eq(
2848 timestamp_partition_column(),
2849 timestamp_partition_value(value),
2850 ))
2851 }
2852
2853 fn timestamp_partition_ne(value: i64) -> DeltaKernelPredicate {
2854 DeltaKernelPredicate::new(Predicate::ne(
2855 timestamp_partition_column(),
2856 timestamp_partition_value(value),
2857 ))
2858 }
2859
2860 fn timestamp_ntz_partition_column() -> Expression {
2861 Expression::Column(ColumnName::new(["event_ts_ntz"]))
2862 }
2863
2864 fn timestamp_ntz_partition_value(value: i64) -> Expression {
2865 Expression::Literal(Scalar::TimestampNtz(value))
2866 }
2867
2868 fn timestamp_ntz_partition_eq(value: i64) -> DeltaKernelPredicate {
2869 DeltaKernelPredicate::new(Predicate::eq(
2870 timestamp_ntz_partition_column(),
2871 timestamp_ntz_partition_value(value),
2872 ))
2873 }
2874
2875 fn timestamp_ntz_partition_ne(value: i64) -> DeltaKernelPredicate {
2876 DeltaKernelPredicate::new(Predicate::ne(
2877 timestamp_ntz_partition_column(),
2878 timestamp_ntz_partition_value(value),
2879 ))
2880 }
2881
2882 fn assert_invalid_integer_partition_error(
2883 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2884 ) -> Result<(), Box<dyn std::error::Error>> {
2885 let error = match result {
2886 Ok(paths) => {
2887 return Err(
2888 format!("invalid integer metadata should fail kernel scan: {paths:?}").into(),
2889 );
2890 }
2891 Err(error) => error,
2892 };
2893 let message = error.to_string();
2894 let debug_message = format!("{error:?}");
2895 assert!(message.contains("not-an-integer"));
2896 assert!(debug_message.contains("Primitive(Long)"));
2897
2898 Ok(())
2899 }
2900
2901 fn assert_invalid_boolean_partition_error(
2902 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2903 ) -> Result<(), Box<dyn std::error::Error>> {
2904 let error = match result {
2905 Ok(paths) => {
2906 return Err(
2907 format!("invalid boolean metadata should fail kernel scan: {paths:?}").into(),
2908 );
2909 }
2910 Err(error) => error,
2911 };
2912 let message = error.to_string();
2913 let debug_message = format!("{error:?}");
2914 assert!(message.contains("not-a-boolean"));
2915 assert!(debug_message.contains("Primitive(Boolean)"));
2916
2917 Ok(())
2918 }
2919
2920 fn assert_invalid_date_partition_error(
2921 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2922 ) -> Result<(), Box<dyn std::error::Error>> {
2923 let error = match result {
2924 Ok(paths) => {
2925 return Err(
2926 format!("invalid date metadata should fail kernel scan: {paths:?}").into(),
2927 );
2928 }
2929 Err(error) => error,
2930 };
2931 let message = error.to_string();
2932 let debug_message = format!("{error:?}");
2933 assert!(message.contains("not-a-date"));
2934 assert!(debug_message.contains("Primitive(Date)"));
2935
2936 Ok(())
2937 }
2938
2939 fn assert_invalid_decimal_partition_error(
2940 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2941 ) -> Result<(), Box<dyn std::error::Error>> {
2942 let error = match result {
2943 Ok(paths) => {
2944 return Err(
2945 format!("invalid decimal metadata should fail kernel scan: {paths:?}").into(),
2946 );
2947 }
2948 Err(error) => error,
2949 };
2950 let message = error.to_string();
2951 let debug_message = format!("{error:?}");
2952 assert!(
2953 message.contains("not-a-decimal")
2954 || message.contains("123.450")
2955 || message.contains("Could not parse int")
2956 || debug_message.contains("ParseIntError")
2957 || debug_message.contains("InvalidDecimal"),
2958 "{message}\n{debug_message}"
2959 );
2960 assert!(
2961 debug_message.contains("Decimal") || debug_message.contains("ParseIntError"),
2962 "{debug_message}"
2963 );
2964
2965 Ok(())
2966 }
2967
2968 fn assert_invalid_floating_partition_error(
2969 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2970 ) -> Result<(), Box<dyn std::error::Error>> {
2971 let error = match result {
2972 Ok(paths) => {
2973 return Err(format!(
2974 "invalid floating metadata should fail kernel scan: {paths:?}"
2975 )
2976 .into());
2977 }
2978 Err(error) => error,
2979 };
2980 let message = error.to_string();
2981 let debug_message = format!("{error:?}");
2982 assert!(
2983 message.contains("not-a-float")
2984 || message.contains("not-a-double")
2985 || debug_message.contains("InvalidFloat")
2986 || debug_message.contains("ParseFloatError"),
2987 "{message}\n{debug_message}"
2988 );
2989
2990 Ok(())
2991 }
2992
2993 fn assert_invalid_timestamp_partition_error(
2994 result: Result<Vec<String>, Box<dyn std::error::Error>>,
2995 ) -> Result<(), Box<dyn std::error::Error>> {
2996 let error = match result {
2997 Ok(paths) => {
2998 return Err(format!(
2999 "invalid timestamp metadata should fail kernel scan: {paths:?}"
3000 )
3001 .into());
3002 }
3003 Err(error) => error,
3004 };
3005 let message = error.to_string();
3006 let debug_message = format!("{error:?}");
3007 assert!(
3008 message.contains("not-a-timestamp")
3009 || debug_message.contains("not-a-timestamp")
3010 || debug_message.contains("Timestamp"),
3011 "{message}\n{debug_message}"
3012 );
3013
3014 Ok(())
3015 }
3016
3017 fn assert_invalid_timestamp_ntz_partition_error(
3018 result: Result<Vec<String>, Box<dyn std::error::Error>>,
3019 ) -> Result<(), Box<dyn std::error::Error>> {
3020 let error = match result {
3021 Ok(paths) => {
3022 return Err(format!(
3023 "invalid timestamp_ntz metadata should fail kernel scan: {paths:?}"
3024 )
3025 .into());
3026 }
3027 Err(error) => error,
3028 };
3029 let message = error.to_string();
3030 let debug_message = format!("{error:?}");
3031 assert!(
3032 message.contains("not-a-timestamp")
3033 || debug_message.contains("not-a-timestamp")
3034 || debug_message.contains("TimestampNtz"),
3035 "{message}\n{debug_message}"
3036 );
3037
3038 Ok(())
3039 }
3040
3041 #[test]
3042 fn kernel_partition_characterization_uses_official_delta_kernel_0_23_0() {
3043 let manifest = include_str!("../../Cargo.toml");
3044 let lockfile = include_str!("../../../../Cargo.lock");
3045
3046 assert!(manifest.contains(r#"delta_kernel = { version = "0.23.0""#));
3047 assert!(lockfile.contains(
3048 "name = \"delta_kernel\"\nversion = \"0.23.0\"\nsource = \"registry+https://github.com/rust-lang/crates.io-index\""
3049 ));
3050 assert!(!manifest.contains("deltalake"));
3051 assert!(!manifest.contains("buoyant_kernel"));
3052 }
3053
3054 #[test]
3055 fn kernel_predicated_scan_prunes_files() -> Result<(), Box<dyn std::error::Error>> {
3056 let table = DeltaLogTable::new_with_metadata_and_adds(
3057 "kernel-predicated-scan-files",
3058 PARTITIONED_METADATA_JSON,
3059 &[
3060 partitioned_add_json("region-us-west.parquet", r#"{"region":"us-west"}"#),
3061 partitioned_add_json("region-us-east.parquet", r#"{"region":"us-east"}"#),
3062 ],
3063 )?;
3064 let source = load_delta_source(DeltaSourceConfig {
3065 name: "orders".to_owned(),
3066 table_uri: table.path.to_string_lossy().to_string(),
3067 version: None,
3068 storage_options: Default::default(),
3069 })?;
3070 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
3071 let predicate = datafusion_expr_to_kernel_predicate(&col("region").eq(lit("us-west")))?;
3072 let predicated_scan =
3073 build_projected_predicated_delta_scan(&source, None, Some(predicate))?;
3074
3075 assert_eq!(
3076 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
3077 vec!["region-us-east.parquet", "region-us-west.parquet"]
3078 );
3079 assert_eq!(
3080 kernel_scan_file_paths(&predicated_scan, source.table_uri())?,
3081 vec!["region-us-west.parquet"]
3082 );
3083
3084 Ok(())
3085 }
3086
3087 #[test]
3088 fn kernel_stats_scan_metadata_exposes_parsed_file_stats_without_data_file_reads()
3089 -> Result<(), Box<dyn std::error::Error>> {
3090 let (table, source) = kernel_data_stats_characterization_source()?;
3091 let stats_scan = build_projected_predicated_stats_delta_scan(&source, None)?;
3092
3093 assert!(!table.path.join("id-impossible.parquet").exists());
3094 assert!(!table.path.join("id-possible.parquet").exists());
3095 assert!(!table.path.join("id-missing-stats.parquet").exists());
3096 assert_eq!(
3097 kernel_scan_file_paths(&stats_scan, source.table_uri())?,
3098 vec![
3099 "id-impossible.parquet",
3100 "id-missing-stats.parquet",
3101 "id-possible.parquet",
3102 ]
3103 );
3104 assert_eq!(
3105 kernel_stats_metadata_boundary(&stats_scan, source.table_uri())?,
3106 KernelStatsMetadataBoundary {
3107 selected_rows: 3,
3108 selected_stats_rows: 2,
3109 selected_missing_stats_rows: 1,
3110 has_id_min_values: true,
3111 has_id_max_values: true,
3112 has_id_null_count: true,
3113 }
3114 );
3115
3116 Ok(())
3117 }
3118
3119 #[test]
3120 fn kernel_data_column_stats_pruning_keeps_possible_and_missing_stats_files()
3121 -> Result<(), Box<dyn std::error::Error>> {
3122 let (_table, source) = kernel_data_stats_characterization_source()?;
3123
3124 assert_eq!(
3125 kernel_predicated_stats_file_paths(&source, &col("id").gt(int32_lit(100)))?,
3126 vec!["id-missing-stats.parquet", "id-possible.parquet"]
3127 );
3128
3129 Ok(())
3130 }
3131
3132 #[test]
3133 fn kernel_data_column_stats_pruning_can_return_empty_when_all_stats_are_impossible()
3134 -> Result<(), Box<dyn std::error::Error>> {
3135 let (_table, source) = kernel_all_sufficient_data_stats_source()?;
3136
3137 assert_eq!(
3138 kernel_predicated_stats_file_paths(&source, &col("id").gt(int32_lit(100)))?,
3139 Vec::<String>::new()
3140 );
3141
3142 Ok(())
3143 }
3144
3145 #[test]
3146 fn kernel_boolean_data_column_stats_pruning_documents_min_max_boundary()
3147 -> Result<(), Box<dyn std::error::Error>> {
3148 let (_table, source) = kernel_boolean_data_stats_characterization_source()?;
3149
3150 assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
3151 &source,
3152 &col("is_current").eq(bool_lit(true)),
3153 ))?;
3154 assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
3155 &source,
3156 &col("is_current").eq(bool_lit(false)),
3157 ))?;
3158 assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
3159 &source,
3160 &col("is_current").not_eq(bool_lit(true)),
3161 ))?;
3162 assert_boolean_stats_min_max_error(kernel_predicated_stats_file_paths(
3163 &source,
3164 &col("is_current").not_eq(bool_lit(false)),
3165 ))?;
3166
3167 Ok(())
3168 }
3169
3170 #[test]
3171 fn kernel_boolean_data_column_stats_pruning_documents_null_checks()
3172 -> Result<(), Box<dyn std::error::Error>> {
3173 let (_table, source) = kernel_boolean_data_stats_characterization_source()?;
3174
3175 assert_eq!(
3176 kernel_predicated_stats_file_paths(&source, &col("is_current").is_null())?,
3177 vec![
3178 "boolean-all-null.parquet",
3179 "boolean-false-with-null.parquet",
3180 "boolean-missing-stats.parquet",
3181 "boolean-true-with-null.parquet",
3182 ]
3183 );
3184 assert_eq!(
3185 kernel_predicated_stats_file_paths(&source, &col("is_current").is_not_null())?,
3186 vec![
3187 "boolean-false-only.parquet",
3188 "boolean-false-with-null.parquet",
3189 "boolean-missing-stats.parquet",
3190 "boolean-mixed.parquet",
3191 "boolean-true-only.parquet",
3192 "boolean-true-with-null.parquet",
3193 ]
3194 );
3195
3196 Ok(())
3197 }
3198
3199 #[test]
3200 fn kernel_boolean_data_column_stats_pruning_keeps_partial_null_counts_uncertain()
3201 -> Result<(), Box<dyn std::error::Error>> {
3202 let (_table, source) = kernel_partial_boolean_data_stats_characterization_source()?;
3203
3204 assert_eq!(
3205 kernel_predicated_stats_file_paths(&source, &col("is_current").is_null())?,
3206 vec![
3207 "boolean-missing-null-count.parquet",
3208 "boolean-missing-stats.parquet",
3209 ]
3210 );
3211 assert_eq!(
3212 kernel_predicated_stats_file_paths(&source, &col("is_current").is_not_null())?,
3213 vec![
3214 "boolean-counts-only.parquet",
3215 "boolean-max-only-true.parquet",
3216 "boolean-min-only-false.parquet",
3217 "boolean-missing-null-count.parquet",
3218 "boolean-missing-stats.parquet",
3219 ]
3220 );
3221
3222 Ok(())
3223 }
3224
3225 #[test]
3226 fn kernel_date_data_column_stats_pruning_documents_comparisons()
3227 -> Result<(), Box<dyn std::error::Error>> {
3228 let (_table, source) = kernel_date_data_stats_characterization_source()?;
3229
3230 assert_eq!(
3231 kernel_predicated_stats_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
3232 vec![
3233 "date-missing-stats.parquet",
3234 "date-new-year-only.parquet",
3235 "date-new-year-with-null.parquet",
3236 "date-range.parquet",
3237 ]
3238 );
3239 assert_eq!(
3240 kernel_predicated_stats_file_paths(
3241 &source,
3242 &col("event_date").gt_eq(date_lit(20_454)),
3243 )?,
3244 vec![
3245 "date-missing-stats.parquet",
3246 "date-new-year-only.parquet",
3247 "date-new-year-with-null.parquet",
3248 "date-range.parquet",
3249 ]
3250 );
3251 assert_eq!(
3252 kernel_predicated_stats_file_paths(&source, &col("event_date").lt_eq(date_lit(-1)),)?,
3253 vec!["date-missing-stats.parquet", "date-pre-epoch-only.parquet"]
3254 );
3255 assert_eq!(
3256 kernel_predicated_stats_file_paths(&source, &date_lit(20_454).gt(col("event_date")))?,
3257 vec![
3258 "date-leap-only.parquet",
3259 "date-missing-stats.parquet",
3260 "date-pre-epoch-only.parquet",
3261 "date-range.parquet",
3262 ]
3263 );
3264
3265 Ok(())
3266 }
3267
3268 #[test]
3269 fn kernel_date_data_column_stats_pruning_documents_equality_and_not_equals()
3270 -> Result<(), Box<dyn std::error::Error>> {
3271 let (_table, source) = kernel_date_data_stats_characterization_source()?;
3272
3273 assert_eq!(
3274 kernel_predicated_stats_file_paths(&source, &col("event_date").eq(date_lit(20_454)))?,
3275 vec![
3276 "date-missing-stats.parquet",
3277 "date-new-year-only.parquet",
3278 "date-new-year-with-null.parquet",
3279 "date-range.parquet",
3280 ]
3281 );
3282 assert_eq!(
3283 kernel_predicated_stats_file_paths(
3284 &source,
3285 &col("event_date").not_eq(date_lit(20_454)),
3286 )?,
3287 vec![
3288 "date-leap-only.parquet",
3289 "date-missing-stats.parquet",
3290 "date-pre-epoch-only.parquet",
3291 "date-range.parquet",
3292 ]
3293 );
3294
3295 Ok(())
3296 }
3297
3298 #[test]
3299 fn kernel_date_data_column_stats_pruning_documents_null_checks()
3300 -> Result<(), Box<dyn std::error::Error>> {
3301 let (_table, source) = kernel_date_data_stats_characterization_source()?;
3302
3303 assert_eq!(
3304 kernel_predicated_stats_file_paths(&source, &col("event_date").is_null())?,
3305 vec![
3306 "date-all-null.parquet",
3307 "date-missing-stats.parquet",
3308 "date-new-year-with-null.parquet",
3309 ]
3310 );
3311 assert_eq!(
3312 kernel_predicated_stats_file_paths(&source, &col("event_date").is_not_null())?,
3313 vec![
3314 "date-leap-only.parquet",
3315 "date-missing-stats.parquet",
3316 "date-new-year-only.parquet",
3317 "date-new-year-with-null.parquet",
3318 "date-pre-epoch-only.parquet",
3319 "date-range.parquet",
3320 ]
3321 );
3322
3323 Ok(())
3324 }
3325
3326 #[test]
3327 fn kernel_date_data_column_stats_pruning_documents_partial_stats_boundary()
3328 -> Result<(), Box<dyn std::error::Error>> {
3329 let (_table, source) = kernel_partial_date_data_stats_characterization_source()?;
3330
3331 assert_eq!(
3332 kernel_predicated_stats_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
3333 vec![
3334 "date-counts-only.parquet",
3335 "date-min-only-high.parquet",
3336 "date-missing-null-count.parquet",
3337 "date-missing-stats.parquet",
3338 ]
3339 );
3340 assert_eq!(
3341 kernel_predicated_stats_file_paths(&source, &col("event_date").is_null())?,
3342 vec![
3343 "date-missing-null-count.parquet",
3344 "date-missing-stats.parquet",
3345 ]
3346 );
3347 assert_eq!(
3348 kernel_predicated_stats_file_paths(&source, &col("event_date").is_not_null())?,
3349 vec![
3350 "date-counts-only.parquet",
3351 "date-max-only-low.parquet",
3352 "date-min-only-high.parquet",
3353 "date-missing-null-count.parquet",
3354 "date-missing-stats.parquet",
3355 ]
3356 );
3357
3358 Ok(())
3359 }
3360
3361 #[test]
3362 fn kernel_decimal_data_column_stats_pruning_documents_comparisons()
3363 -> Result<(), Box<dyn std::error::Error>> {
3364 let (_table, source) = kernel_decimal_data_stats_characterization_source()?;
3365
3366 assert_eq!(
3367 kernel_predicated_stats_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
3368 vec![
3369 "decimal-large-only.parquet",
3370 "decimal-missing-stats.parquet",
3371 "decimal-range.parquet",
3372 "decimal-ten-only.parquet",
3373 ]
3374 );
3375 assert_eq!(
3376 kernel_predicated_stats_file_paths(&source, &col("amount").gt_eq(decimal_lit(1_000)))?,
3377 vec![
3378 "decimal-large-only.parquet",
3379 "decimal-missing-stats.parquet",
3380 "decimal-range.parquet",
3381 "decimal-ten-only.parquet",
3382 ]
3383 );
3384 assert_eq!(
3385 kernel_predicated_stats_file_paths(&source, &col("amount").lt_eq(decimal_lit(-123)))?,
3386 vec![
3387 "decimal-missing-stats.parquet",
3388 "decimal-negative-only.parquet",
3389 ]
3390 );
3391 assert_eq!(
3392 kernel_predicated_stats_file_paths(&source, &decimal_lit(1_000).gt(col("amount")))?,
3393 vec![
3394 "decimal-missing-stats.parquet",
3395 "decimal-negative-only.parquet",
3396 "decimal-range.parquet",
3397 "decimal-two-only.parquet",
3398 "decimal-two-with-null.parquet",
3399 "decimal-zero-only.parquet",
3400 ]
3401 );
3402
3403 Ok(())
3404 }
3405
3406 #[test]
3407 fn kernel_decimal_data_column_stats_pruning_documents_equality_and_not_equals()
3408 -> Result<(), Box<dyn std::error::Error>> {
3409 let (_table, source) = kernel_decimal_data_stats_characterization_source()?;
3410
3411 assert_eq!(
3412 kernel_predicated_stats_file_paths(&source, &col("amount").eq(decimal_lit(200)))?,
3413 vec![
3414 "decimal-missing-stats.parquet",
3415 "decimal-range.parquet",
3416 "decimal-two-only.parquet",
3417 "decimal-two-with-null.parquet",
3418 ]
3419 );
3420 assert_eq!(
3421 kernel_predicated_stats_file_paths(&source, &col("amount").not_eq(decimal_lit(200)))?,
3422 vec![
3423 "decimal-large-only.parquet",
3424 "decimal-missing-stats.parquet",
3425 "decimal-negative-only.parquet",
3426 "decimal-range.parquet",
3427 "decimal-ten-only.parquet",
3428 "decimal-zero-only.parquet",
3429 ]
3430 );
3431
3432 Ok(())
3433 }
3434
3435 #[test]
3436 fn kernel_decimal_data_column_stats_pruning_documents_null_checks()
3437 -> Result<(), Box<dyn std::error::Error>> {
3438 let (_table, source) = kernel_decimal_data_stats_characterization_source()?;
3439
3440 assert_eq!(
3441 kernel_predicated_stats_file_paths(&source, &col("amount").is_null())?,
3442 vec![
3443 "decimal-all-null.parquet",
3444 "decimal-missing-stats.parquet",
3445 "decimal-two-with-null.parquet",
3446 ]
3447 );
3448 assert_eq!(
3449 kernel_predicated_stats_file_paths(&source, &col("amount").is_not_null())?,
3450 vec![
3451 "decimal-large-only.parquet",
3452 "decimal-missing-stats.parquet",
3453 "decimal-negative-only.parquet",
3454 "decimal-range.parquet",
3455 "decimal-ten-only.parquet",
3456 "decimal-two-only.parquet",
3457 "decimal-two-with-null.parquet",
3458 "decimal-zero-only.parquet",
3459 ]
3460 );
3461
3462 Ok(())
3463 }
3464
3465 #[test]
3466 fn kernel_decimal_data_column_stats_pruning_documents_partial_stats_boundary()
3467 -> Result<(), Box<dyn std::error::Error>> {
3468 let (_table, source) = kernel_partial_decimal_data_stats_characterization_source()?;
3469
3470 assert_eq!(
3471 kernel_predicated_stats_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
3472 vec![
3473 "decimal-counts-only.parquet",
3474 "decimal-min-only-high.parquet",
3475 "decimal-missing-null-count.parquet",
3476 "decimal-missing-stats.parquet",
3477 ]
3478 );
3479 assert_eq!(
3480 kernel_predicated_stats_file_paths(&source, &col("amount").is_null())?,
3481 vec![
3482 "decimal-missing-null-count.parquet",
3483 "decimal-missing-stats.parquet",
3484 ]
3485 );
3486 assert_eq!(
3487 kernel_predicated_stats_file_paths(&source, &col("amount").is_not_null())?,
3488 vec![
3489 "decimal-counts-only.parquet",
3490 "decimal-max-only-low.parquet",
3491 "decimal-min-only-high.parquet",
3492 "decimal-missing-null-count.parquet",
3493 "decimal-missing-stats.parquet",
3494 ]
3495 );
3496
3497 Ok(())
3498 }
3499
3500 #[test]
3501 fn kernel_decimal_data_column_stats_pruning_documents_precision_and_scale_boundary()
3502 -> Result<(), Box<dyn std::error::Error>> {
3503 let (_table, source) = kernel_decimal_data_stats_characterization_source()?;
3504
3505 for (description, literal) in [
3506 ("scale mismatch", decimal_lit_with_type(2_000, 11, 3)),
3507 ("precision mismatch", decimal_lit_with_type(200, 11, 2)),
3508 ] {
3509 let error = kernel_predicated_stats_file_paths(&source, &col("amount").eq(literal))
3510 .err()
3511 .ok_or_else(|| {
3512 format!("{description} decimal predicate should fail kernel scan")
3513 })?;
3514 let message = error.to_string();
3515 let debug_message = format!("{error:?}");
3516 assert!(
3517 message.contains("Invalid comparison operation")
3518 || debug_message.contains("Invalid comparison operation"),
3519 "{message}\n{debug_message}"
3520 );
3521 }
3522
3523 Ok(())
3524 }
3525
3526 #[test]
3527 fn kernel_binary_data_column_stats_pruning_documents_equality_and_empty_boundary()
3528 -> Result<(), Box<dyn std::error::Error>> {
3529 let (_table, source) = kernel_binary_data_stats_characterization_source()?;
3530
3531 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3532 &source,
3533 &col("payload").eq(binary_lit(b"hello")),
3534 ))?;
3535 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3536 &source,
3537 &col("payload").not_eq(binary_lit(b"hello")),
3538 ))?;
3539 assert_unsupported_literal_error(kernel_predicated_stats_file_paths(
3540 &source,
3541 &col("payload").eq(binary_lit(b"")),
3542 ))?;
3543 assert_unsupported_literal_error(kernel_predicated_stats_file_paths(
3544 &source,
3545 &col("payload").not_eq(binary_lit(b"")),
3546 ))?;
3547
3548 Ok(())
3549 }
3550
3551 #[test]
3552 fn kernel_binary_data_column_stats_pruning_documents_ordering()
3553 -> Result<(), Box<dyn std::error::Error>> {
3554 let (_table, source) = kernel_binary_data_stats_characterization_source()?;
3555
3556 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3557 &source,
3558 &col("payload").gt(binary_lit(b"hello")),
3559 ))?;
3560 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3561 &source,
3562 &col("payload").lt(binary_lit(b"hello")),
3563 ))?;
3564 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3565 &source,
3566 &binary_lit(b"hello").gt(col("payload")),
3567 ))?;
3568
3569 Ok(())
3570 }
3571
3572 #[test]
3573 fn kernel_binary_data_column_stats_pruning_documents_null_checks()
3574 -> Result<(), Box<dyn std::error::Error>> {
3575 let (_table, source) = kernel_binary_data_stats_characterization_source()?;
3576
3577 assert_eq!(
3578 kernel_predicated_stats_file_paths(&source, &col("payload").is_null())?,
3579 vec![
3580 "binary-all-null.parquet",
3581 "binary-missing-stats.parquet",
3582 "binary-with-null.parquet",
3583 ]
3584 );
3585 assert_eq!(
3586 kernel_predicated_stats_file_paths(&source, &col("payload").is_not_null())?,
3587 vec![
3588 "binary-HELLO.parquet",
3589 "binary-empty.parquet",
3590 "binary-hello.parquet",
3591 "binary-missing-stats.parquet",
3592 "binary-range.parquet",
3593 "binary-special.parquet",
3594 "binary-with-null.parquet",
3595 ]
3596 );
3597
3598 Ok(())
3599 }
3600
3601 #[test]
3602 fn kernel_binary_data_column_stats_pruning_documents_partial_stats_boundary()
3603 -> Result<(), Box<dyn std::error::Error>> {
3604 let (_table, source) = kernel_partial_binary_data_stats_characterization_source()?;
3605
3606 assert_binary_stats_min_max_error(kernel_predicated_stats_file_paths(
3607 &source,
3608 &col("payload").gt(binary_lit(b"hello")),
3609 ))?;
3610 assert_eq!(
3611 kernel_predicated_stats_file_paths(&source, &col("payload").is_null())?,
3612 vec![
3613 "binary-missing-null-count.parquet",
3614 "binary-missing-stats.parquet",
3615 ]
3616 );
3617 assert_eq!(
3618 kernel_predicated_stats_file_paths(&source, &col("payload").is_not_null())?,
3619 vec![
3620 "binary-counts-only.parquet",
3621 "binary-max-only-low.parquet",
3622 "binary-min-only-high.parquet",
3623 "binary-missing-null-count.parquet",
3624 "binary-missing-stats.parquet",
3625 ]
3626 );
3627
3628 Ok(())
3629 }
3630
3631 #[test]
3632 fn kernel_floating_data_column_stats_pruning_documents_comparisons()
3633 -> Result<(), Box<dyn std::error::Error>> {
3634 let (_table, source) = kernel_floating_data_stats_characterization_source()?;
3635
3636 assert_eq!(
3637 kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.5)))?,
3638 vec![
3639 "floating-missing-stats.parquet",
3640 "floating-range.parquet",
3641 "floating-ten.parquet",
3642 ]
3643 );
3644 assert_eq!(
3645 kernel_predicated_stats_file_paths(&source, &col("double_score").lt(float64_lit(0.0)),)?,
3646 vec![
3647 "floating-missing-stats.parquet",
3648 "floating-neg-zero.parquet",
3649 "floating-neg.parquet",
3650 "floating-range.parquet",
3651 ]
3652 );
3653 assert_eq!(
3654 kernel_predicated_stats_file_paths(&source, &float64_lit(0.0).gt(col("double_score")))?,
3655 vec![
3656 "floating-missing-stats.parquet",
3657 "floating-neg-zero.parquet",
3658 "floating-neg.parquet",
3659 "floating-range.parquet",
3660 ]
3661 );
3662 assert_eq!(
3663 kernel_predicated_stats_file_paths(&source, &col("float_score").lt(float32_lit(10.0)))?,
3664 vec![
3665 "floating-missing-stats.parquet",
3666 "floating-neg-zero.parquet",
3667 "floating-neg.parquet",
3668 "floating-one-with-null.parquet",
3669 "floating-one.parquet",
3670 "floating-pos-zero.parquet",
3671 "floating-range.parquet",
3672 ]
3673 );
3674
3675 Ok(())
3676 }
3677
3678 #[test]
3679 fn kernel_floating_data_column_stats_pruning_documents_equality_and_signed_zero()
3680 -> Result<(), Box<dyn std::error::Error>> {
3681 let (_table, source) = kernel_floating_data_stats_characterization_source()?;
3682
3683 assert_eq!(
3684 kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(1.5)))?,
3685 vec![
3686 "floating-missing-stats.parquet",
3687 "floating-one-with-null.parquet",
3688 "floating-one.parquet",
3689 "floating-range.parquet",
3690 ]
3691 );
3692 assert_eq!(
3693 kernel_predicated_stats_file_paths(
3694 &source,
3695 &col("float_score").not_eq(float32_lit(1.5)),
3696 )?,
3697 vec![
3698 "floating-missing-stats.parquet",
3699 "floating-neg-zero.parquet",
3700 "floating-neg.parquet",
3701 "floating-pos-zero.parquet",
3702 "floating-range.parquet",
3703 "floating-ten.parquet",
3704 ]
3705 );
3706 assert_eq!(
3707 kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(-0.0)))?,
3708 vec![
3709 "floating-missing-stats.parquet",
3710 "floating-neg-zero.parquet",
3711 "floating-range.parquet",
3712 ]
3713 );
3714 assert_eq!(
3715 kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(0.0)))?,
3716 vec![
3717 "floating-missing-stats.parquet",
3718 "floating-pos-zero.parquet",
3719 "floating-range.parquet",
3720 ]
3721 );
3722
3723 Ok(())
3724 }
3725
3726 #[test]
3727 fn kernel_floating_data_column_stats_pruning_documents_signed_zero_operator_boundary()
3728 -> Result<(), Box<dyn std::error::Error>> {
3729 let (_table, source) = kernel_floating_data_stats_characterization_source()?;
3730
3731 assert_eq!(
3732 kernel_predicated_stats_file_paths(&source, &col("float_score").lt(float32_lit(0.0)))?,
3733 vec![
3734 "floating-missing-stats.parquet",
3735 "floating-neg-zero.parquet",
3736 "floating-neg.parquet",
3737 "floating-range.parquet",
3738 ]
3739 );
3740 assert_eq!(
3741 kernel_predicated_stats_file_paths(
3742 &source,
3743 &col("float_score").lt_eq(float32_lit(0.0)),
3744 )?,
3745 vec![
3746 "floating-missing-stats.parquet",
3747 "floating-neg-zero.parquet",
3748 "floating-neg.parquet",
3749 "floating-pos-zero.parquet",
3750 "floating-range.parquet",
3751 ]
3752 );
3753 assert_eq!(
3754 kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(0.0)))?,
3755 vec![
3756 "floating-missing-stats.parquet",
3757 "floating-one-with-null.parquet",
3758 "floating-one.parquet",
3759 "floating-range.parquet",
3760 "floating-ten.parquet",
3761 ]
3762 );
3763 assert_eq!(
3764 kernel_predicated_stats_file_paths(
3765 &source,
3766 &col("float_score").gt_eq(float32_lit(0.0)),
3767 )?,
3768 vec![
3769 "floating-missing-stats.parquet",
3770 "floating-one-with-null.parquet",
3771 "floating-one.parquet",
3772 "floating-pos-zero.parquet",
3773 "floating-range.parquet",
3774 "floating-ten.parquet",
3775 ]
3776 );
3777 assert_eq!(
3778 kernel_predicated_stats_file_paths(
3779 &source,
3780 &col("float_score").not_eq(float32_lit(0.0)),
3781 )?,
3782 vec![
3783 "floating-missing-stats.parquet",
3784 "floating-neg-zero.parquet",
3785 "floating-neg.parquet",
3786 "floating-one-with-null.parquet",
3787 "floating-one.parquet",
3788 "floating-range.parquet",
3789 "floating-ten.parquet",
3790 ]
3791 );
3792
3793 Ok(())
3794 }
3795
3796 #[test]
3797 fn kernel_floating_data_column_stats_pruning_documents_null_checks()
3798 -> Result<(), Box<dyn std::error::Error>> {
3799 let (_table, source) = kernel_floating_data_stats_characterization_source()?;
3800
3801 assert_eq!(
3802 kernel_predicated_stats_file_paths(&source, &col("float_score").is_null())?,
3803 vec![
3804 "floating-all-null.parquet",
3805 "floating-missing-stats.parquet",
3806 "floating-one-with-null.parquet",
3807 ]
3808 );
3809 assert_eq!(
3810 kernel_predicated_stats_file_paths(&source, &col("double_score").is_not_null())?,
3811 vec![
3812 "floating-missing-stats.parquet",
3813 "floating-neg-zero.parquet",
3814 "floating-neg.parquet",
3815 "floating-one-with-null.parquet",
3816 "floating-one.parquet",
3817 "floating-pos-zero.parquet",
3818 "floating-range.parquet",
3819 "floating-ten.parquet",
3820 ]
3821 );
3822
3823 Ok(())
3824 }
3825
3826 #[test]
3827 fn kernel_floating_data_column_stats_pruning_documents_partial_stats_boundary()
3828 -> Result<(), Box<dyn std::error::Error>> {
3829 let (_table, source) = kernel_partial_floating_data_stats_characterization_source()?;
3830
3831 assert_eq!(
3832 kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.0)))?,
3833 vec![
3834 "floating-counts-only.parquet",
3835 "floating-min-only-high.parquet",
3836 "floating-missing-null-count.parquet",
3837 "floating-missing-stats.parquet",
3838 ]
3839 );
3840 assert_eq!(
3841 kernel_predicated_stats_file_paths(&source, &col("float_score").is_null())?,
3842 vec![
3843 "floating-missing-null-count.parquet",
3844 "floating-missing-stats.parquet",
3845 ]
3846 );
3847 assert_eq!(
3848 kernel_predicated_stats_file_paths(&source, &col("float_score").is_not_null())?,
3849 vec![
3850 "floating-counts-only.parquet",
3851 "floating-max-only-low.parquet",
3852 "floating-min-only-high.parquet",
3853 "floating-missing-null-count.parquet",
3854 "floating-missing-stats.parquet",
3855 ]
3856 );
3857
3858 Ok(())
3859 }
3860
3861 #[test]
3862 fn kernel_floating_data_column_stats_pruning_documents_nonfinite_stats_boundary()
3863 -> Result<(), Box<dyn std::error::Error>> {
3864 let (_table, source) = kernel_nonfinite_floating_data_stats_characterization_source()?;
3865
3866 assert_eq!(
3867 kernel_predicated_stats_file_paths(&source, &col("float_score").gt(float32_lit(1.0)))?,
3868 vec![
3869 "floating-inf.parquet",
3870 "floating-missing-stats.parquet",
3871 "floating-nan.parquet",
3872 "floating-valid.parquet",
3873 ]
3874 );
3875 assert_eq!(
3876 kernel_predicated_stats_file_paths(&source, &col("double_score").lt(float64_lit(0.0)))?,
3877 vec!["floating-missing-stats.parquet", "floating-neg-inf.parquet"]
3878 );
3879 assert_eq!(
3880 kernel_predicated_stats_file_paths(&source, &col("float_score").eq(float32_lit(1.5)),)?,
3881 vec!["floating-missing-stats.parquet", "floating-valid.parquet"]
3882 );
3883 assert_eq!(
3884 kernel_predicated_stats_file_paths(
3885 &source,
3886 &col("float_score").not_eq(float32_lit(1.5)),
3887 )?,
3888 vec![
3889 "floating-inf.parquet",
3890 "floating-missing-stats.parquet",
3891 "floating-nan.parquet",
3892 "floating-neg-inf.parquet",
3893 ]
3894 );
3895
3896 Ok(())
3897 }
3898
3899 #[test]
3900 fn kernel_string_data_column_stats_pruning_documents_comparisons()
3901 -> Result<(), Box<dyn std::error::Error>> {
3902 let (_table, source) = kernel_string_data_stats_characterization_source()?;
3903
3904 assert_eq!(
3905 kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(string_lit("m")),)?,
3906 vec![
3907 "string-missing-stats.parquet",
3908 "string-range.parquet",
3909 "string-zed-only.parquet",
3910 ]
3911 );
3912 assert_eq!(
3913 kernel_predicated_stats_file_paths(
3914 &source,
3915 &col("customer_name").gt_eq(string_lit("morgan")),
3916 )?,
3917 vec![
3918 "string-missing-stats.parquet",
3919 "string-range.parquet",
3920 "string-zed-only.parquet",
3921 ]
3922 );
3923 assert_eq!(
3924 kernel_predicated_stats_file_paths(
3925 &source,
3926 &col("customer_name").lt_eq(string_lit("Alice")),
3927 )?,
3928 vec![
3929 "string-empty-only.parquet",
3930 "string-missing-stats.parquet",
3931 "string-mixed-case-only.parquet",
3932 ]
3933 );
3934 assert_eq!(
3935 kernel_predicated_stats_file_paths(&source, &string_lit("m").gt(col("customer_name")),)?,
3936 vec![
3937 "string-alice-only.parquet",
3938 "string-alice-with-null.parquet",
3939 "string-bob-only.parquet",
3940 "string-empty-only.parquet",
3941 "string-missing-stats.parquet",
3942 "string-mixed-case-only.parquet",
3943 "string-range.parquet",
3944 ]
3945 );
3946
3947 Ok(())
3948 }
3949
3950 #[test]
3951 fn kernel_string_data_column_stats_pruning_documents_equality_and_not_equals()
3952 -> Result<(), Box<dyn std::error::Error>> {
3953 let (_table, source) = kernel_string_data_stats_characterization_source()?;
3954
3955 assert_eq!(
3956 kernel_predicated_stats_file_paths(
3957 &source,
3958 &col("customer_name").eq(string_lit("alice")),
3959 )?,
3960 vec![
3961 "string-alice-only.parquet",
3962 "string-alice-with-null.parquet",
3963 "string-missing-stats.parquet",
3964 "string-range.parquet",
3965 ]
3966 );
3967 assert_eq!(
3968 kernel_predicated_stats_file_paths(
3969 &source,
3970 &col("customer_name").not_eq(string_lit("alice")),
3971 )?,
3972 vec![
3973 "string-bob-only.parquet",
3974 "string-empty-only.parquet",
3975 "string-missing-stats.parquet",
3976 "string-mixed-case-only.parquet",
3977 "string-range.parquet",
3978 "string-zed-only.parquet",
3979 ]
3980 );
3981
3982 Ok(())
3983 }
3984
3985 #[test]
3986 fn kernel_string_data_column_stats_pruning_documents_null_checks()
3987 -> Result<(), Box<dyn std::error::Error>> {
3988 let (_table, source) = kernel_string_data_stats_characterization_source()?;
3989
3990 assert_eq!(
3991 kernel_predicated_stats_file_paths(&source, &col("customer_name").is_null())?,
3992 vec![
3993 "string-alice-with-null.parquet",
3994 "string-all-null.parquet",
3995 "string-missing-stats.parquet",
3996 ]
3997 );
3998 assert_eq!(
3999 kernel_predicated_stats_file_paths(&source, &col("customer_name").is_not_null())?,
4000 vec![
4001 "string-alice-only.parquet",
4002 "string-alice-with-null.parquet",
4003 "string-bob-only.parquet",
4004 "string-empty-only.parquet",
4005 "string-missing-stats.parquet",
4006 "string-mixed-case-only.parquet",
4007 "string-range.parquet",
4008 "string-zed-only.parquet",
4009 ]
4010 );
4011
4012 Ok(())
4013 }
4014
4015 #[test]
4016 fn kernel_string_data_column_stats_pruning_documents_partial_stats_boundary()
4017 -> Result<(), Box<dyn std::error::Error>> {
4018 let (_table, source) = kernel_partial_string_data_stats_characterization_source()?;
4019
4020 assert_eq!(
4021 kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(string_lit("m")),)?,
4022 vec![
4023 "string-counts-only.parquet",
4024 "string-min-only-morgan.parquet",
4025 "string-missing-null-count.parquet",
4026 "string-missing-stats.parquet",
4027 ]
4028 );
4029 assert_eq!(
4030 kernel_predicated_stats_file_paths(&source, &col("customer_name").is_null())?,
4031 vec![
4032 "string-missing-null-count.parquet",
4033 "string-missing-stats.parquet",
4034 ]
4035 );
4036 assert_eq!(
4037 kernel_predicated_stats_file_paths(&source, &col("customer_name").is_not_null())?,
4038 vec![
4039 "string-counts-only.parquet",
4040 "string-max-only-alice.parquet",
4041 "string-min-only-morgan.parquet",
4042 "string-missing-null-count.parquet",
4043 "string-missing-stats.parquet",
4044 ]
4045 );
4046
4047 Ok(())
4048 }
4049
4050 #[test]
4051 fn kernel_string_data_column_stats_pruning_documents_large_utf8_literal_boundary()
4052 -> Result<(), Box<dyn std::error::Error>> {
4053 let (_table, source) = kernel_string_data_stats_characterization_source()?;
4054
4055 assert_eq!(
4056 kernel_predicated_stats_file_paths(
4057 &source,
4058 &col("customer_name").eq(large_string_lit("alice")),
4059 )?,
4060 vec![
4061 "string-alice-only.parquet",
4062 "string-alice-with-null.parquet",
4063 "string-missing-stats.parquet",
4064 "string-range.parquet",
4065 ]
4066 );
4067 assert_eq!(
4068 kernel_predicated_stats_file_paths(
4069 &source,
4070 &col("customer_name").gt(large_string_lit("m")),
4071 )?,
4072 vec![
4073 "string-missing-stats.parquet",
4074 "string-range.parquet",
4075 "string-zed-only.parquet",
4076 ]
4077 );
4078
4079 Ok(())
4080 }
4081
4082 #[test]
4083 fn kernel_string_data_column_stats_pruning_documents_non_ascii_ordering()
4084 -> Result<(), Box<dyn std::error::Error>> {
4085 let (_table, source) = kernel_non_ascii_string_data_stats_characterization_source()?;
4086 let eclair = string_lit("\u{00e9}clair");
4087
4088 assert_eq!(
4089 kernel_predicated_stats_file_paths(&source, &col("customer_name").eq(eclair.clone()),)?,
4090 vec!["string-eclair.parquet", "string-missing-stats.parquet"]
4091 );
4092 assert_eq!(
4093 kernel_predicated_stats_file_paths(
4094 &source,
4095 &col("customer_name").eq(large_string_lit("\u{00e9}clair")),
4096 )?,
4097 vec!["string-eclair.parquet", "string-missing-stats.parquet"]
4098 );
4099 assert_eq!(
4100 kernel_predicated_stats_file_paths(
4101 &source,
4102 &col("customer_name").gt_eq(eclair.clone()),
4103 )?,
4104 vec![
4105 "string-eclair.parquet",
4106 "string-emile.parquet",
4107 "string-missing-stats.parquet",
4108 ]
4109 );
4110 assert_eq!(
4111 kernel_predicated_stats_file_paths(&source, &col("customer_name").lt(eclair.clone()),)?,
4112 vec![
4113 "string-ascii-cafe.parquet",
4114 "string-ascii-zulu.parquet",
4115 "string-missing-stats.parquet",
4116 ]
4117 );
4118 assert_eq!(
4119 kernel_predicated_stats_file_paths(&source, &col("customer_name").gt(eclair),)?,
4120 vec!["string-emile.parquet", "string-missing-stats.parquet"]
4121 );
4122
4123 Ok(())
4124 }
4125
4126 #[test]
4127 fn kernel_timestamp_data_column_stats_pruning_documents_comparisons()
4128 -> Result<(), Box<dyn std::error::Error>> {
4129 let (_table, source) = kernel_timestamp_data_stats_characterization_source()?;
4130 let low = 1_767_225_599_999_999_i64;
4131 let target = 1_767_225_600_123_456_i64;
4132 let high = 1_767_225_600_123_457_i64;
4133
4134 assert_eq!(
4135 kernel_predicated_stats_file_paths(
4136 &source,
4137 &col("event_ts").lt(timestamp_lit(target, "UTC")),
4138 )?,
4139 vec![
4140 "timestamp-low-only.parquet",
4141 "timestamp-missing-stats.parquet",
4142 "timestamp-pre-epoch-only.parquet",
4143 "timestamp-range.parquet",
4144 ]
4145 );
4146 assert_eq!(
4147 kernel_predicated_stats_file_paths(
4148 &source,
4149 &col("event_ts").gt_eq(timestamp_lit(target, "UTC")),
4150 )?,
4151 vec![
4152 "timestamp-high-only.parquet",
4153 "timestamp-missing-stats.parquet",
4154 "timestamp-range.parquet",
4155 "timestamp-target-only.parquet",
4156 "timestamp-target-with-null.parquet",
4157 ]
4158 );
4159 assert_eq!(
4160 kernel_predicated_stats_file_paths(
4161 &source,
4162 ×tamp_lit(high, "UTC").gt(col("event_ts")),
4163 )?,
4164 vec![
4165 "timestamp-low-only.parquet",
4166 "timestamp-missing-stats.parquet",
4167 "timestamp-pre-epoch-only.parquet",
4168 "timestamp-range.parquet",
4169 "timestamp-target-only.parquet",
4170 "timestamp-target-with-null.parquet",
4171 ]
4172 );
4173 assert_eq!(
4174 kernel_predicated_stats_file_paths(
4175 &source,
4176 &col("event_ts").eq(timestamp_lit(low, "UTC")),
4177 )?,
4178 vec![
4179 "timestamp-low-only.parquet",
4180 "timestamp-missing-stats.parquet",
4181 "timestamp-range.parquet",
4182 ]
4183 );
4184 assert_eq!(
4185 kernel_predicated_stats_file_paths(
4186 &source,
4187 &col("event_ts").not_eq(timestamp_lit(target, "UTC")),
4188 )?,
4189 vec![
4190 "timestamp-high-only.parquet",
4191 "timestamp-low-only.parquet",
4192 "timestamp-missing-stats.parquet",
4193 "timestamp-pre-epoch-only.parquet",
4194 "timestamp-range.parquet",
4195 "timestamp-target-only.parquet",
4196 "timestamp-target-with-null.parquet",
4197 ]
4198 );
4199
4200 Ok(())
4201 }
4202
4203 #[test]
4204 fn kernel_timestamp_data_column_stats_pruning_documents_null_checks()
4205 -> Result<(), Box<dyn std::error::Error>> {
4206 let (_table, source) = kernel_timestamp_data_stats_characterization_source()?;
4207
4208 assert_eq!(
4209 kernel_predicated_stats_file_paths(&source, &col("event_ts").is_null())?,
4210 vec![
4211 "timestamp-all-null.parquet",
4212 "timestamp-missing-stats.parquet",
4213 "timestamp-target-with-null.parquet",
4214 ]
4215 );
4216 assert_eq!(
4217 kernel_predicated_stats_file_paths(&source, &col("event_ts").is_not_null())?,
4218 vec![
4219 "timestamp-high-only.parquet",
4220 "timestamp-low-only.parquet",
4221 "timestamp-missing-stats.parquet",
4222 "timestamp-pre-epoch-only.parquet",
4223 "timestamp-range.parquet",
4224 "timestamp-target-only.parquet",
4225 "timestamp-target-with-null.parquet",
4226 ]
4227 );
4228
4229 Ok(())
4230 }
4231
4232 #[test]
4233 fn kernel_timestamp_data_column_stats_pruning_documents_partial_stats_boundary()
4234 -> Result<(), Box<dyn std::error::Error>> {
4235 let (_table, source) = kernel_partial_timestamp_data_stats_characterization_source()?;
4236 let low = 1_767_225_599_999_999_i64;
4237
4238 assert_eq!(
4239 kernel_predicated_stats_file_paths(
4240 &source,
4241 &col("event_ts").gt(timestamp_lit(low, "UTC")),
4242 )?,
4243 vec![
4244 "timestamp-counts-only.parquet",
4245 "timestamp-max-only-low.parquet",
4246 "timestamp-min-only-target.parquet",
4247 "timestamp-missing-null-count.parquet",
4248 "timestamp-missing-stats.parquet",
4249 ]
4250 );
4251 assert_eq!(
4252 kernel_predicated_stats_file_paths(&source, &col("event_ts").is_null())?,
4253 vec![
4254 "timestamp-missing-null-count.parquet",
4255 "timestamp-missing-stats.parquet",
4256 ]
4257 );
4258 assert_eq!(
4259 kernel_predicated_stats_file_paths(&source, &col("event_ts").is_not_null())?,
4260 vec![
4261 "timestamp-counts-only.parquet",
4262 "timestamp-max-only-low.parquet",
4263 "timestamp-min-only-target.parquet",
4264 "timestamp-missing-null-count.parquet",
4265 "timestamp-missing-stats.parquet",
4266 ]
4267 );
4268
4269 Ok(())
4270 }
4271
4272 #[test]
4273 fn kernel_timestamp_ntz_data_column_stats_pruning_documents_comparisons()
4274 -> Result<(), Box<dyn std::error::Error>> {
4275 let (_table, source) = kernel_timestamp_ntz_data_stats_characterization_source()?;
4276 let low = 1_767_225_599_999_999_i64;
4277 let target = 1_767_225_600_123_456_i64;
4278 let high = 1_767_225_600_123_457_i64;
4279
4280 assert_eq!(
4281 kernel_predicated_stats_file_paths(
4282 &source,
4283 &col("event_ts_ntz").lt(timestamp_ntz_lit(target)),
4284 )?,
4285 vec![
4286 "timestamp-ntz-low-only.parquet",
4287 "timestamp-ntz-missing-stats.parquet",
4288 "timestamp-ntz-pre-epoch-only.parquet",
4289 "timestamp-ntz-range.parquet",
4290 ]
4291 );
4292 assert_eq!(
4293 kernel_predicated_stats_file_paths(
4294 &source,
4295 &col("event_ts_ntz").gt_eq(timestamp_ntz_lit(target)),
4296 )?,
4297 vec![
4298 "timestamp-ntz-high-only.parquet",
4299 "timestamp-ntz-missing-stats.parquet",
4300 "timestamp-ntz-range.parquet",
4301 "timestamp-ntz-target-only.parquet",
4302 "timestamp-ntz-target-with-null.parquet",
4303 ]
4304 );
4305 assert_eq!(
4306 kernel_predicated_stats_file_paths(
4307 &source,
4308 ×tamp_ntz_lit(high).gt(col("event_ts_ntz")),
4309 )?,
4310 vec![
4311 "timestamp-ntz-low-only.parquet",
4312 "timestamp-ntz-missing-stats.parquet",
4313 "timestamp-ntz-pre-epoch-only.parquet",
4314 "timestamp-ntz-range.parquet",
4315 "timestamp-ntz-target-only.parquet",
4316 "timestamp-ntz-target-with-null.parquet",
4317 ]
4318 );
4319 assert_eq!(
4320 kernel_predicated_stats_file_paths(
4321 &source,
4322 &col("event_ts_ntz").eq(timestamp_ntz_lit(low)),
4323 )?,
4324 vec![
4325 "timestamp-ntz-low-only.parquet",
4326 "timestamp-ntz-missing-stats.parquet",
4327 "timestamp-ntz-range.parquet",
4328 ]
4329 );
4330 assert_eq!(
4331 kernel_predicated_stats_file_paths(
4332 &source,
4333 &col("event_ts_ntz").not_eq(timestamp_ntz_lit(target)),
4334 )?,
4335 vec![
4336 "timestamp-ntz-high-only.parquet",
4337 "timestamp-ntz-low-only.parquet",
4338 "timestamp-ntz-missing-stats.parquet",
4339 "timestamp-ntz-pre-epoch-only.parquet",
4340 "timestamp-ntz-range.parquet",
4341 "timestamp-ntz-target-only.parquet",
4342 "timestamp-ntz-target-with-null.parquet",
4343 ]
4344 );
4345
4346 Ok(())
4347 }
4348
4349 #[test]
4350 fn kernel_timestamp_ntz_data_column_stats_pruning_documents_null_checks()
4351 -> Result<(), Box<dyn std::error::Error>> {
4352 let (_table, source) = kernel_timestamp_ntz_data_stats_characterization_source()?;
4353
4354 assert_eq!(
4355 kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_null())?,
4356 vec![
4357 "timestamp-ntz-all-null.parquet",
4358 "timestamp-ntz-missing-stats.parquet",
4359 "timestamp-ntz-target-with-null.parquet",
4360 ]
4361 );
4362 assert_eq!(
4363 kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_not_null())?,
4364 vec![
4365 "timestamp-ntz-high-only.parquet",
4366 "timestamp-ntz-low-only.parquet",
4367 "timestamp-ntz-missing-stats.parquet",
4368 "timestamp-ntz-pre-epoch-only.parquet",
4369 "timestamp-ntz-range.parquet",
4370 "timestamp-ntz-target-only.parquet",
4371 "timestamp-ntz-target-with-null.parquet",
4372 ]
4373 );
4374
4375 Ok(())
4376 }
4377
4378 #[test]
4379 fn kernel_timestamp_ntz_data_column_stats_pruning_documents_partial_stats_boundary()
4380 -> Result<(), Box<dyn std::error::Error>> {
4381 let (_table, source) = kernel_partial_timestamp_ntz_data_stats_characterization_source()?;
4382 let low = 1_767_225_599_999_999_i64;
4383
4384 assert_eq!(
4385 kernel_predicated_stats_file_paths(
4386 &source,
4387 &col("event_ts_ntz").gt(timestamp_ntz_lit(low)),
4388 )?,
4389 vec![
4390 "timestamp-ntz-counts-only.parquet",
4391 "timestamp-ntz-max-only-low.parquet",
4392 "timestamp-ntz-min-only-target.parquet",
4393 "timestamp-ntz-missing-null-count.parquet",
4394 "timestamp-ntz-missing-stats.parquet",
4395 ]
4396 );
4397 assert_eq!(
4398 kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_null())?,
4399 vec![
4400 "timestamp-ntz-missing-null-count.parquet",
4401 "timestamp-ntz-missing-stats.parquet",
4402 ]
4403 );
4404 assert_eq!(
4405 kernel_predicated_stats_file_paths(&source, &col("event_ts_ntz").is_not_null())?,
4406 vec![
4407 "timestamp-ntz-counts-only.parquet",
4408 "timestamp-ntz-max-only-low.parquet",
4409 "timestamp-ntz-min-only-target.parquet",
4410 "timestamp-ntz-missing-null-count.parquet",
4411 "timestamp-ntz-missing-stats.parquet",
4412 ]
4413 );
4414
4415 Ok(())
4416 }
4417
4418 #[test]
4419 fn kernel_predicated_stats_scan_exposes_stats_for_surviving_files()
4420 -> Result<(), Box<dyn std::error::Error>> {
4421 let (_table, source) = kernel_data_stats_characterization_source()?;
4422 let predicate = datafusion_expr_to_kernel_predicate(&col("id").gt(int32_lit(100)))?;
4423 let stats_scan = build_projected_predicated_stats_delta_scan(&source, Some(predicate))?;
4424
4425 assert_eq!(
4426 kernel_scan_file_paths(&stats_scan, source.table_uri())?,
4427 vec!["id-missing-stats.parquet", "id-possible.parquet"]
4428 );
4429 assert_eq!(
4430 kernel_stats_metadata_boundary(&stats_scan, source.table_uri())?,
4431 KernelStatsMetadataBoundary {
4432 selected_rows: 2,
4433 selected_stats_rows: 1,
4434 selected_missing_stats_rows: 1,
4435 has_id_min_values: true,
4436 has_id_max_values: true,
4437 has_id_null_count: true,
4438 }
4439 );
4440
4441 Ok(())
4442 }
4443
4444 #[test]
4445 fn kernel_stats_pruning_preserves_surviving_dv_metadata_boundary()
4446 -> Result<(), Box<dyn std::error::Error>> {
4447 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
4448 "kernel-dv-data-stats-boundary",
4449 DELETION_VECTOR_PROTOCOL_JSON,
4450 METADATA_JSON,
4451 &[
4452 dv_add_json_with_id_stats("id-dv-impossible.parquet", 10, 1, 50, 0),
4453 dv_add_json_with_id_stats("id-dv-possible.parquet", 10, 101, 150, 0),
4454 add_json_with_id_stats("id-plain-impossible.parquet", 10, 1, 50, 0),
4455 add_json("id-plain-missing-stats.parquet"),
4456 partitioned_dv_add_json("id-dv-missing-stats.parquet", r#"{}"#),
4457 ],
4458 )?;
4459 let source = load_delta_source(DeltaSourceConfig {
4460 name: "orders".to_owned(),
4461 table_uri: table.path.to_string_lossy().to_string(),
4462 version: None,
4463 storage_options: Default::default(),
4464 })?;
4465 let predicate = datafusion_expr_to_kernel_predicate(&col("id").gt(int32_lit(100)))?;
4466 let stats_scan = build_projected_predicated_stats_delta_scan(&source, Some(predicate))?;
4467
4468 assert_eq!(
4469 kernel_scan_file_boundaries(&stats_scan, source.table_uri())?,
4470 vec![
4471 KernelScanFileBoundary {
4472 path: "id-dv-missing-stats.parquet".to_owned(),
4473 has_deletion_vector: true,
4474 },
4475 KernelScanFileBoundary {
4476 path: "id-dv-possible.parquet".to_owned(),
4477 has_deletion_vector: true,
4478 },
4479 KernelScanFileBoundary {
4480 path: "id-plain-missing-stats.parquet".to_owned(),
4481 has_deletion_vector: false,
4482 },
4483 ]
4484 );
4485
4486 Ok(())
4487 }
4488
4489 #[test]
4490 fn delta_table_format_production_paths_do_not_load_dv_payloads()
4491 -> Result<(), Box<dyn std::error::Error>> {
4492 let source_root = Path::new(env!("CARGO_MANIFEST_DIR"))
4493 .join("src")
4494 .join("table_formats");
4495 let production_source = [
4496 source_root.join("delta.rs"),
4497 source_root.join("delta").join("kernel.rs"),
4498 ]
4499 .iter()
4500 .map(fs::read_to_string)
4501 .collect::<Result<Vec<_>, _>>()?
4502 .into_iter()
4503 .map(|source| {
4504 source
4505 .split("\n#[cfg(test)]")
4506 .next()
4507 .unwrap_or(source.as_str())
4508 .to_owned()
4509 })
4510 .collect::<Vec<_>>()
4511 .join("\n");
4512
4513 assert!(!production_source.contains("get_selection_vector"));
4514 assert!(!production_source.contains("get_row_indexes"));
4515
4516 Ok(())
4517 }
4518
4519 #[test]
4520 fn kernel_predicated_scan_preserves_dv_metadata_boundary()
4521 -> Result<(), Box<dyn std::error::Error>> {
4522 let table = DeltaLogTable::new_with_protocol_metadata_and_adds(
4523 "kernel-dv-partition-boundary",
4524 DELETION_VECTOR_PROTOCOL_JSON,
4525 PARTITIONED_METADATA_JSON,
4526 &[
4527 partitioned_dv_add_json("region-us-west-dv.parquet", r#"{"region":"us-west"}"#),
4528 partitioned_add_json("region-us-west-plain.parquet", r#"{"region":"us-west"}"#),
4529 partitioned_dv_add_json("region-us-east-dv.parquet", r#"{"region":"us-east"}"#),
4530 partitioned_add_json("region-us-east-plain.parquet", r#"{"region":"us-east"}"#),
4531 ],
4532 )?;
4533 let source = load_delta_source(DeltaSourceConfig {
4534 name: "orders".to_owned(),
4535 table_uri: table.path.to_string_lossy().to_string(),
4536 version: None,
4537 storage_options: Default::default(),
4538 })?;
4539 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
4540 let predicate = datafusion_expr_to_kernel_predicate(&col("region").eq(lit("us-west")))?;
4541 let predicated_scan =
4542 build_projected_predicated_delta_scan(&source, None, Some(predicate))?;
4543
4544 assert_eq!(
4545 kernel_scan_file_boundaries(&unfiltered_scan, source.table_uri())?,
4546 vec![
4547 KernelScanFileBoundary {
4548 path: "region-us-east-dv.parquet".to_owned(),
4549 has_deletion_vector: true,
4550 },
4551 KernelScanFileBoundary {
4552 path: "region-us-east-plain.parquet".to_owned(),
4553 has_deletion_vector: false,
4554 },
4555 KernelScanFileBoundary {
4556 path: "region-us-west-dv.parquet".to_owned(),
4557 has_deletion_vector: true,
4558 },
4559 KernelScanFileBoundary {
4560 path: "region-us-west-plain.parquet".to_owned(),
4561 has_deletion_vector: false,
4562 },
4563 ]
4564 );
4565 assert_eq!(
4566 kernel_scan_file_boundaries(&predicated_scan, source.table_uri())?,
4567 vec![
4568 KernelScanFileBoundary {
4569 path: "region-us-west-dv.parquet".to_owned(),
4570 has_deletion_vector: true,
4571 },
4572 KernelScanFileBoundary {
4573 path: "region-us-west-plain.parquet".to_owned(),
4574 has_deletion_vector: false,
4575 },
4576 ]
4577 );
4578
4579 Ok(())
4580 }
4581
4582 #[test]
4583 fn kernel_partition_characterization_documents_string_null_and_empty_semantics()
4584 -> Result<(), Box<dyn std::error::Error>> {
4585 let (_table, source) = kernel_partition_characterization_source()?;
4586 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
4587
4588 assert_eq!(
4589 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
4590 vec![
4591 "region-empty-string.parquet",
4592 "region-missing.parquet",
4593 "region-null.parquet",
4594 "region-us-east.parquet",
4595 "region-us-west.parquet",
4596 ]
4597 );
4598 assert_eq!(
4599 kernel_predicated_file_paths(&source, &col("region").is_null())?,
4600 vec![
4601 "region-empty-string.parquet",
4602 "region-missing.parquet",
4603 "region-null.parquet",
4604 ]
4605 );
4606 assert_eq!(
4607 kernel_predicated_file_paths(&source, &col("region").is_not_null())?,
4608 vec!["region-us-east.parquet", "region-us-west.parquet"]
4609 );
4610 assert_eq!(
4611 kernel_predicated_file_paths(&source, &col("region").eq(lit("us-west")))?,
4612 vec!["region-us-west.parquet"]
4613 );
4614 assert_eq!(
4615 kernel_predicated_file_paths(&source, &col("region").not_eq(lit("us-west")))?,
4616 vec!["region-us-east.parquet"]
4617 );
4618 assert_eq!(
4619 kernel_predicated_file_paths(&source, &col("region").eq(lit("")))?,
4620 Vec::<String>::new()
4621 );
4622 assert_eq!(
4623 kernel_predicated_file_paths(&source, &col("region").not_eq(lit("")))?,
4624 vec!["region-us-east.parquet", "region-us-west.parquet"]
4625 );
4626 assert_eq!(
4627 kernel_predicated_file_paths(
4628 &source,
4629 &col("region").in_list(vec![lit("us-west"), lit("")], false),
4630 )?,
4631 vec!["region-us-west.parquet"]
4632 );
4633 assert_eq!(
4634 kernel_predicated_file_paths(
4635 &source,
4636 &col("region").in_list(vec![lit("us-west"), lit("")], true),
4637 )?,
4638 vec!["region-us-east.parquet"]
4639 );
4640
4641 Ok(())
4642 }
4643
4644 #[test]
4645 fn kernel_partition_characterization_documents_boolean_null_and_empty_semantics()
4646 -> Result<(), Box<dyn std::error::Error>> {
4647 let (_table, source) = kernel_boolean_partition_characterization_source()?;
4648 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
4649
4650 assert_eq!(
4651 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
4652 vec![
4653 "boolean-empty.parquet",
4654 "boolean-false.parquet",
4655 "boolean-missing.parquet",
4656 "boolean-null.parquet",
4657 "boolean-true.parquet",
4658 ]
4659 );
4660 assert_eq!(
4661 kernel_predicated_file_paths(&source, &col("is_current").is_null())?,
4662 vec![
4663 "boolean-empty.parquet",
4664 "boolean-missing.parquet",
4665 "boolean-null.parquet",
4666 ]
4667 );
4668 assert_eq!(
4669 kernel_predicated_file_paths(&source, &col("is_current").is_not_null())?,
4670 vec!["boolean-false.parquet", "boolean-true.parquet"]
4671 );
4672 assert_eq!(
4673 kernel_predicated_file_paths(&source, &col("is_current").eq(bool_lit(true)))?,
4674 vec!["boolean-true.parquet"]
4675 );
4676 assert_eq!(
4677 kernel_predicated_file_paths(&source, &col("is_current").eq(bool_lit(false)))?,
4678 vec!["boolean-false.parquet"]
4679 );
4680 assert_eq!(
4681 kernel_predicated_file_paths(&source, &col("is_current").not_eq(bool_lit(true)))?,
4682 vec!["boolean-false.parquet"]
4683 );
4684
4685 Ok(())
4686 }
4687
4688 #[test]
4689 fn kernel_partition_characterization_documents_boolean_membership_and_composition()
4690 -> Result<(), Box<dyn std::error::Error>> {
4691 let (_table, source) = kernel_boolean_partition_characterization_source()?;
4692
4693 assert_eq!(
4694 kernel_predicated_file_paths(
4695 &source,
4696 &col("is_current").in_list(vec![bool_lit(true), bool_lit(false)], false),
4697 )?,
4698 vec!["boolean-false.parquet", "boolean-true.parquet"]
4699 );
4700 assert_eq!(
4701 kernel_predicated_file_paths(
4702 &source,
4703 &col("is_current").in_list(vec![bool_lit(true)], true),
4704 )?,
4705 vec!["boolean-false.parquet"]
4706 );
4707 assert_eq!(
4708 kernel_predicated_file_paths(
4709 &source,
4710 &col("is_current")
4711 .eq(bool_lit(true))
4712 .or(col("is_current").is_null()),
4713 )?,
4714 vec![
4715 "boolean-empty.parquet",
4716 "boolean-missing.parquet",
4717 "boolean-null.parquet",
4718 "boolean-true.parquet",
4719 ]
4720 );
4721 assert_eq!(
4722 kernel_predicated_file_paths(
4723 &source,
4724 &col("is_current")
4725 .eq(bool_lit(true))
4726 .and(col("is_current").is_not_null()),
4727 )?,
4728 vec!["boolean-true.parquet"]
4729 );
4730 assert_eq!(
4731 kernel_predicated_file_paths(
4732 &source,
4733 &Expr::Not(Box::new(col("is_current").eq(bool_lit(true)))),
4734 )?,
4735 vec!["boolean-false.parquet"]
4736 );
4737
4738 Ok(())
4739 }
4740
4741 #[test]
4742 fn kernel_partition_characterization_documents_date_ordering()
4743 -> Result<(), Box<dyn std::error::Error>> {
4744 let (_table, source) = kernel_date_partition_characterization_source()?;
4745 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
4746
4747 assert_eq!(
4748 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
4749 vec![
4750 "date-empty.parquet",
4751 "date-epoch.parquet",
4752 "date-leap-day.parquet",
4753 "date-missing.parquet",
4754 "date-new-year.parquet",
4755 "date-null.parquet",
4756 "date-pre-epoch.parquet",
4757 ]
4758 );
4759 assert_eq!(
4760 kernel_predicated_file_paths(&source, &col("event_date").gt(date_lit(19_782)))?,
4761 vec!["date-new-year.parquet"]
4762 );
4763 assert_eq!(
4764 kernel_predicated_file_paths(&source, &col("event_date").lt_eq(date_lit(0)))?,
4765 vec!["date-epoch.parquet", "date-pre-epoch.parquet"]
4766 );
4767 assert_eq!(
4768 kernel_predicated_file_paths(&source, &date_lit(20_454).gt(col("event_date")))?,
4769 vec![
4770 "date-epoch.parquet",
4771 "date-leap-day.parquet",
4772 "date-pre-epoch.parquet",
4773 ]
4774 );
4775
4776 Ok(())
4777 }
4778
4779 #[test]
4780 fn kernel_partition_characterization_documents_date_null_and_empty_semantics()
4781 -> Result<(), Box<dyn std::error::Error>> {
4782 let (_table, source) = kernel_date_partition_characterization_source()?;
4783
4784 assert_eq!(
4785 kernel_predicated_file_paths(&source, &col("event_date").is_null())?,
4786 vec![
4787 "date-empty.parquet",
4788 "date-missing.parquet",
4789 "date-null.parquet"
4790 ]
4791 );
4792 assert_eq!(
4793 kernel_predicated_file_paths(&source, &col("event_date").is_not_null())?,
4794 vec![
4795 "date-epoch.parquet",
4796 "date-leap-day.parquet",
4797 "date-new-year.parquet",
4798 "date-pre-epoch.parquet",
4799 ]
4800 );
4801 assert_eq!(
4802 kernel_predicated_file_paths(&source, &col("event_date").eq(date_lit(20_454)))?,
4803 vec!["date-new-year.parquet"]
4804 );
4805 assert_eq!(
4806 kernel_predicated_file_paths(&source, &col("event_date").not_eq(date_lit(20_454)))?,
4807 vec![
4808 "date-epoch.parquet",
4809 "date-leap-day.parquet",
4810 "date-pre-epoch.parquet",
4811 ]
4812 );
4813
4814 Ok(())
4815 }
4816
4817 #[test]
4818 fn kernel_partition_characterization_documents_date_membership_between_and_composition()
4819 -> Result<(), Box<dyn std::error::Error>> {
4820 let (_table, source) = kernel_date_partition_characterization_source()?;
4821
4822 assert_eq!(
4823 kernel_predicated_file_paths(
4824 &source,
4825 &col("event_date").in_list(vec![date_lit(-1), date_lit(20_454)], false),
4826 )?,
4827 vec!["date-new-year.parquet", "date-pre-epoch.parquet"]
4828 );
4829 assert_eq!(
4830 kernel_predicated_file_paths(
4831 &source,
4832 &col("event_date").in_list(vec![date_lit(19_782)], true),
4833 )?,
4834 vec![
4835 "date-epoch.parquet",
4836 "date-new-year.parquet",
4837 "date-pre-epoch.parquet",
4838 ]
4839 );
4840 assert_eq!(
4841 kernel_predicated_file_paths(
4842 &source,
4843 &col("event_date").between(date_lit(-1), date_lit(19_782)),
4844 )?,
4845 vec![
4846 "date-epoch.parquet",
4847 "date-leap-day.parquet",
4848 "date-pre-epoch.parquet",
4849 ]
4850 );
4851 assert_eq!(
4852 kernel_predicated_file_paths(
4853 &source,
4854 &col("event_date").not_between(date_lit(-1), date_lit(19_782)),
4855 )?,
4856 vec!["date-new-year.parquet"]
4857 );
4858 assert_eq!(
4859 kernel_predicated_file_paths(
4860 &source,
4861 &col("event_date")
4862 .gt_eq(date_lit(0))
4863 .and(col("event_date").lt(date_lit(20_454))),
4864 )?,
4865 vec!["date-epoch.parquet", "date-leap-day.parquet"]
4866 );
4867 assert_eq!(
4868 kernel_predicated_file_paths(
4869 &source,
4870 &col("event_date")
4871 .eq(date_lit(-1))
4872 .or(col("event_date").eq(date_lit(20_454))),
4873 )?,
4874 vec!["date-new-year.parquet", "date-pre-epoch.parquet"]
4875 );
4876 assert_eq!(
4877 kernel_predicated_file_paths(
4878 &source,
4879 &Expr::Not(Box::new(col("event_date").eq(date_lit(19_782)))),
4880 )?,
4881 vec![
4882 "date-epoch.parquet",
4883 "date-new-year.parquet",
4884 "date-pre-epoch.parquet",
4885 ]
4886 );
4887
4888 Ok(())
4889 }
4890
4891 #[test]
4892 fn kernel_partition_characterization_documents_decimal_numeric_ordering()
4893 -> Result<(), Box<dyn std::error::Error>> {
4894 let (_table, source) = kernel_decimal_partition_characterization_source()?;
4895 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
4896
4897 assert_eq!(
4898 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
4899 vec![
4900 "decimal-empty.parquet",
4901 "decimal-large.parquet",
4902 "decimal-missing.parquet",
4903 "decimal-negative.parquet",
4904 "decimal-null.parquet",
4905 "decimal-ten.parquet",
4906 "decimal-two.parquet",
4907 "decimal-zero.parquet",
4908 ]
4909 );
4910 assert_eq!(
4911 kernel_predicated_file_paths(&source, &col("amount").gt(decimal_lit(200)))?,
4912 vec!["decimal-large.parquet", "decimal-ten.parquet"]
4913 );
4914 assert_eq!(
4915 kernel_predicated_file_paths(&source, &col("amount").lt(decimal_lit(1000)))?,
4916 vec![
4917 "decimal-negative.parquet",
4918 "decimal-two.parquet",
4919 "decimal-zero.parquet",
4920 ]
4921 );
4922 assert_eq!(
4923 kernel_predicated_file_paths(&source, &decimal_lit(1000).gt(col("amount")))?,
4924 vec![
4925 "decimal-negative.parquet",
4926 "decimal-two.parquet",
4927 "decimal-zero.parquet",
4928 ]
4929 );
4930
4931 Ok(())
4932 }
4933
4934 #[test]
4935 fn kernel_partition_characterization_documents_decimal_null_and_empty_semantics()
4936 -> Result<(), Box<dyn std::error::Error>> {
4937 let (_table, source) = kernel_decimal_partition_characterization_source()?;
4938
4939 assert_eq!(
4940 kernel_predicated_file_paths(&source, &col("amount").is_null())?,
4941 vec![
4942 "decimal-empty.parquet",
4943 "decimal-missing.parquet",
4944 "decimal-null.parquet",
4945 ]
4946 );
4947 assert_eq!(
4948 kernel_predicated_file_paths(&source, &col("amount").is_not_null())?,
4949 vec![
4950 "decimal-large.parquet",
4951 "decimal-negative.parquet",
4952 "decimal-ten.parquet",
4953 "decimal-two.parquet",
4954 "decimal-zero.parquet",
4955 ]
4956 );
4957 assert_eq!(
4958 kernel_predicated_file_paths(&source, &col("amount").eq(decimal_lit(12_345)))?,
4959 vec!["decimal-large.parquet"]
4960 );
4961 assert_eq!(
4962 kernel_predicated_file_paths(&source, &col("amount").not_eq(decimal_lit(12_345)))?,
4963 vec![
4964 "decimal-negative.parquet",
4965 "decimal-ten.parquet",
4966 "decimal-two.parquet",
4967 "decimal-zero.parquet",
4968 ]
4969 );
4970
4971 Ok(())
4972 }
4973
4974 #[test]
4975 fn kernel_partition_characterization_documents_decimal_membership_between_and_composition()
4976 -> Result<(), Box<dyn std::error::Error>> {
4977 let (_table, source) = kernel_decimal_partition_characterization_source()?;
4978
4979 assert_eq!(
4980 kernel_predicated_file_paths(
4981 &source,
4982 &col("amount").in_list(vec![decimal_lit(-123), decimal_lit(12_345)], false),
4983 )?,
4984 vec!["decimal-large.parquet", "decimal-negative.parquet"]
4985 );
4986 assert_eq!(
4987 kernel_predicated_file_paths(
4988 &source,
4989 &col("amount").in_list(vec![decimal_lit(200)], true),
4990 )?,
4991 vec![
4992 "decimal-large.parquet",
4993 "decimal-negative.parquet",
4994 "decimal-ten.parquet",
4995 "decimal-zero.parquet",
4996 ]
4997 );
4998 assert_eq!(
4999 kernel_predicated_file_paths(
5000 &source,
5001 &col("amount").between(decimal_lit(-123), decimal_lit(200)),
5002 )?,
5003 vec![
5004 "decimal-negative.parquet",
5005 "decimal-two.parquet",
5006 "decimal-zero.parquet",
5007 ]
5008 );
5009 assert_eq!(
5010 kernel_predicated_file_paths(
5011 &source,
5012 &col("amount").not_between(decimal_lit(-123), decimal_lit(200)),
5013 )?,
5014 vec!["decimal-large.parquet", "decimal-ten.parquet"]
5015 );
5016 assert_eq!(
5017 kernel_predicated_file_paths(
5018 &source,
5019 &col("amount")
5020 .gt_eq(decimal_lit(0))
5021 .and(col("amount").lt(decimal_lit(12345))),
5022 )?,
5023 vec![
5024 "decimal-ten.parquet",
5025 "decimal-two.parquet",
5026 "decimal-zero.parquet",
5027 ]
5028 );
5029 assert_eq!(
5030 kernel_predicated_file_paths(
5031 &source,
5032 &col("amount")
5033 .eq(decimal_lit(-123))
5034 .or(col("amount").eq(decimal_lit(12_345))),
5035 )?,
5036 vec!["decimal-large.parquet", "decimal-negative.parquet"]
5037 );
5038 assert_eq!(
5039 kernel_predicated_file_paths(
5040 &source,
5041 &Expr::Not(Box::new(col("amount").eq(decimal_lit(200)))),
5042 )?,
5043 vec![
5044 "decimal-large.parquet",
5045 "decimal-negative.parquet",
5046 "decimal-ten.parquet",
5047 "decimal-zero.parquet",
5048 ]
5049 );
5050
5051 Ok(())
5052 }
5053
5054 #[test]
5055 fn kernel_partition_characterization_documents_invalid_decimal_metadata()
5056 -> Result<(), Box<dyn std::error::Error>> {
5057 let invalid_text_table = DeltaLogTable::new_with_metadata_and_adds(
5058 "kernel-invalid-decimal-characterization",
5059 DECIMAL_PARTITIONED_METADATA_JSON,
5060 &[
5061 partitioned_add_json("decimal-valid.parquet", r#"{"amount":"123.45"}"#),
5062 partitioned_add_json("decimal-invalid.parquet", r#"{"amount":"not-a-decimal"}"#),
5063 ],
5064 )?;
5065 let invalid_text_source = load_delta_source(DeltaSourceConfig {
5066 name: "orders".to_owned(),
5067 table_uri: invalid_text_table.path.to_string_lossy().to_string(),
5068 version: None,
5069 storage_options: Default::default(),
5070 })?;
5071 let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;
5072
5073 assert_invalid_decimal_partition_error(kernel_scan_file_paths(
5074 &unfiltered_scan,
5075 invalid_text_source.table_uri(),
5076 ))?;
5077 assert_invalid_decimal_partition_error(kernel_predicated_file_paths(
5078 &invalid_text_source,
5079 &col("amount").eq(decimal_lit(12_345)),
5080 ))?;
5081
5082 let scale_mismatch_table = DeltaLogTable::new_with_metadata_and_adds(
5083 "kernel-invalid-decimal-scale-characterization",
5084 DECIMAL_PARTITIONED_METADATA_JSON,
5085 &[
5086 partitioned_add_json("decimal-valid.parquet", r#"{"amount":"123.45"}"#),
5087 partitioned_add_json("decimal-scale.parquet", r#"{"amount":"123.450"}"#),
5088 ],
5089 )?;
5090 let scale_mismatch_source = load_delta_source(DeltaSourceConfig {
5091 name: "orders".to_owned(),
5092 table_uri: scale_mismatch_table.path.to_string_lossy().to_string(),
5093 version: None,
5094 storage_options: Default::default(),
5095 })?;
5096
5097 assert_invalid_decimal_partition_error(kernel_predicated_file_paths(
5098 &scale_mismatch_source,
5099 &col("amount").eq(decimal_lit(12_345)),
5100 ))?;
5101
5102 Ok(())
5103 }
5104
5105 #[test]
5106 fn kernel_partition_characterization_documents_floating_numeric_ordering()
5107 -> Result<(), Box<dyn std::error::Error>> {
5108 let (_table, source) = kernel_floating_partition_characterization_source()?;
5109 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
5110
5111 assert_eq!(
5112 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
5113 vec![
5114 "floating-empty.parquet",
5115 "floating-missing.parquet",
5116 "floating-neg-zero.parquet",
5117 "floating-neg.parquet",
5118 "floating-null.parquet",
5119 "floating-one.parquet",
5120 "floating-pos-zero.parquet",
5121 "floating-ten.parquet",
5122 ]
5123 );
5124 assert_eq!(
5125 kernel_predicated_file_paths(&source, &col("float_part").gt(float32_lit(1.5)))?,
5126 vec!["floating-ten.parquet"]
5127 );
5128 assert_eq!(
5129 kernel_predicated_file_paths(&source, &col("double_part").lt(float64_lit(0.0)))?,
5130 vec!["floating-neg-zero.parquet", "floating-neg.parquet"]
5131 );
5132 assert_eq!(
5133 kernel_predicated_file_paths(&source, &float64_lit(0.0).gt(col("double_part")))?,
5134 vec!["floating-neg-zero.parquet", "floating-neg.parquet"]
5135 );
5136 assert_eq!(
5137 kernel_predicated_file_paths(&source, &col("float_part").lt(float32_lit(10.0)))?,
5138 vec![
5139 "floating-neg-zero.parquet",
5140 "floating-neg.parquet",
5141 "floating-one.parquet",
5142 "floating-pos-zero.parquet",
5143 ]
5144 );
5145
5146 Ok(())
5147 }
5148
5149 #[test]
5150 fn kernel_partition_characterization_documents_floating_null_and_zero_semantics()
5151 -> Result<(), Box<dyn std::error::Error>> {
5152 let (_table, source) = kernel_floating_partition_characterization_source()?;
5153
5154 assert_eq!(
5155 kernel_predicated_file_paths(&source, &col("float_part").is_null())?,
5156 vec![
5157 "floating-empty.parquet",
5158 "floating-missing.parquet",
5159 "floating-null.parquet",
5160 ]
5161 );
5162 assert_eq!(
5163 kernel_predicated_file_paths(&source, &col("double_part").is_not_null())?,
5164 vec![
5165 "floating-neg-zero.parquet",
5166 "floating-neg.parquet",
5167 "floating-one.parquet",
5168 "floating-pos-zero.parquet",
5169 "floating-ten.parquet",
5170 ]
5171 );
5172 assert_eq!(
5173 kernel_predicated_file_paths(&source, &col("float_part").eq(float32_lit(-0.0)))?,
5174 vec!["floating-neg-zero.parquet"]
5175 );
5176 assert_eq!(
5177 kernel_predicated_file_paths(&source, &col("float_part").eq(float32_lit(0.0)))?,
5178 vec!["floating-pos-zero.parquet"]
5179 );
5180 assert_eq!(
5181 kernel_predicated_file_paths(&source, &col("double_part").not_eq(float64_lit(0.0)))?,
5182 vec![
5183 "floating-neg-zero.parquet",
5184 "floating-neg.parquet",
5185 "floating-one.parquet",
5186 "floating-ten.parquet",
5187 ]
5188 );
5189
5190 Ok(())
5191 }
5192
5193 #[test]
5194 fn kernel_partition_characterization_documents_floating_membership_between_and_composition()
5195 -> Result<(), Box<dyn std::error::Error>> {
5196 let (_table, source) = kernel_floating_partition_characterization_source()?;
5197
5198 assert_eq!(
5199 kernel_predicated_file_paths(
5200 &source,
5201 &col("float_part").in_list(vec![float32_lit(-1.5), float32_lit(1.5)], false),
5202 )?,
5203 vec!["floating-neg.parquet", "floating-one.parquet"]
5204 );
5205 assert_eq!(
5206 kernel_predicated_file_paths(
5207 &source,
5208 &col("double_part").in_list(vec![float64_lit(2.25)], true),
5209 )?,
5210 vec![
5211 "floating-neg-zero.parquet",
5212 "floating-neg.parquet",
5213 "floating-pos-zero.parquet",
5214 "floating-ten.parquet",
5215 ]
5216 );
5217 assert_eq!(
5218 kernel_predicated_file_paths(
5219 &source,
5220 &col("float_part").between(float32_lit(-0.0), float32_lit(1.5)),
5221 )?,
5222 vec![
5223 "floating-neg-zero.parquet",
5224 "floating-one.parquet",
5225 "floating-pos-zero.parquet",
5226 ]
5227 );
5228 assert_eq!(
5229 kernel_predicated_file_paths(
5230 &source,
5231 &col("double_part").not_between(float64_lit(0.0), float64_lit(2.25)),
5232 )?,
5233 vec![
5234 "floating-neg-zero.parquet",
5235 "floating-neg.parquet",
5236 "floating-ten.parquet",
5237 ]
5238 );
5239 assert_eq!(
5240 kernel_predicated_file_paths(
5241 &source,
5242 &col("float_part")
5243 .gt_eq(float32_lit(0.0))
5244 .and(col("float_part").lt(float32_lit(10.0))),
5245 )?,
5246 vec!["floating-one.parquet", "floating-pos-zero.parquet"]
5247 );
5248 assert_eq!(
5249 kernel_predicated_file_paths(
5250 &source,
5251 &col("double_part")
5252 .eq(float64_lit(-2.25))
5253 .or(col("double_part").eq(float64_lit(10.0))),
5254 )?,
5255 vec!["floating-neg.parquet", "floating-ten.parquet"]
5256 );
5257 assert_eq!(
5258 kernel_predicated_file_paths(
5259 &source,
5260 &Expr::Not(Box::new(col("float_part").eq(float32_lit(1.5)))),
5261 )?,
5262 vec![
5263 "floating-neg-zero.parquet",
5264 "floating-neg.parquet",
5265 "floating-pos-zero.parquet",
5266 "floating-ten.parquet",
5267 ]
5268 );
5269
5270 Ok(())
5271 }
5272
5273 #[test]
5274 fn kernel_partition_characterization_documents_binary_null_empty_and_equality_semantics()
5275 -> Result<(), Box<dyn std::error::Error>> {
5276 let (_table, source) = kernel_binary_partition_characterization_source()?;
5277 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
5278
5279 assert_eq!(
5280 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
5281 vec![
5282 "binary-HELLO.parquet",
5283 "binary-empty.parquet",
5284 "binary-hello.parquet",
5285 "binary-missing.parquet",
5286 "binary-null.parquet",
5287 "binary-special.parquet",
5288 ]
5289 );
5290 assert_eq!(
5291 kernel_predicate_file_paths(
5292 &source,
5293 DeltaKernelPredicate::new(Predicate::is_null(binary_partition_column())),
5294 )?,
5295 vec![
5296 "binary-empty.parquet",
5297 "binary-missing.parquet",
5298 "binary-null.parquet",
5299 ]
5300 );
5301 assert_eq!(
5302 kernel_predicate_file_paths(
5303 &source,
5304 DeltaKernelPredicate::new(Predicate::is_not_null(binary_partition_column())),
5305 )?,
5306 vec![
5307 "binary-HELLO.parquet",
5308 "binary-hello.parquet",
5309 "binary-special.parquet",
5310 ]
5311 );
5312 assert_eq!(
5313 kernel_predicate_file_paths(&source, binary_partition_eq(b"HELLO"))?,
5314 vec!["binary-HELLO.parquet"]
5315 );
5316 assert_eq!(
5317 kernel_predicate_file_paths(&source, binary_partition_eq(b"hello"))?,
5318 vec!["binary-hello.parquet"]
5319 );
5320 assert_eq!(
5321 kernel_predicate_file_paths(&source, binary_partition_ne(b"hello"))?,
5322 vec!["binary-HELLO.parquet", "binary-special.parquet"]
5323 );
5324 assert_eq!(
5325 kernel_predicate_file_paths(&source, binary_partition_eq(&[]))?,
5326 Vec::<String>::new()
5327 );
5328 assert_eq!(
5329 kernel_predicate_file_paths(&source, binary_partition_ne(&[]))?,
5330 vec![
5331 "binary-HELLO.parquet",
5332 "binary-hello.parquet",
5333 "binary-special.parquet",
5334 ]
5335 );
5336
5337 Ok(())
5338 }
5339
5340 #[test]
5341 fn kernel_partition_characterization_documents_binary_membership_and_composition()
5342 -> Result<(), Box<dyn std::error::Error>> {
5343 let (_table, source) = kernel_binary_partition_characterization_source()?;
5344
5345 assert_eq!(
5346 kernel_predicate_file_paths(
5347 &source,
5348 DeltaKernelPredicate::new(Predicate::or(
5349 Predicate::eq(binary_partition_column(), binary_partition_value(b"HELLO")),
5350 Predicate::eq(binary_partition_column(), binary_partition_value(b"/=%")),
5351 )),
5352 )?,
5353 vec!["binary-HELLO.parquet", "binary-special.parquet"]
5354 );
5355 assert_eq!(
5356 kernel_predicate_file_paths(
5357 &source,
5358 DeltaKernelPredicate::new(Predicate::and(
5359 Predicate::ne(binary_partition_column(), binary_partition_value(b"hello")),
5360 Predicate::ne(binary_partition_column(), binary_partition_value(b"/=%")),
5361 )),
5362 )?,
5363 vec!["binary-HELLO.parquet"]
5364 );
5365 assert_eq!(
5366 kernel_predicate_file_paths(
5367 &source,
5368 DeltaKernelPredicate::new(Predicate::and(
5369 Predicate::is_not_null(binary_partition_column()),
5370 Predicate::ne(binary_partition_column(), binary_partition_value(b"HELLO")),
5371 )),
5372 )?,
5373 vec!["binary-hello.parquet", "binary-special.parquet"]
5374 );
5375 assert_eq!(
5376 kernel_predicate_file_paths(
5377 &source,
5378 DeltaKernelPredicate::new(Predicate::or(
5379 Predicate::eq(binary_partition_column(), binary_partition_value(b"HELLO")),
5380 Predicate::is_null(binary_partition_column()),
5381 )),
5382 )?,
5383 vec![
5384 "binary-HELLO.parquet",
5385 "binary-empty.parquet",
5386 "binary-missing.parquet",
5387 "binary-null.parquet",
5388 ]
5389 );
5390 assert_eq!(
5391 kernel_predicate_file_paths(
5392 &source,
5393 DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
5394 binary_partition_column(),
5395 binary_partition_value(b"hello"),
5396 ))),
5397 )?,
5398 vec!["binary-HELLO.parquet", "binary-special.parquet"]
5399 );
5400
5401 Ok(())
5402 }
5403
5404 #[test]
5405 fn kernel_partition_characterization_documents_non_utf8_binary_literals_do_not_match()
5406 -> Result<(), Box<dyn std::error::Error>> {
5407 let (_table, source) = kernel_binary_partition_characterization_source()?;
5408
5409 assert_eq!(
5410 kernel_predicate_file_paths(&source, binary_partition_eq(&[0xDE, 0xAD, 0xBE, 0xEF]))?,
5411 Vec::<String>::new()
5412 );
5413 assert_eq!(
5414 kernel_predicate_file_paths(&source, binary_partition_ne(&[0xDE, 0xAD, 0xBE, 0xEF]))?,
5415 vec![
5416 "binary-HELLO.parquet",
5417 "binary-hello.parquet",
5418 "binary-special.parquet",
5419 ]
5420 );
5421
5422 Ok(())
5423 }
5424
5425 #[test]
5426 fn kernel_partition_characterization_documents_timestamp_ordering_and_precision()
5427 -> Result<(), Box<dyn std::error::Error>> {
5428 let (_table, source) = kernel_timestamp_partition_characterization_source()?;
5429 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
5430 let pre_epoch = -1_i64;
5431 let low = 1_767_225_599_999_999_i64;
5432 let target = 1_767_225_600_123_456_i64;
5433 let high = 1_767_225_600_123_457_i64;
5434
5435 assert_eq!(
5436 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
5437 vec![
5438 "timestamp-empty.parquet",
5439 "timestamp-high.parquet",
5440 "timestamp-low.parquet",
5441 "timestamp-missing.parquet",
5442 "timestamp-null.parquet",
5443 "timestamp-pre-epoch.parquet",
5444 "timestamp-target.parquet",
5445 ]
5446 );
5447 assert_eq!(
5448 kernel_predicate_file_paths(
5449 &source,
5450 DeltaKernelPredicate::new(Predicate::lt(
5451 timestamp_partition_column(),
5452 timestamp_partition_value(target),
5453 )),
5454 )?,
5455 vec!["timestamp-low.parquet", "timestamp-pre-epoch.parquet"]
5456 );
5457 assert_eq!(
5458 kernel_predicate_file_paths(
5459 &source,
5460 DeltaKernelPredicate::new(Predicate::ge(
5461 timestamp_partition_column(),
5462 timestamp_partition_value(target),
5463 )),
5464 )?,
5465 vec!["timestamp-high.parquet", "timestamp-target.parquet"]
5466 );
5467 assert_eq!(
5468 kernel_predicate_file_paths(
5469 &source,
5470 DeltaKernelPredicate::new(Predicate::gt(
5471 timestamp_partition_value(high),
5472 timestamp_partition_column(),
5473 )),
5474 )?,
5475 vec![
5476 "timestamp-low.parquet",
5477 "timestamp-pre-epoch.parquet",
5478 "timestamp-target.parquet",
5479 ]
5480 );
5481 assert_eq!(
5482 kernel_predicate_file_paths(&source, timestamp_partition_eq(pre_epoch))?,
5483 vec!["timestamp-pre-epoch.parquet"]
5484 );
5485 assert_eq!(
5486 kernel_predicate_file_paths(&source, timestamp_partition_eq(target))?,
5487 vec!["timestamp-target.parquet"]
5488 );
5489 assert_eq!(
5490 kernel_predicate_file_paths(&source, timestamp_partition_ne(target))?,
5491 vec![
5492 "timestamp-high.parquet",
5493 "timestamp-low.parquet",
5494 "timestamp-pre-epoch.parquet",
5495 ]
5496 );
5497 assert_eq!(
5498 kernel_predicate_file_paths(&source, timestamp_partition_eq(target + 1))?,
5499 vec!["timestamp-high.parquet"]
5500 );
5501 assert_eq!(
5502 kernel_predicate_file_paths(&source, timestamp_partition_eq(low))?,
5503 vec!["timestamp-low.parquet"]
5504 );
5505
5506 Ok(())
5507 }
5508
5509 #[test]
5510 fn kernel_partition_characterization_documents_timestamp_null_empty_and_membership()
5511 -> Result<(), Box<dyn std::error::Error>> {
5512 let (_table, source) = kernel_timestamp_partition_characterization_source()?;
5513 let low = 1_767_225_599_999_999_i64;
5514 let target = 1_767_225_600_123_456_i64;
5515
5516 assert_eq!(
5517 kernel_predicate_file_paths(
5518 &source,
5519 DeltaKernelPredicate::new(Predicate::is_null(timestamp_partition_column())),
5520 )?,
5521 vec![
5522 "timestamp-empty.parquet",
5523 "timestamp-missing.parquet",
5524 "timestamp-null.parquet",
5525 ]
5526 );
5527 assert_eq!(
5528 kernel_predicate_file_paths(
5529 &source,
5530 DeltaKernelPredicate::new(Predicate::is_not_null(timestamp_partition_column())),
5531 )?,
5532 vec![
5533 "timestamp-high.parquet",
5534 "timestamp-low.parquet",
5535 "timestamp-pre-epoch.parquet",
5536 "timestamp-target.parquet",
5537 ]
5538 );
5539 assert_eq!(
5540 kernel_predicate_file_paths(
5541 &source,
5542 DeltaKernelPredicate::new(Predicate::or(
5543 Predicate::eq(timestamp_partition_column(), timestamp_partition_value(low)),
5544 Predicate::eq(
5545 timestamp_partition_column(),
5546 timestamp_partition_value(target)
5547 ),
5548 )),
5549 )?,
5550 vec!["timestamp-low.parquet", "timestamp-target.parquet"]
5551 );
5552 assert_eq!(
5553 kernel_predicate_file_paths(
5554 &source,
5555 DeltaKernelPredicate::new(Predicate::and(
5556 Predicate::ne(timestamp_partition_column(), timestamp_partition_value(low)),
5557 Predicate::ne(
5558 timestamp_partition_column(),
5559 timestamp_partition_value(target)
5560 ),
5561 )),
5562 )?,
5563 vec!["timestamp-high.parquet", "timestamp-pre-epoch.parquet"]
5564 );
5565
5566 Ok(())
5567 }
5568
5569 #[test]
5570 fn kernel_partition_characterization_documents_timestamp_between_and_composition()
5571 -> Result<(), Box<dyn std::error::Error>> {
5572 let (_table, source) = kernel_timestamp_partition_characterization_source()?;
5573 let low = 1_767_225_599_999_999_i64;
5574 let target = 1_767_225_600_123_456_i64;
5575 let high = 1_767_225_600_123_457_i64;
5576
5577 assert_eq!(
5578 kernel_predicate_file_paths(
5579 &source,
5580 DeltaKernelPredicate::new(Predicate::and(
5581 Predicate::ge(timestamp_partition_column(), timestamp_partition_value(low)),
5582 Predicate::le(
5583 timestamp_partition_column(),
5584 timestamp_partition_value(target)
5585 ),
5586 )),
5587 )?,
5588 vec!["timestamp-low.parquet", "timestamp-target.parquet"]
5589 );
5590 assert_eq!(
5591 kernel_predicate_file_paths(
5592 &source,
5593 DeltaKernelPredicate::new(Predicate::or(
5594 Predicate::lt(timestamp_partition_column(), timestamp_partition_value(low)),
5595 Predicate::gt(
5596 timestamp_partition_column(),
5597 timestamp_partition_value(target)
5598 ),
5599 )),
5600 )?,
5601 vec!["timestamp-high.parquet", "timestamp-pre-epoch.parquet"]
5602 );
5603 assert_eq!(
5604 kernel_predicate_file_paths(
5605 &source,
5606 DeltaKernelPredicate::new(Predicate::and(
5607 Predicate::gt(timestamp_partition_column(), timestamp_partition_value(low)),
5608 Predicate::le(
5609 timestamp_partition_column(),
5610 timestamp_partition_value(high)
5611 ),
5612 )),
5613 )?,
5614 vec!["timestamp-high.parquet", "timestamp-target.parquet"]
5615 );
5616 assert_eq!(
5617 kernel_predicate_file_paths(
5618 &source,
5619 DeltaKernelPredicate::new(Predicate::or(
5620 Predicate::eq(timestamp_partition_column(), timestamp_partition_value(low)),
5621 Predicate::is_null(timestamp_partition_column()),
5622 )),
5623 )?,
5624 vec![
5625 "timestamp-empty.parquet",
5626 "timestamp-low.parquet",
5627 "timestamp-missing.parquet",
5628 "timestamp-null.parquet",
5629 ]
5630 );
5631 assert_eq!(
5632 kernel_predicate_file_paths(
5633 &source,
5634 DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
5635 timestamp_partition_column(),
5636 timestamp_partition_value(target),
5637 ))),
5638 )?,
5639 vec![
5640 "timestamp-high.parquet",
5641 "timestamp-low.parquet",
5642 "timestamp-pre-epoch.parquet",
5643 ]
5644 );
5645
5646 Ok(())
5647 }
5648
5649 #[test]
5650 fn kernel_partition_characterization_documents_timestamp_ntz_ordering_and_formatting()
5651 -> Result<(), Box<dyn std::error::Error>> {
5652 let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
5653 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
5654 let pre_epoch = -1_i64;
5655 let low = 1_767_225_599_999_999_i64;
5656 let target = 1_767_225_600_123_456_i64;
5657 let high = 1_767_225_600_123_457_i64;
5658
5659 assert_eq!(
5660 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
5661 vec![
5662 "timestamp-ntz-empty.parquet",
5663 "timestamp-ntz-high.parquet",
5664 "timestamp-ntz-low-space.parquet",
5665 "timestamp-ntz-missing.parquet",
5666 "timestamp-ntz-null.parquet",
5667 "timestamp-ntz-pre-epoch.parquet",
5668 "timestamp-ntz-target-space.parquet",
5669 ]
5670 );
5671 assert_eq!(
5672 kernel_predicate_file_paths(
5673 &source,
5674 DeltaKernelPredicate::new(Predicate::lt(
5675 timestamp_ntz_partition_column(),
5676 timestamp_ntz_partition_value(target),
5677 )),
5678 )?,
5679 vec![
5680 "timestamp-ntz-low-space.parquet",
5681 "timestamp-ntz-pre-epoch.parquet",
5682 ]
5683 );
5684 assert_eq!(
5685 kernel_predicate_file_paths(
5686 &source,
5687 DeltaKernelPredicate::new(Predicate::ge(
5688 timestamp_ntz_partition_column(),
5689 timestamp_ntz_partition_value(target),
5690 )),
5691 )?,
5692 vec![
5693 "timestamp-ntz-high.parquet",
5694 "timestamp-ntz-target-space.parquet",
5695 ]
5696 );
5697 assert_eq!(
5698 kernel_predicate_file_paths(
5699 &source,
5700 DeltaKernelPredicate::new(Predicate::gt(
5701 timestamp_ntz_partition_value(high),
5702 timestamp_ntz_partition_column(),
5703 )),
5704 )?,
5705 vec![
5706 "timestamp-ntz-low-space.parquet",
5707 "timestamp-ntz-pre-epoch.parquet",
5708 "timestamp-ntz-target-space.parquet",
5709 ]
5710 );
5711 assert_eq!(
5712 kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(pre_epoch))?,
5713 vec!["timestamp-ntz-pre-epoch.parquet"]
5714 );
5715 assert_eq!(
5716 kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(target))?,
5717 vec!["timestamp-ntz-target-space.parquet"]
5718 );
5719 assert_eq!(
5720 kernel_predicate_file_paths(&source, timestamp_ntz_partition_ne(target))?,
5721 vec![
5722 "timestamp-ntz-high.parquet",
5723 "timestamp-ntz-low-space.parquet",
5724 "timestamp-ntz-pre-epoch.parquet",
5725 ]
5726 );
5727 assert_eq!(
5728 kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(high))?,
5729 vec!["timestamp-ntz-high.parquet"]
5730 );
5731 assert_eq!(
5732 kernel_predicate_file_paths(&source, timestamp_ntz_partition_eq(low))?,
5733 vec!["timestamp-ntz-low-space.parquet"]
5734 );
5735
5736 Ok(())
5737 }
5738
5739 #[test]
5740 fn kernel_partition_characterization_documents_timestamp_ntz_null_empty_and_membership()
5741 -> Result<(), Box<dyn std::error::Error>> {
5742 let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
5743 let low = 1_767_225_599_999_999_i64;
5744 let target = 1_767_225_600_123_456_i64;
5745
5746 assert_eq!(
5747 kernel_predicate_file_paths(
5748 &source,
5749 DeltaKernelPredicate::new(Predicate::is_null(timestamp_ntz_partition_column())),
5750 )?,
5751 vec![
5752 "timestamp-ntz-empty.parquet",
5753 "timestamp-ntz-missing.parquet",
5754 "timestamp-ntz-null.parquet",
5755 ]
5756 );
5757 assert_eq!(
5758 kernel_predicate_file_paths(
5759 &source,
5760 DeltaKernelPredicate::new(Predicate::is_not_null(timestamp_ntz_partition_column())),
5761 )?,
5762 vec![
5763 "timestamp-ntz-high.parquet",
5764 "timestamp-ntz-low-space.parquet",
5765 "timestamp-ntz-pre-epoch.parquet",
5766 "timestamp-ntz-target-space.parquet",
5767 ]
5768 );
5769 assert_eq!(
5770 kernel_predicate_file_paths(
5771 &source,
5772 DeltaKernelPredicate::new(Predicate::or(
5773 Predicate::eq(
5774 timestamp_ntz_partition_column(),
5775 timestamp_ntz_partition_value(low)
5776 ),
5777 Predicate::eq(
5778 timestamp_ntz_partition_column(),
5779 timestamp_ntz_partition_value(target)
5780 ),
5781 )),
5782 )?,
5783 vec![
5784 "timestamp-ntz-low-space.parquet",
5785 "timestamp-ntz-target-space.parquet",
5786 ]
5787 );
5788 assert_eq!(
5789 kernel_predicate_file_paths(
5790 &source,
5791 DeltaKernelPredicate::new(Predicate::and(
5792 Predicate::ne(
5793 timestamp_ntz_partition_column(),
5794 timestamp_ntz_partition_value(low)
5795 ),
5796 Predicate::ne(
5797 timestamp_ntz_partition_column(),
5798 timestamp_ntz_partition_value(target)
5799 ),
5800 )),
5801 )?,
5802 vec![
5803 "timestamp-ntz-high.parquet",
5804 "timestamp-ntz-pre-epoch.parquet",
5805 ]
5806 );
5807
5808 Ok(())
5809 }
5810
5811 #[test]
5812 fn kernel_partition_characterization_documents_timestamp_ntz_between_and_composition()
5813 -> Result<(), Box<dyn std::error::Error>> {
5814 let (_table, source) = kernel_timestamp_ntz_partition_characterization_source()?;
5815 let low = 1_767_225_599_999_999_i64;
5816 let target = 1_767_225_600_123_456_i64;
5817 let high = 1_767_225_600_123_457_i64;
5818
5819 assert_eq!(
5820 kernel_predicate_file_paths(
5821 &source,
5822 DeltaKernelPredicate::new(Predicate::and(
5823 Predicate::ge(
5824 timestamp_ntz_partition_column(),
5825 timestamp_ntz_partition_value(low)
5826 ),
5827 Predicate::le(
5828 timestamp_ntz_partition_column(),
5829 timestamp_ntz_partition_value(target)
5830 ),
5831 )),
5832 )?,
5833 vec![
5834 "timestamp-ntz-low-space.parquet",
5835 "timestamp-ntz-target-space.parquet",
5836 ]
5837 );
5838 assert_eq!(
5839 kernel_predicate_file_paths(
5840 &source,
5841 DeltaKernelPredicate::new(Predicate::or(
5842 Predicate::lt(
5843 timestamp_ntz_partition_column(),
5844 timestamp_ntz_partition_value(low)
5845 ),
5846 Predicate::gt(
5847 timestamp_ntz_partition_column(),
5848 timestamp_ntz_partition_value(target)
5849 ),
5850 )),
5851 )?,
5852 vec![
5853 "timestamp-ntz-high.parquet",
5854 "timestamp-ntz-pre-epoch.parquet",
5855 ]
5856 );
5857 assert_eq!(
5858 kernel_predicate_file_paths(
5859 &source,
5860 DeltaKernelPredicate::new(Predicate::and(
5861 Predicate::gt(
5862 timestamp_ntz_partition_column(),
5863 timestamp_ntz_partition_value(low)
5864 ),
5865 Predicate::le(
5866 timestamp_ntz_partition_column(),
5867 timestamp_ntz_partition_value(high)
5868 ),
5869 )),
5870 )?,
5871 vec![
5872 "timestamp-ntz-high.parquet",
5873 "timestamp-ntz-target-space.parquet",
5874 ]
5875 );
5876 assert_eq!(
5877 kernel_predicate_file_paths(
5878 &source,
5879 DeltaKernelPredicate::new(Predicate::or(
5880 Predicate::eq(
5881 timestamp_ntz_partition_column(),
5882 timestamp_ntz_partition_value(low)
5883 ),
5884 Predicate::is_null(timestamp_ntz_partition_column()),
5885 )),
5886 )?,
5887 vec![
5888 "timestamp-ntz-empty.parquet",
5889 "timestamp-ntz-low-space.parquet",
5890 "timestamp-ntz-missing.parquet",
5891 "timestamp-ntz-null.parquet",
5892 ]
5893 );
5894 assert_eq!(
5895 kernel_predicate_file_paths(
5896 &source,
5897 DeltaKernelPredicate::new(Predicate::not(Predicate::eq(
5898 timestamp_ntz_partition_column(),
5899 timestamp_ntz_partition_value(target),
5900 ))),
5901 )?,
5902 vec![
5903 "timestamp-ntz-high.parquet",
5904 "timestamp-ntz-low-space.parquet",
5905 "timestamp-ntz-pre-epoch.parquet",
5906 ]
5907 );
5908
5909 Ok(())
5910 }
5911
5912 #[test]
5913 fn kernel_partition_characterization_documents_invalid_timestamp_metadata()
5914 -> Result<(), Box<dyn std::error::Error>> {
5915 let table = DeltaLogTable::new_with_metadata_and_adds(
5916 "kernel-invalid-timestamp-characterization",
5917 TIMESTAMP_PARTITIONED_METADATA_JSON,
5918 &[
5919 partitioned_add_json(
5920 "timestamp-valid.parquet",
5921 r#"{"event_ts":"2026-01-01T00:00:00.123456Z"}"#,
5922 ),
5923 partitioned_add_json(
5924 "timestamp-invalid.parquet",
5925 r#"{"event_ts":"not-a-timestamp"}"#,
5926 ),
5927 ],
5928 )?;
5929 let source = load_delta_source(DeltaSourceConfig {
5930 name: "orders".to_owned(),
5931 table_uri: table.path.to_string_lossy().to_string(),
5932 version: None,
5933 storage_options: Default::default(),
5934 })?;
5935 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
5936
5937 assert_invalid_timestamp_partition_error(kernel_scan_file_paths(
5938 &unfiltered_scan,
5939 source.table_uri(),
5940 ))?;
5941 assert_invalid_timestamp_partition_error(kernel_predicate_file_paths(
5942 &source,
5943 timestamp_partition_eq(1_767_225_600_123_456),
5944 ))?;
5945
5946 Ok(())
5947 }
5948
5949 #[test]
5950 fn kernel_partition_characterization_documents_invalid_timestamp_ntz_metadata()
5951 -> Result<(), Box<dyn std::error::Error>> {
5952 let invalid_text_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
5953 "kernel-invalid-timestamp-ntz-characterization",
5954 TIMESTAMP_NTZ_PROTOCOL_JSON,
5955 TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
5956 &[
5957 partitioned_add_json(
5958 "timestamp-ntz-valid.parquet",
5959 r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
5960 ),
5961 partitioned_add_json(
5962 "timestamp-ntz-invalid.parquet",
5963 r#"{"event_ts_ntz":"not-a-timestamp"}"#,
5964 ),
5965 ],
5966 )?;
5967 let invalid_text_source = load_delta_source(DeltaSourceConfig {
5968 name: "orders".to_owned(),
5969 table_uri: invalid_text_table.path.to_string_lossy().to_string(),
5970 version: None,
5971 storage_options: Default::default(),
5972 })?;
5973 let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;
5974
5975 assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
5976 &unfiltered_scan,
5977 invalid_text_source.table_uri(),
5978 ))?;
5979 assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
5980 &invalid_text_source,
5981 timestamp_ntz_partition_eq(1_767_225_600_123_456),
5982 ))?;
5983
5984 let t_separator_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
5985 "kernel-t-separator-timestamp-ntz-characterization",
5986 TIMESTAMP_NTZ_PROTOCOL_JSON,
5987 TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
5988 &[
5989 partitioned_add_json(
5990 "timestamp-ntz-valid.parquet",
5991 r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
5992 ),
5993 partitioned_add_json(
5994 "timestamp-ntz-t-separator.parquet",
5995 r#"{"event_ts_ntz":"2026-01-01T00:00:00.123456"}"#,
5996 ),
5997 ],
5998 )?;
5999 let t_separator_source = load_delta_source(DeltaSourceConfig {
6000 name: "orders".to_owned(),
6001 table_uri: t_separator_table.path.to_string_lossy().to_string(),
6002 version: None,
6003 storage_options: Default::default(),
6004 })?;
6005 let unfiltered_scan = build_projected_delta_scan(&t_separator_source, None)?;
6006
6007 assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
6008 &unfiltered_scan,
6009 t_separator_source.table_uri(),
6010 ))?;
6011 assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
6012 &t_separator_source,
6013 timestamp_ntz_partition_eq(1_767_225_600_123_456),
6014 ))?;
6015
6016 let zone_table = DeltaLogTable::new_with_protocol_metadata_and_adds(
6017 "kernel-zone-timestamp-ntz-characterization",
6018 TIMESTAMP_NTZ_PROTOCOL_JSON,
6019 TIMESTAMP_NTZ_PARTITIONED_METADATA_JSON,
6020 &[
6021 partitioned_add_json(
6022 "timestamp-ntz-valid.parquet",
6023 r#"{"event_ts_ntz":"2026-01-01 00:00:00.123456"}"#,
6024 ),
6025 partitioned_add_json(
6026 "timestamp-ntz-zone.parquet",
6027 r#"{"event_ts_ntz":"2026-01-01T00:00:00.123456Z"}"#,
6028 ),
6029 ],
6030 )?;
6031 let zone_source = load_delta_source(DeltaSourceConfig {
6032 name: "orders".to_owned(),
6033 table_uri: zone_table.path.to_string_lossy().to_string(),
6034 version: None,
6035 storage_options: Default::default(),
6036 })?;
6037 let unfiltered_scan = build_projected_delta_scan(&zone_source, None)?;
6038
6039 assert_invalid_timestamp_ntz_partition_error(kernel_scan_file_paths(
6040 &unfiltered_scan,
6041 zone_source.table_uri(),
6042 ))?;
6043 assert_invalid_timestamp_ntz_partition_error(kernel_predicate_file_paths(
6044 &zone_source,
6045 timestamp_ntz_partition_eq(1_767_225_600_123_456),
6046 ))?;
6047
6048 Ok(())
6049 }
6050
6051 #[test]
6052 fn kernel_partition_characterization_documents_invalid_floating_metadata()
6053 -> Result<(), Box<dyn std::error::Error>> {
6054 let invalid_text_table = DeltaLogTable::new_with_metadata_and_adds(
6055 "kernel-invalid-floating-characterization",
6056 FLOATING_PARTITIONED_METADATA_JSON,
6057 &[
6058 partitioned_add_json(
6059 "floating-valid.parquet",
6060 r#"{"float_part":"1.5","double_part":"2.25"}"#,
6061 ),
6062 partitioned_add_json(
6063 "floating-invalid.parquet",
6064 r#"{"float_part":"not-a-float","double_part":"not-a-double"}"#,
6065 ),
6066 ],
6067 )?;
6068 let invalid_text_source = load_delta_source(DeltaSourceConfig {
6069 name: "orders".to_owned(),
6070 table_uri: invalid_text_table.path.to_string_lossy().to_string(),
6071 version: None,
6072 storage_options: Default::default(),
6073 })?;
6074 let unfiltered_scan = build_projected_delta_scan(&invalid_text_source, None)?;
6075
6076 assert_invalid_floating_partition_error(kernel_scan_file_paths(
6077 &unfiltered_scan,
6078 invalid_text_source.table_uri(),
6079 ))?;
6080 assert_invalid_floating_partition_error(kernel_predicated_file_paths(
6081 &invalid_text_source,
6082 &col("float_part").eq(float32_lit(1.5)),
6083 ))?;
6084
6085 Ok(())
6086 }
6087
6088 #[test]
6089 fn kernel_partition_characterization_documents_nonfinite_floating_metadata()
6090 -> Result<(), Box<dyn std::error::Error>> {
6091 let table = DeltaLogTable::new_with_metadata_and_adds(
6092 "kernel-nonfinite-floating-characterization",
6093 FLOATING_PARTITIONED_METADATA_JSON,
6094 &[
6095 partitioned_add_json(
6096 "floating-valid.parquet",
6097 r#"{"float_part":"1.5","double_part":"2.25"}"#,
6098 ),
6099 partitioned_add_json(
6100 "floating-nan.parquet",
6101 r#"{"float_part":"NaN","double_part":"NaN"}"#,
6102 ),
6103 partitioned_add_json(
6104 "floating-inf.parquet",
6105 r#"{"float_part":"Infinity","double_part":"Infinity"}"#,
6106 ),
6107 partitioned_add_json(
6108 "floating-neg-inf.parquet",
6109 r#"{"float_part":"-Infinity","double_part":"-Infinity"}"#,
6110 ),
6111 ],
6112 )?;
6113 let source = load_delta_source(DeltaSourceConfig {
6114 name: "orders".to_owned(),
6115 table_uri: table.path.to_string_lossy().to_string(),
6116 version: None,
6117 storage_options: Default::default(),
6118 })?;
6119 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
6120
6121 assert_eq!(
6122 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
6123 vec![
6124 "floating-inf.parquet",
6125 "floating-nan.parquet",
6126 "floating-neg-inf.parquet",
6127 "floating-valid.parquet",
6128 ]
6129 );
6130 assert_eq!(
6131 kernel_predicated_file_paths(&source, &col("float_part").gt(float32_lit(0.0)))?,
6132 vec![
6133 "floating-inf.parquet",
6134 "floating-nan.parquet",
6135 "floating-valid.parquet",
6136 ]
6137 );
6138 assert_eq!(
6139 kernel_predicated_file_paths(&source, &col("double_part").lt(float64_lit(0.0)))?,
6140 vec!["floating-neg-inf.parquet"]
6141 );
6142 assert_eq!(
6143 kernel_predicate_file_paths(
6144 &source,
6145 DeltaKernelPredicate::new(Predicate::eq(
6146 Expression::Column(ColumnName::new(["float_part"])),
6147 Expression::Literal(Scalar::Float(f32::NAN)),
6148 )),
6149 )?,
6150 vec!["floating-nan.parquet"]
6151 );
6152
6153 Ok(())
6154 }
6155
6156 #[test]
6157 fn kernel_partition_characterization_documents_invalid_date_metadata()
6158 -> Result<(), Box<dyn std::error::Error>> {
6159 let table = DeltaLogTable::new_with_metadata_and_adds(
6160 "kernel-invalid-date-characterization",
6161 DATE_PARTITIONED_METADATA_JSON,
6162 &[
6163 partitioned_add_json("date-valid.parquet", r#"{"event_date":"2026-01-01"}"#),
6164 partitioned_add_json("date-invalid.parquet", r#"{"event_date":"not-a-date"}"#),
6165 ],
6166 )?;
6167 let source = load_delta_source(DeltaSourceConfig {
6168 name: "orders".to_owned(),
6169 table_uri: table.path.to_string_lossy().to_string(),
6170 version: None,
6171 storage_options: Default::default(),
6172 })?;
6173 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
6174
6175 assert_invalid_date_partition_error(kernel_scan_file_paths(
6176 &unfiltered_scan,
6177 source.table_uri(),
6178 ))?;
6179 assert_invalid_date_partition_error(kernel_predicated_file_paths(
6180 &source,
6181 &col("event_date").eq(date_lit(20_454)),
6182 ))?;
6183
6184 Ok(())
6185 }
6186
6187 #[test]
6188 fn kernel_partition_characterization_documents_invalid_boolean_metadata()
6189 -> Result<(), Box<dyn std::error::Error>> {
6190 let table = DeltaLogTable::new_with_metadata_and_adds(
6191 "kernel-invalid-boolean-characterization",
6192 BOOLEAN_PARTITIONED_METADATA_JSON,
6193 &[
6194 partitioned_add_json("boolean-valid.parquet", r#"{"is_current":"true"}"#),
6195 partitioned_add_json(
6196 "boolean-invalid.parquet",
6197 r#"{"is_current":"not-a-boolean"}"#,
6198 ),
6199 ],
6200 )?;
6201 let source = load_delta_source(DeltaSourceConfig {
6202 name: "orders".to_owned(),
6203 table_uri: table.path.to_string_lossy().to_string(),
6204 version: None,
6205 storage_options: Default::default(),
6206 })?;
6207 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
6208
6209 assert_invalid_boolean_partition_error(kernel_scan_file_paths(
6210 &unfiltered_scan,
6211 source.table_uri(),
6212 ))?;
6213 assert_invalid_boolean_partition_error(kernel_predicated_file_paths(
6214 &source,
6215 &col("is_current").eq(bool_lit(true)),
6216 ))?;
6217
6218 Ok(())
6219 }
6220
6221 #[test]
6222 fn kernel_partition_characterization_documents_integer_numeric_ordering()
6223 -> Result<(), Box<dyn std::error::Error>> {
6224 let (_table, source) = kernel_integer_partition_characterization_source()?;
6225 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
6226
6227 assert_eq!(
6228 kernel_scan_file_paths(&unfiltered_scan, source.table_uri())?,
6229 vec![
6230 "integer--1.parquet",
6231 "integer-10.parquet",
6232 "integer-2.parquet",
6233 "integer-empty.parquet",
6234 "integer-missing.parquet",
6235 "integer-null.parquet",
6236 ]
6237 );
6238 assert_eq!(
6239 kernel_predicated_file_paths(&source, &col("long_part").gt(int64_lit(2)))?,
6240 vec!["integer-10.parquet"]
6241 );
6242 assert_eq!(
6243 kernel_predicated_file_paths(&source, &col("long_part").lt(int64_lit(10)))?,
6244 vec!["integer--1.parquet", "integer-2.parquet"]
6245 );
6246 assert_eq!(
6247 kernel_predicated_file_paths(&source, &int64_lit(10).gt(col("long_part")))?,
6248 vec!["integer--1.parquet", "integer-2.parquet"]
6249 );
6250
6251 Ok(())
6252 }
6253
6254 #[test]
6255 fn kernel_partition_characterization_documents_integer_null_and_empty_semantics()
6256 -> Result<(), Box<dyn std::error::Error>> {
6257 let (_table, source) = kernel_integer_partition_characterization_source()?;
6258
6259 assert_eq!(
6260 kernel_predicated_file_paths(&source, &col("long_part").is_null())?,
6261 vec![
6262 "integer-empty.parquet",
6263 "integer-missing.parquet",
6264 "integer-null.parquet",
6265 ]
6266 );
6267 assert_eq!(
6268 kernel_predicated_file_paths(&source, &col("long_part").is_not_null())?,
6269 vec![
6270 "integer--1.parquet",
6271 "integer-10.parquet",
6272 "integer-2.parquet",
6273 ]
6274 );
6275 assert_eq!(
6276 kernel_predicated_file_paths(&source, &col("long_part").eq(int64_lit(2)))?,
6277 vec!["integer-2.parquet"]
6278 );
6279 assert_eq!(
6280 kernel_predicated_file_paths(&source, &col("long_part").not_eq(int64_lit(2)))?,
6281 vec!["integer--1.parquet", "integer-10.parquet"]
6282 );
6283
6284 Ok(())
6285 }
6286
6287 #[test]
6288 fn kernel_partition_characterization_documents_integer_membership_between_and_composition()
6289 -> Result<(), Box<dyn std::error::Error>> {
6290 let (_table, source) = kernel_integer_partition_characterization_source()?;
6291
6292 assert_eq!(
6293 kernel_predicated_file_paths(
6294 &source,
6295 &col("long_part").in_list(vec![int64_lit(-1), int64_lit(10)], false),
6296 )?,
6297 vec!["integer--1.parquet", "integer-10.parquet"]
6298 );
6299 assert_eq!(
6300 kernel_predicated_file_paths(
6301 &source,
6302 &col("long_part").in_list(vec![int64_lit(2)], true),
6303 )?,
6304 vec!["integer--1.parquet", "integer-10.parquet"]
6305 );
6306 assert_eq!(
6307 kernel_predicated_file_paths(
6308 &source,
6309 &col("long_part").between(int64_lit(-1), int64_lit(2)),
6310 )?,
6311 vec!["integer--1.parquet", "integer-2.parquet"]
6312 );
6313 assert_eq!(
6314 kernel_predicated_file_paths(
6315 &source,
6316 &col("long_part").not_between(int64_lit(-1), int64_lit(2)),
6317 )?,
6318 vec!["integer-10.parquet"]
6319 );
6320 assert_eq!(
6321 kernel_predicated_file_paths(
6322 &source,
6323 &col("long_part")
6324 .gt_eq(int64_lit(-1))
6325 .and(col("long_part").lt(int64_lit(10))),
6326 )?,
6327 vec!["integer--1.parquet", "integer-2.parquet"]
6328 );
6329 assert_eq!(
6330 kernel_predicated_file_paths(
6331 &source,
6332 &col("long_part")
6333 .eq(int64_lit(-1))
6334 .or(col("long_part").eq(int64_lit(10))),
6335 )?,
6336 vec!["integer--1.parquet", "integer-10.parquet"]
6337 );
6338 assert_eq!(
6339 kernel_predicated_file_paths(
6340 &source,
6341 &Expr::Not(Box::new(col("long_part").eq(int64_lit(2)))),
6342 )?,
6343 vec!["integer--1.parquet", "integer-10.parquet"]
6344 );
6345
6346 Ok(())
6347 }
6348
6349 #[test]
6350 fn kernel_partition_characterization_documents_integer_width_scalars()
6351 -> Result<(), Box<dyn std::error::Error>> {
6352 let table = DeltaLogTable::new_with_metadata_and_adds(
6353 "kernel-integer-width-characterization",
6354 INTEGER_PARTITIONED_METADATA_JSON,
6355 &[
6356 partitioned_add_json(
6357 "width-byte-min.parquet",
6358 r#"{"byte_part":"-128","short_part":"0","int_part":"0","long_part":"0"}"#,
6359 ),
6360 partitioned_add_json(
6361 "width-short-max.parquet",
6362 r#"{"byte_part":"0","short_part":"32767","int_part":"0","long_part":"0"}"#,
6363 ),
6364 partitioned_add_json(
6365 "width-int-max.parquet",
6366 r#"{"byte_part":"0","short_part":"0","int_part":"2147483647","long_part":"0"}"#,
6367 ),
6368 partitioned_add_json(
6369 "width-long-max.parquet",
6370 r#"{"byte_part":"0","short_part":"0","int_part":"0","long_part":"9223372036854775807"}"#,
6371 ),
6372 ],
6373 )?;
6374 let source = load_delta_source(DeltaSourceConfig {
6375 name: "orders".to_owned(),
6376 table_uri: table.path.to_string_lossy().to_string(),
6377 version: None,
6378 storage_options: Default::default(),
6379 })?;
6380
6381 assert_eq!(
6382 kernel_predicated_file_paths(&source, &col("byte_part").eq(int8_lit(i8::MIN)))?,
6383 vec!["width-byte-min.parquet"]
6384 );
6385 assert_eq!(
6386 kernel_predicated_file_paths(&source, &col("short_part").eq(int16_lit(i16::MAX)))?,
6387 vec!["width-short-max.parquet"]
6388 );
6389 assert_eq!(
6390 kernel_predicated_file_paths(&source, &col("int_part").eq(int32_lit(i32::MAX)))?,
6391 vec!["width-int-max.parquet"]
6392 );
6393 assert_eq!(
6394 kernel_predicated_file_paths(&source, &col("long_part").eq(int64_lit(i64::MAX)))?,
6395 vec!["width-long-max.parquet"]
6396 );
6397
6398 Ok(())
6399 }
6400
6401 #[test]
6402 fn kernel_partition_characterization_documents_invalid_integer_metadata()
6403 -> Result<(), Box<dyn std::error::Error>> {
6404 let table = DeltaLogTable::new_with_metadata_and_adds(
6405 "kernel-invalid-integer-characterization",
6406 INTEGER_PARTITIONED_METADATA_JSON,
6407 &[
6408 integer_partitioned_add_json("integer-valid.parquet", "7"),
6409 partitioned_add_json(
6410 "integer-invalid.parquet",
6411 r#"{"byte_part":"0","short_part":"0","int_part":"0","long_part":"not-an-integer"}"#,
6412 ),
6413 ],
6414 )?;
6415 let source = load_delta_source(DeltaSourceConfig {
6416 name: "orders".to_owned(),
6417 table_uri: table.path.to_string_lossy().to_string(),
6418 version: None,
6419 storage_options: Default::default(),
6420 })?;
6421 let unfiltered_scan = build_projected_delta_scan(&source, None)?;
6422
6423 assert_invalid_integer_partition_error(kernel_scan_file_paths(
6424 &unfiltered_scan,
6425 source.table_uri(),
6426 ))?;
6427 assert_invalid_integer_partition_error(kernel_predicated_file_paths(
6428 &source,
6429 &col("long_part").eq(int64_lit(7)),
6430 ))?;
6431
6432 Ok(())
6433 }
6434
6435 #[test]
6436 fn loads_named_delta_source() -> Result<(), Box<dyn std::error::Error>> {
6437 let table = DeltaLogTable::new("success")?;
6438 let requested_table_uri = table.path.to_string_lossy().to_string();
6439
6440 let source = load_delta_source(DeltaSourceConfig {
6441 name: "orders".to_owned(),
6442 table_uri: requested_table_uri.clone(),
6443 version: None,
6444 storage_options: Default::default(),
6445 })?;
6446
6447 assert_eq!(source.name(), "orders");
6448 assert_eq!(source.requested_table_uri(), requested_table_uri);
6449 assert!(source.table_uri().starts_with("file://"));
6450 assert_eq!(source.version(), 1);
6451 assert_eq!(source.loaded_snapshot().version(), 1);
6452
6453 Ok(())
6454 }
6455
6456 #[test]
6457 fn planned_delta_source_owns_and_isolates_storage_options()
6458 -> Result<(), Box<dyn std::error::Error>> {
6459 let table = DeltaLogTable::new("storage-options-owned")?;
6460 let mut caller_options = storage_options(&[
6461 ("authorization", "caller-original"),
6462 ("region", "us-west-2"),
6463 ]);
6464 let source = load_delta_source(
6465 DeltaSourceConfig::new("orders", table.path.to_string_lossy())
6466 .with_storage_options(caller_options.clone()),
6467 )?;
6468
6469 caller_options.insert("authorization".to_owned(), "caller-mutated".to_owned());
6470 caller_options.insert("new-secret".to_owned(), "caller-added".to_owned());
6471
6472 assert_eq!(
6473 source
6474 .storage_options()
6475 .get("authorization")
6476 .map(String::as_str),
6477 Some("caller-original")
6478 );
6479 assert_eq!(
6480 source.storage_options().get("region").map(String::as_str),
6481 Some("us-west-2")
6482 );
6483 assert!(!source.storage_options().contains_key("new-secret"));
6484
6485 Ok(())
6486 }
6487
6488 #[test]
6489 fn s3_effective_storage_options_merge_aws_env_when_absent() {
6490 for table_uri in [
6491 "s3://bucket/root",
6492 "s3a://bucket/root",
6493 "https://s3.us-east-1.amazonaws.com/bucket/root",
6494 "https://bucket.s3.us-east-1.amazonaws.com/root",
6495 "https://ACCOUNT_ID.r2.cloudflarestorage.com/bucket/root",
6496 ] {
6497 let effective = effective_storage_options_for_source_from_env(
6498 table_uri,
6499 DeltaStorageOptions::default(),
6500 env_vars(&[
6501 ("AWS_ACCESS_KEY_ID", "env-access"),
6502 ("AWS_SECRET_ACCESS_KEY", "env-secret"),
6503 ("AWS_REGION", "us-east-1"),
6504 ("AWS_CUSTOM_FUTURE_OPTION", "future"),
6505 ("NOT_AWS", "ignored"),
6506 ]),
6507 );
6508
6509 assert_eq!(
6510 effective.get("AWS_ACCESS_KEY_ID").map(String::as_str),
6511 Some("env-access"),
6512 "{table_uri}"
6513 );
6514 assert_eq!(
6515 effective.get("AWS_SECRET_ACCESS_KEY").map(String::as_str),
6516 Some("env-secret"),
6517 "{table_uri}"
6518 );
6519 assert_eq!(
6520 effective.get("AWS_REGION").map(String::as_str),
6521 Some("us-east-1"),
6522 "{table_uri}"
6523 );
6524 assert_eq!(
6525 effective
6526 .get("AWS_CUSTOM_FUTURE_OPTION")
6527 .map(String::as_str),
6528 Some("future"),
6529 "{table_uri}"
6530 );
6531 assert!(!effective.contains_key("NOT_AWS"), "{table_uri}");
6532 }
6533 }
6534
6535 #[test]
6536 fn s3_effective_storage_options_prefer_explicit_caller_options() {
6537 let caller_options = storage_options(&[
6538 ("aws_access_key_id", "caller-access"),
6539 ("AWS_SECRET_ACCESS_KEY", "caller-secret"),
6540 ("region", "caller-region"),
6541 ]);
6542 let effective = effective_storage_options_for_source_from_env(
6543 "s3://bucket/root",
6544 caller_options,
6545 env_vars(&[
6546 ("AWS_ACCESS_KEY_ID", "env-access"),
6547 ("AWS_SECRET_ACCESS_KEY", "env-secret"),
6548 ("AWS_REGION", "env-region"),
6549 ("AWS_CUSTOM_FUTURE_OPTION", "future"),
6550 ]),
6551 );
6552
6553 assert_eq!(
6554 effective.get("aws_access_key_id").map(String::as_str),
6555 Some("caller-access")
6556 );
6557 assert_eq!(
6558 effective.get("AWS_SECRET_ACCESS_KEY").map(String::as_str),
6559 Some("caller-secret")
6560 );
6561 assert_eq!(
6562 effective.get("region").map(String::as_str),
6563 Some("caller-region")
6564 );
6565 assert_eq!(
6566 effective
6567 .get("AWS_CUSTOM_FUTURE_OPTION")
6568 .map(String::as_str),
6569 Some("future")
6570 );
6571 assert!(!effective.contains_key("AWS_ACCESS_KEY_ID"));
6572 assert!(!effective.contains_key("AWS_REGION"));
6573 }
6574
6575 #[test]
6576 fn non_s3_effective_storage_options_do_not_merge_aws_env() {
6577 let caller_options = storage_options(&[("authorization", "caller-token")]);
6578 let effective = effective_storage_options_for_source_from_env(
6579 "file:///tmp/table",
6580 caller_options.clone(),
6581 env_vars(&[
6582 ("AWS_ACCESS_KEY_ID", "env-access"),
6583 ("AWS_SECRET_ACCESS_KEY", "env-secret"),
6584 ("AWS_REGION", "us-east-1"),
6585 ]),
6586 );
6587
6588 assert_eq!(effective, caller_options);
6589 }
6590
6591 #[test]
6592 fn s3_effective_storage_options_are_resolved_before_source_load() {
6593 let storage_options = effective_storage_options_for_source_from_env(
6594 "s3://bucket/root",
6595 storage_options(&[("region", "caller-region")]),
6596 env_vars(&[
6597 ("AWS_ACCESS_KEY_ID", "env-access"),
6598 ("AWS_SECRET_ACCESS_KEY", "env-secret"),
6599 ("AWS_REGION", "env-region"),
6600 ]),
6601 );
6602
6603 assert_eq!(
6604 storage_options.get("AWS_ACCESS_KEY_ID").map(String::as_str),
6605 Some("env-access")
6606 );
6607 assert_eq!(
6608 storage_options
6609 .get("AWS_SECRET_ACCESS_KEY")
6610 .map(String::as_str),
6611 Some("env-secret")
6612 );
6613 assert_eq!(
6614 storage_options.get("region").map(String::as_str),
6615 Some("caller-region")
6616 );
6617 assert!(!storage_options.contains_key("AWS_REGION"));
6618 }
6619
6620 #[test]
6621 fn s3_storage_options_accept_documented_credential_keys() {
6622 for (option_key, config_key, expected_value) in [
6623 (
6624 "AWS_ACCESS_KEY_ID",
6625 AmazonS3ConfigKey::AccessKeyId,
6626 "upper-access",
6627 ),
6628 (
6629 "aws_access_key_id",
6630 AmazonS3ConfigKey::AccessKeyId,
6631 "lower-access",
6632 ),
6633 (
6634 "AWS_SECRET_ACCESS_KEY",
6635 AmazonS3ConfigKey::SecretAccessKey,
6636 "upper-secret",
6637 ),
6638 (
6639 "aws_secret_access_key",
6640 AmazonS3ConfigKey::SecretAccessKey,
6641 "lower-secret",
6642 ),
6643 ("AWS_SESSION_TOKEN", AmazonS3ConfigKey::Token, "upper-token"),
6644 ("aws_session_token", AmazonS3ConfigKey::Token, "lower-token"),
6645 ] {
6646 let options = storage_options(&[(option_key, expected_value)]);
6647 let builder = s3_builder_from_storage_options(&options);
6648
6649 assert_eq!(
6650 builder.get_config_value(&config_key).as_deref(),
6651 Some(expected_value),
6652 "{option_key}"
6653 );
6654 }
6655 }
6656
6657 #[test]
6658 fn s3_storage_options_accept_documented_region_keys() {
6659 for option_key in ["AWS_REGION", "aws_region", "region"] {
6660 let options = storage_options(&[(option_key, "explicit-region")]);
6661 let builder = s3_builder_from_storage_options(&options);
6662
6663 assert_eq!(
6664 builder
6665 .get_config_value(&AmazonS3ConfigKey::Region)
6666 .as_deref(),
6667 Some("explicit-region"),
6668 "{option_key}"
6669 );
6670 }
6671
6672 for option_key in ["AWS_DEFAULT_REGION", "aws_default_region"] {
6673 let options = storage_options(&[(option_key, "fallback-region")]);
6674 let builder = s3_builder_from_storage_options(&options);
6675
6676 assert_eq!(
6677 builder
6678 .get_config_value(&AmazonS3ConfigKey::Region)
6679 .as_deref(),
6680 Some("fallback-region"),
6681 "{option_key}"
6682 );
6683 }
6684 }
6685
6686 #[test]
6687 fn s3_default_region_only_applies_without_explicit_region() {
6688 for default_key in ["AWS_DEFAULT_REGION", "aws_default_region"] {
6689 for explicit_key in ["AWS_REGION", "aws_region", "region"] {
6690 let options = storage_options(&[
6691 (default_key, "fallback-region"),
6692 (explicit_key, "explicit-region"),
6693 ]);
6694 let builder = s3_builder_from_storage_options(&options);
6695
6696 assert_eq!(
6697 builder
6698 .get_config_value(&AmazonS3ConfigKey::Region)
6699 .as_deref(),
6700 Some("explicit-region"),
6701 "{default_key} with {explicit_key}"
6702 );
6703 }
6704 }
6705 }
6706
6707 #[test]
6708 fn s3_store_construction_accepts_documented_storage_option_keys()
6709 -> Result<(), Box<dyn std::error::Error>> {
6710 let cases = [
6711 (
6712 "AWS_ACCESS_KEY_ID",
6713 "AWS_SECRET_ACCESS_KEY",
6714 "AWS_SESSION_TOKEN",
6715 "AWS_REGION",
6716 ),
6717 (
6718 "aws_access_key_id",
6719 "aws_secret_access_key",
6720 "aws_session_token",
6721 "aws_region",
6722 ),
6723 (
6724 "aws_access_key_id",
6725 "aws_secret_access_key",
6726 "aws_session_token",
6727 "region",
6728 ),
6729 (
6730 "aws_access_key_id",
6731 "aws_secret_access_key",
6732 "aws_session_token",
6733 "AWS_DEFAULT_REGION",
6734 ),
6735 (
6736 "aws_access_key_id",
6737 "aws_secret_access_key",
6738 "aws_session_token",
6739 "aws_default_region",
6740 ),
6741 ];
6742 let table_url = super::kernel::try_parse_uri("s3://bucket/root")?;
6743
6744 for (access_key, secret_key, token_key, region_key) in cases {
6745 let options = storage_options(&[
6746 (access_key, "access"),
6747 (secret_key, "secret"),
6748 (token_key, "token"),
6749 (region_key, "us-east-1"),
6750 ]);
6751
6752 let _store = super::kernel::store_from_url_opts(
6753 &table_url,
6754 options
6755 .iter()
6756 .map(|(key, value)| (key.as_str(), value.as_str())),
6757 )?;
6758 }
6759
6760 Ok(())
6761 }
6762
6763 #[test]
6764 fn source_loading_passes_storage_options_to_kernel_store_construction()
6765 -> Result<(), Box<dyn std::error::Error>> {
6766 let scheme = unique_storage_scheme("sourceoptions")?;
6767 let captured = CapturedStorageOptions::default();
6768 register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
6769 let options =
6770 storage_options(&[("authorization", "source-token"), ("region", "us-east-1")]);
6771
6772 let result = load_delta_source(
6773 DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
6774 .with_storage_options(options.clone()),
6775 );
6776
6777 assert!(matches!(
6778 result,
6779 Err(DeltaFunnelError::DeltaSnapshotLoad { .. })
6780 ));
6781 assert_eq!(captured_storage_options(&captured), vec![options]);
6782
6783 Ok(())
6784 }
6785
6786 #[test]
6787 fn reader_construction_passes_storage_options_to_each_store_construction()
6788 -> Result<(), Box<dyn std::error::Error>> {
6789 let scheme = unique_storage_scheme("readeroptions")?;
6790 let captured = CapturedStorageOptions::default();
6791 register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
6792 let table_uri = format!("{scheme}://table/root/");
6793 let options = storage_options(&[
6794 ("authorization", "reader-token"),
6795 ("endpoint", "http://storage.example"),
6796 ]);
6797
6798 let _data_reader = KernelDataFileReader::try_new(KernelDataFileReaderConfig {
6799 source_name: "orders",
6800 table_uri: &table_uri,
6801 snapshot_version: 42,
6802 storage_options: &options,
6803 })?;
6804 let _dv_reader = KernelDeletionVectorReader::try_new(KernelDeletionVectorReaderConfig {
6805 source_name: "orders",
6806 table_uri: &table_uri,
6807 snapshot_version: 42,
6808 storage_options: &options,
6809 })?;
6810 let captured_options = captured_storage_options(&captured);
6811 assert_eq!(captured_options.len(), 2);
6812 assert!(captured_options.iter().all(|captured| captured == &options));
6813
6814 Ok(())
6815 }
6816
6817 #[test]
6818 fn reader_construction_passes_effective_s3_storage_options_to_each_store_construction()
6819 -> Result<(), Box<dyn std::error::Error>> {
6820 let scheme = unique_storage_scheme("readereffectiveoptions")?;
6821 let captured = CapturedStorageOptions::default();
6822 register_capturing_storage_handler(&scheme, Arc::clone(&captured))?;
6823 let table_uri = format!("{scheme}://table/root/");
6824 let effective = effective_storage_options_for_source_from_env(
6825 "s3://bucket/root",
6826 storage_options(&[("region", "caller-region")]),
6827 env_vars(&[
6828 ("AWS_ACCESS_KEY_ID", "env-access"),
6829 ("AWS_SECRET_ACCESS_KEY", "env-secret"),
6830 ("AWS_REGION", "env-region"),
6831 ]),
6832 );
6833
6834 let _data_reader = KernelDataFileReader::try_new(KernelDataFileReaderConfig {
6835 source_name: "orders",
6836 table_uri: &table_uri,
6837 snapshot_version: 42,
6838 storage_options: &effective,
6839 })?;
6840 let _dv_reader = KernelDeletionVectorReader::try_new(KernelDeletionVectorReaderConfig {
6841 source_name: "orders",
6842 table_uri: &table_uri,
6843 snapshot_version: 42,
6844 storage_options: &effective,
6845 })?;
6846 let captured_options = captured_storage_options(&captured);
6847 assert_eq!(captured_options.len(), 2);
6848 assert!(
6849 captured_options
6850 .iter()
6851 .all(|captured| captured == &effective)
6852 );
6853
6854 Ok(())
6855 }
6856
6857 #[test]
6858 fn storage_option_values_are_not_exposed_when_store_construction_fails()
6859 -> Result<(), Box<dyn std::error::Error>> {
6860 for secret_key in [
6861 "AWS_SECRET_ACCESS_KEY",
6862 "aws_secret_access_key",
6863 "AWS_SESSION_TOKEN",
6864 "aws_session_token",
6865 ] {
6866 let scheme = unique_storage_scheme("redactedoptions")?;
6867 let captured = CapturedStorageOptions::default();
6868 let secret_value = "super-secret-token-value";
6869 register_failing_storage_handler(&scheme, Arc::clone(&captured), secret_value)?;
6870 let options =
6871 storage_options(&[(secret_key, secret_value), ("AWS_REGION", "us-west-2")]);
6872
6873 let result = load_delta_source(
6874 DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
6875 .with_storage_options(options.clone()),
6876 );
6877 let error = result
6878 .err()
6879 .map(|error| error.to_string())
6880 .unwrap_or_default();
6881
6882 assert_eq!(captured_storage_options(&captured), vec![options]);
6883 assert!(!error.contains(secret_value), "{secret_key}");
6884 assert!(!error.contains(secret_key), "{secret_key}");
6885 }
6886
6887 Ok(())
6888 }
6889
6890 #[test]
6891 fn merged_s3_env_storage_option_values_are_not_exposed_when_store_construction_fails()
6892 -> Result<(), Box<dyn std::error::Error>> {
6893 let scheme = unique_storage_scheme("redacteds3envoptions")?;
6894 let captured = CapturedStorageOptions::default();
6895 let secret_value = "super-secret-env-token-value";
6896 register_failing_storage_handler(&scheme, Arc::clone(&captured), secret_value)?;
6897 let effective = effective_storage_options_for_source_from_env(
6898 "s3://bucket/root",
6899 DeltaStorageOptions::default(),
6900 env_vars(&[
6901 ("AWS_ACCESS_KEY_ID", "env-access"),
6902 ("AWS_SECRET_ACCESS_KEY", secret_value),
6903 ("AWS_REGION", "us-east-1"),
6904 ]),
6905 );
6906
6907 let result = load_delta_source(
6908 DeltaSourceConfig::new("orders", format!("{scheme}://table/root"))
6909 .with_storage_options(effective.clone()),
6910 );
6911 let error = result
6912 .err()
6913 .map(|error| error.to_string())
6914 .unwrap_or_default();
6915
6916 assert_eq!(captured_storage_options(&captured), vec![effective]);
6917 assert!(!error.contains(secret_value));
6918 assert!(!error.contains("AWS_SECRET_ACCESS_KEY"));
6919
6920 Ok(())
6921 }
6922
6923 #[test]
6924 fn invalid_recognized_s3_aws_env_values_fail_store_construction() {
6925 let storage_options = effective_storage_options_for_source_from_env(
6926 "s3://bucket/root",
6927 DeltaStorageOptions::default(),
6928 env_vars(&[
6929 ("AWS_ALLOW_HTTP", "not-a-bool"),
6930 ("AWS_REGION", "us-east-1"),
6931 ]),
6932 );
6933
6934 let result = load_delta_table_snapshot("s3://bucket/root", None, &storage_options);
6935
6936 assert!(matches!(
6937 result,
6938 Err(DeltaFunnelError::DeltaSourceEngine { .. })
6939 ));
6940 }
6941
6942 #[test]
6943 fn loads_named_delta_source_at_fixed_version() -> Result<(), Box<dyn std::error::Error>> {
6944 let table = DeltaLogTable::new("fixed")?;
6945
6946 let source = load_delta_source(DeltaSourceConfig {
6947 name: "orders".to_owned(),
6948 table_uri: table.path.to_string_lossy().to_string(),
6949 version: Some(0),
6950 storage_options: Default::default(),
6951 })?;
6952
6953 assert_eq!(source.version(), 0);
6954
6955 Ok(())
6956 }
6957
6958 #[test]
6959 fn validates_name_before_table_uri() {
6960 let result = load_delta_source(DeltaSourceConfig {
6961 name: "orders.latest".to_owned(),
6962 table_uri: "missing/path".to_owned(),
6963 version: None,
6964 storage_options: Default::default(),
6965 });
6966
6967 assert!(matches!(
6968 result,
6969 Err(DeltaFunnelError::InvalidSourceName { .. })
6970 ));
6971 }
6972
6973 #[test]
6974 fn rejects_sql_keyword_name_before_table_uri() {
6975 let result = load_delta_source(DeltaSourceConfig {
6976 name: "select".to_owned(),
6977 table_uri: "missing/path".to_owned(),
6978 version: None,
6979 storage_options: Default::default(),
6980 });
6981
6982 assert!(matches!(
6983 result,
6984 Err(DeltaFunnelError::InvalidSourceName { name, reason })
6985 if name == "select" && reason == "source names must not be SQL keywords"
6986 ));
6987 }
6988
6989 #[test]
6990 fn rejects_blank_table_uri_as_invalid_source_uri() {
6991 let result = load_delta_source(DeltaSourceConfig {
6992 name: "orders".to_owned(),
6993 table_uri: " \t\n".to_owned(),
6994 version: None,
6995 storage_options: Default::default(),
6996 });
6997
6998 assert!(matches!(
6999 result,
7000 Err(DeltaFunnelError::InvalidSourceUri { reason })
7001 if reason == "table location must not be empty"
7002 ));
7003 }
7004
7005 #[test]
7006 fn loads_multiple_named_delta_sources() -> Result<(), Box<dyn std::error::Error>> {
7007 let orders = DeltaLogTable::new("orders")?;
7008 let customers = DeltaLogTable::new("customers")?;
7009
7010 let sources = load_delta_sources([
7011 DeltaSourceConfig {
7012 name: "orders".to_owned(),
7013 table_uri: orders.path.to_string_lossy().to_string(),
7014 version: None,
7015 storage_options: Default::default(),
7016 },
7017 DeltaSourceConfig {
7018 name: "customers".to_owned(),
7019 table_uri: customers.path.to_string_lossy().to_string(),
7020 version: Some(0),
7021 storage_options: Default::default(),
7022 },
7023 ])?;
7024
7025 assert_eq!(sources.len(), 2);
7026 assert_eq!(sources[0].name(), "orders");
7027 assert_eq!(sources[0].version(), 1);
7028 assert_eq!(sources[1].name(), "customers");
7029 assert_eq!(sources[1].version(), 0);
7030
7031 Ok(())
7032 }
7033
7034 #[test]
7035 fn rejects_duplicate_names_before_table_uri_loading() {
7036 let result = load_delta_sources([
7037 DeltaSourceConfig {
7038 name: "orders".to_owned(),
7039 table_uri: "missing/orders".to_owned(),
7040 version: None,
7041 storage_options: Default::default(),
7042 },
7043 DeltaSourceConfig {
7044 name: "Orders".to_owned(),
7045 table_uri: "missing/customers".to_owned(),
7046 version: None,
7047 storage_options: Default::default(),
7048 },
7049 ]);
7050
7051 assert!(matches!(
7052 result,
7053 Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "Orders"
7054 ));
7055 }
7056
7057 #[test]
7058 fn rejects_invalid_name_before_multi_source_table_uri_loading() {
7059 let result = load_delta_sources([
7060 DeltaSourceConfig {
7061 name: "orders".to_owned(),
7062 table_uri: "missing/orders".to_owned(),
7063 version: None,
7064 storage_options: Default::default(),
7065 },
7066 DeltaSourceConfig {
7067 name: "customers.latest".to_owned(),
7068 table_uri: "missing/customers".to_owned(),
7069 version: None,
7070 storage_options: Default::default(),
7071 },
7072 ]);
7073
7074 assert!(matches!(
7075 result,
7076 Err(DeltaFunnelError::InvalidSourceName { name, .. }) if name == "customers.latest"
7077 ));
7078 }
7079
7080 #[test]
7081 fn source_load_errors_do_not_expose_secret_bearing_uri() {
7082 let result = load_delta_source(DeltaSourceConfig {
7083 name: "orders".to_owned(),
7084 table_uri: "ftp://user:password@example.com/table".to_owned(),
7085 version: None,
7086 storage_options: Default::default(),
7087 });
7088
7089 let error = result
7090 .err()
7091 .map(|error| error.to_string())
7092 .unwrap_or_default();
7093
7094 assert!(!error.contains("user"));
7095 assert!(!error.contains("password"));
7096 assert!(!error.contains("example.com"));
7097 assert!(!error.contains("ftp://"));
7098 }
7099}