// SPDX-License-Identifier: Apache-2.0

// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership

// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

#![doc(

html_logo_url = "https://raw.githubusercontent.com/apache/datafusion/19fe44cf2f30cbdd63d4a4f52c74055163c6cc38/docs/logos/standalone_logo/logo_original.svg",

html_favicon_url = "https://raw.githubusercontent.com/apache/datafusion/19fe44cf2f30cbdd63d4a4f52c74055163c6cc38/docs/logos/standalone_logo/logo_original.svg"

)]

#![cfg_attr(docsrs, feature(doc_cfg))]

// Make sure fast / cheap clones on Arc are explicit:

// https://github.com/apache/datafusion/issues/11143

#![cfg_attr(not(test), deny(clippy::clone_on_ref_ptr))]

#![cfg_attr(test, allow(clippy::needless_pass_by_value))]

//! A table that uses the `ObjectStore` listing capability

//! to get the list of files to process.

pub mod boundary_stream;

pub mod decoder;

pub mod display;

pub mod file;

pub mod file_compression_type;

pub mod file_format;

pub mod file_groups;

pub mod file_scan_config;

pub mod file_sink_config;

pub mod file_stream;

pub mod memory;

pub mod morsel;

pub mod projection;

/// Protobuf conversions for [`FileRange`], [`PartitionedFile`] and

/// [`FileGroup`], gated on the `proto` feature.

#[cfg(feature = "proto")]

mod proto;

pub mod schema_adapter;

pub mod sink;

pub mod source;

mod statistics;

pub mod table_schema;

#[cfg(test)]

pub mod test_util;

pub mod url;

pub mod write;

pub use self::file::as_file_source;

pub use self::url::ListingTableUrl;

use crate::file_groups::FileGroup;

use arrow::datatypes::SchemaRef;

use chrono::TimeZone;

use datafusion_common::stats::{Precision, is_known_empty};

use datafusion_common::{ColumnStatistics, Result, TableReference};

use datafusion_common::{ScalarValue, Statistics};

use datafusion_physical_expr::LexOrdering;

use futures::Stream;

use object_store::{ObjectMeta, path::Path};

pub use statistics::compute_all_files_statistics;

use std::any::Any;

use std::pin::Pin;

use std::sync::Arc;

pub use table_schema::{TableSchema, TableSchemaBuilder};

/// User-defined per-file extension data, keyed by concrete Rust type.

///

/// Re-exported from [`datafusion_common::extensions::Extensions`]; the same

/// type backs `SessionConfig::extensions`, `ExtendedStatistics::extensions`,

/// and other extension fields throughout DataFusion.

pub type FileExtensions = datafusion_common::extensions::Extensions;

/// Stream of files get listed from object store

#[deprecated(

since = "54.0.0",

note = "This type is unused and will be removed in a future release"

)]

pub type PartitionedFileStream =

Pin<Box<dyn Stream<Item = Result<PartitionedFile>> + Send + Sync + 'static>>;

/// A `[start, end)` byte range of a file. Non-overlapping ranges can be scanned

/// in parallel. For Parquet, [`RowGroupRangeAssignment`] picks the row groups a

/// range reads.

///

/// [`RowGroupRangeAssignment`]: datafusion_common::parquet_config::RowGroupRangeAssignment

#[derive(Debug, Clone, PartialEq, Hash, Eq, PartialOrd, Ord)]

pub struct FileRange {

/// Range start

pub start: i64,

/// Range end

pub end: i64,

}

impl FileRange {

/// returns true if this file range contains the specified offset

pub fn contains(&self, offset: i64) -> bool {

offset >= self.start && offset < self.end

}

}

#[derive(Debug, Clone)]

/// A single file or part of a file that should be read, along with its schema, statistics

/// and partition column values that need to be appended to each row.

///

/// # Statistics

///

/// The [`Self::statistics`] field contains statistics for the **full table schema**,

/// which includes both file columns and partition columns. When statistics are set via

/// [`Self::with_statistics`], exact statistics for partition columns are automatically

/// computed from [`Self::partition_values`]:

///

/// - `min = max = partition_value` (all rows in a file share the same partition value)

/// - `null_count = 0` (every row has the file's partition value)

/// - `distinct_count = 1` (single distinct value per file for each partition column)

///

/// A null partition value instead gives `null_count = num_rows`, `distinct_count = 0`

/// and absent `min`/`max`.

///

/// This enables query optimizers to use partition column bounds for pruning and planning.

pub struct PartitionedFile {

/// Path for the file (e.g. URL, filesystem path, etc)

pub object_meta: ObjectMeta,

/// Values of partition columns to be appended to each row.

///

/// These MUST have the same count, order, and type than the [`table_partition_cols`].

///

/// You may use [`wrap_partition_value_in_dict`] to wrap them if you have used [`wrap_partition_type_in_dict`] to wrap the column type.

///

///

/// [`wrap_partition_type_in_dict`]: crate::file_scan_config::wrap_partition_type_in_dict

/// [`wrap_partition_value_in_dict`]: crate::file_scan_config::wrap_partition_value_in_dict

/// [`table_partition_cols`]: https://github.com/apache/datafusion/blob/main/datafusion/core/src/datasource/file_format/options.rs#L87

pub partition_values: Vec<ScalarValue>,

/// An optional file range for a more fine-grained parallel execution

pub range: Option<FileRange>,

/// Optional statistics that describe the data in this file if known.

///

/// DataFusion relies on these statistics for planning (in particular to sort file groups),

/// so if they are incorrect, incorrect answers may result.

///

/// These statistics cover the full table schema: file columns plus partition columns.

/// When set via [`Self::with_statistics`], partition column statistics are automatically

/// computed from [`Self::partition_values`] with exact min/max/null_count/distinct_count.

pub statistics: Option<Arc<Statistics>>,

/// The known lexicographical ordering of the rows in this file, if any.

///

/// This describes how the data within the file is sorted with respect to one or more

/// columns, and is used by the optimizer for planning operations that depend on input

/// ordering (e.g. merges, sorts, and certain aggregations).

///

/// When available, this is typically inferred from file-level metadata exposed by the

/// underlying format (for example, Parquet `sorting_columns`), but it may also be set

/// explicitly via [`Self::with_ordering`].

pub ordering: Option<LexOrdering>,

/// User-defined per-file metadata, keyed by Rust type. Multiple

/// independent components can each attach their own data here without

/// conflict — see [`FileExtensions`].

pub extensions: FileExtensions,

/// The estimated size of the parquet metadata, in bytes

pub metadata_size_hint: Option<usize>,

pub table_reference: Option<TableReference>,

/// A user-provided physical Arrow schema for this file.

///

/// This schema describes only the columns stored in the file. It must not

/// include partition columns; those are represented separately by

/// [`Self::partition_values`] and the scan's table partition columns.

///

/// When provided, this field will be used by the Parquet reader to avoid

/// parsing the Arrow schema from the `ARROW:schema` metadata key. Other

/// built-in file sources ignore it for now.

pub arrow_schema: Option<SchemaRef>,

}

impl PartitionedFile {

/// Create a simple file without metadata or partition

pub fn new(path: impl Into<String>, size: u64) -> Self {

Self {

arrow_schema: None,

object_meta: ObjectMeta {

location: Path::from(path.into()),

last_modified: chrono::Utc.timestamp_nanos(0),

size,

e_tag: None,

version: None,

},

partition_values: vec![],

range: None,

statistics: None,

ordering: None,

extensions: FileExtensions::new(),

metadata_size_hint: None,

table_reference: None,

}

}

/// Create a file from a known ObjectMeta without partition

pub fn new_from_meta(object_meta: ObjectMeta) -> Self {

Self {

arrow_schema: None,

object_meta,

partition_values: vec![],

range: None,

statistics: None,

ordering: None,

extensions: FileExtensions::new(),

metadata_size_hint: None,

table_reference: None,

}

}

/// Create a file range without metadata or partition

pub fn new_with_range(path: String, size: u64, start: i64, end: i64) -> Self {

Self {

arrow_schema: None,

object_meta: ObjectMeta {

location: Path::from(path),

last_modified: chrono::Utc.timestamp_nanos(0),

size,

e_tag: None,

version: None,

},

partition_values: vec![],

range: Some(FileRange { start, end }),

statistics: None,

ordering: None,

extensions: FileExtensions::new(),

metadata_size_hint: None,

table_reference: None,

}

.with_range(start, end)

}

/// Provide a physical Arrow schema for this file.

///

/// The schema must describe only columns stored in the file and must not

/// include partition columns. See [`Self::arrow_schema`] for details.

pub fn with_arrow_schema(mut self, schema: SchemaRef) -> Self {

self.arrow_schema = Some(schema);

self

}

/// Attach partition values to this file.

/// This replaces any existing partition values.

pub fn with_partition_values(mut self, partition_values: Vec<ScalarValue>) -> Self {

self.partition_values = partition_values;

self

}

pub fn with_table_reference(

mut self,

table_reference: Option<TableReference>,

) -> Self {

self.table_reference = table_reference;

self

}

/// Size of the file to be scanned (taking into account the range, if present).

pub fn effective_size(&self) -> u64 {

if let Some(range) = &self.range {

(range.end - range.start) as u64

} else {

self.object_meta.size

}

}

/// Effective range of the file to be scanned.

pub fn range(&self) -> (u64, u64) {

if let Some(range) = &self.range {

(range.start as u64, range.end as u64)

} else {

(0, self.object_meta.size)

}

}

/// Provide a hint to the size of the file metadata. If a hint is provided

/// the reader will try and fetch the last `size_hint` bytes of the parquet file optimistically.

/// Without an appropriate hint, two read may be required to fetch the metadata.

pub fn with_metadata_size_hint(mut self, metadata_size_hint: usize) -> Self {

self.metadata_size_hint = Some(metadata_size_hint);

self

}

/// Return a file reference from the given path

pub fn from_path(path: String) -> Result<Self> {

let size = std::fs::metadata(path.clone())?.len();

Ok(Self::new(path, size))

}

/// Return the path of this partitioned file

pub fn path(&self) -> &Path {

&self.object_meta.location

}

/// Update the file to only scan the specified range (in bytes)

pub fn with_range(mut self, start: i64, end: i64) -> Self {

self.range = Some(FileRange { start, end });

self

}

/// Attach a typed user-defined extension to this file. Multiple

/// independent extensions can be attached, each keyed by its concrete

/// Rust type. Inserting a value of a type that already has an extension

/// replaces the previous one.

///

/// This can be used to pass reader-specific information (e.g. a

/// `ParquetAccessPlan`, or a custom index entry).

pub fn with_extension<T: Any + Send + Sync>(mut self, value: T) -> Self {

self.extensions.insert(value);

self

}

/// Borrow the extension of type `T`, if one is attached.

pub fn extension<T: Any + Send + Sync>(&self) -> Option<&T> {

self.extensions.get::<T>()

}

/// Attach a type-erased extension to this file.

///

/// Kept as a backwards-compatible shim; prefer [`Self::with_extension`]

/// which keys the extension by its concrete Rust type at the call site.

#[deprecated(

since = "54.0.0",

note = "use `with_extension`; the extension is keyed by its concrete type"

)]

pub fn with_extensions(mut self, extensions: Arc<dyn Any + Send + Sync>) -> Self {

#[expect(deprecated)]

self.extensions.insert_dyn(extensions);

self

}

/// Update the statistics for this file.

///

/// The provided `statistics` should cover only the file schema columns.

/// This method will automatically append exact statistics for partition columns

/// based on `partition_values`:

/// - `min = max = partition_value` (all rows have the same value)

/// - `null_count = 0` (all rows have a non-null value)

/// - `distinct_count = 1` (all rows have the same partition value)

///

/// A null partition value (e.g. a Hive default partition) instead gives

/// `null_count = num_rows`, `distinct_count = 0` and absent `min`/`max`.

pub fn with_statistics(mut self, file_statistics: Arc<Statistics>) -> Self {

if self.partition_values.is_empty() {

// No partition columns, use stats as-is

self.statistics = Some(file_statistics);

} else {

// Extend stats with exact partition column statistics

let mut stats = Arc::unwrap_or_clone(file_statistics);

for partition_value in &self.partition_values {

let (null_count, min_max, distinct_count) = if partition_value.is_null() {

(stats.num_rows, Precision::Absent, Precision::Exact(0))

} else {

(

Precision::Exact(0),

Precision::Exact(partition_value.clone()),

Precision::Exact(1),

)

};

let col_stats = ColumnStatistics {

null_count,

max_value: min_max.clone(),

min_value: min_max,

distinct_count,

sum_value: Precision::Absent,

byte_size: partition_value

.data_type()

.primitive_width()

.map(|w| stats.num_rows.multiply(&Precision::Exact(w)))

.unwrap_or_else(|| Precision::Absent),

};

stats.column_statistics.push(col_stats);

}

self.statistics = Some(Arc::new(stats));

}

self

}

/// Check if this file has any statistics.

/// This returns `true` if the file has any Exact or Inexact statistics

/// and `false` if all statistics are `Precision::Absent`.

pub fn has_statistics(&self) -> bool {

if let Some(stats) = &self.statistics {

stats.column_statistics.iter().any(|col_stats| {

col_stats.null_count != Precision::Absent

|| col_stats.max_value != Precision::Absent

|| col_stats.min_value != Precision::Absent

|| col_stats.sum_value != Precision::Absent

|| col_stats.distinct_count != Precision::Absent

})

} else {

false

}

}

/// Set the known ordering of data in this file.

///

/// The ordering represents the lexicographical sort order of the data,

/// typically inferred from file metadata (e.g., Parquet sorting_columns).

pub fn with_ordering(mut self, ordering: Option<LexOrdering>) -> Self {

self.ordering = ordering;

self

}

}

impl From<ObjectMeta> for PartitionedFile {

fn from(object_meta: ObjectMeta) -> Self {

PartitionedFile {

object_meta,

arrow_schema: None,

partition_values: vec![],

range: None,

statistics: None,

ordering: None,

extensions: FileExtensions::new(),

metadata_size_hint: None,

table_reference: None,

}

}

}

/// Generates test files with min-max statistics in different overlap patterns.

///

/// # Overlap Factors

///

/// The `overlap_factor` parameter controls how much the value ranges in generated test files overlap:

/// - `0.0`: No overlap between files (completely disjoint ranges)

/// - `0.2`: Low overlap (20% of the range size overlaps with adjacent files)

/// - `0.5`: Medium overlap (50% of ranges overlap)

/// - `0.8`: High overlap (80% of ranges overlap between files)

///

/// # Examples

///

/// With 5 files and different overlap factors showing `[min, max]` ranges:

///

/// overlap_factor = 0.0 (no overlap):

///

/// File 0: [0, 20]

/// File 1: [20, 40]

/// File 2: [40, 60]

/// File 3: [60, 80]

/// File 4: [80, 100]

///

/// overlap_factor = 0.5 (50% overlap):

///

/// File 0: [0, 40]

/// File 1: [20, 60]

/// File 2: [40, 80]

/// File 3: [60, 100]

/// File 4: [80, 120]

///

/// overlap_factor = 0.8 (80% overlap):

///

/// File 0: [0, 100]

/// File 1: [20, 120]

/// File 2: [40, 140]

/// File 3: [60, 160]

/// File 4: [80, 180]

///

/// # Public Only for Internal Use:

///

/// This is not a public API and is for internal use only; see [API policy] for details.

///

/// [API policy]: https://datafusion.apache.org/contributor-guide/api-health.html#internal-public-apis

#[doc(hidden)]

pub fn generate_test_files(num_files: usize, overlap_factor: f64) -> Vec<FileGroup> {

let mut files = Vec::with_capacity(num_files);

if num_files == 0 {

return vec![];

}

let range_size = if overlap_factor == 0.0 {

100 / num_files as i64

} else {

(100.0 / (overlap_factor * num_files as f64)).max(1.0) as i64

};

for i in 0..num_files {

let base = (i as f64 * range_size as f64 * (1.0 - overlap_factor)) as i64;

let min = base as f64;

let max = (base + range_size) as f64;

let file = PartitionedFile {

arrow_schema: None,

object_meta: ObjectMeta {

location: Path::from(format!("file_{i}.parquet")),

last_modified: chrono::Utc::now(),

size: 1000,

e_tag: None,

version: None,

},

partition_values: vec![],

range: None,

statistics: Some(Arc::new(Statistics {

num_rows: Precision::Exact(100),

total_byte_size: Precision::Exact(1000),

column_statistics: vec![ColumnStatistics {

null_count: Precision::Exact(0),

max_value: Precision::Exact(ScalarValue::Float64(Some(max))),

min_value: Precision::Exact(ScalarValue::Float64(Some(min))),

sum_value: Precision::Absent,

distinct_count: Precision::Absent,

byte_size: Precision::Absent,

}],

})),

ordering: None,

extensions: FileExtensions::new(),

metadata_size_hint: None,

table_reference: None,

};

files.push(file);

}

vec![FileGroup::new(files)]

}

/// Verifies that files within each group maintain sort order.

///

/// # Public Only for Internal Use:

///

/// This is not a public API and is for internal use only; see [API policy] for details.

///

/// [API policy]: https://datafusion.apache.org/contributor-guide/api-health.html#internal-public-apis

#[doc(hidden)]

pub fn verify_sort_integrity(file_groups: &[FileGroup]) -> bool {

for group in file_groups {

// Known-empty files contribute no rows and may not have min/max

// statistics, so they cannot violate the ordering.

let files = group

.iter()

.filter(|file| !file.statistics.as_deref().is_some_and(is_known_empty))

.collect::<Vec<_>>();

for i in 1..files.len() {

let prev_file = files[i - 1];

let curr_file = files[i];

// Check if the min value of current file is greater than max value of previous file

if let (Some(prev_stats), Some(curr_stats)) =

(&prev_file.statistics, &curr_file.statistics)

{

let prev_max = &prev_stats.column_statistics[0].max_value;

let curr_min = &curr_stats.column_statistics[0].min_value;

if curr_min.get_value().unwrap() <= prev_max.get_value().unwrap() {

return false;

}

}

}

}

true

}

#[cfg(test)]

mod tests {

use super::ListingTableUrl;

use arrow::{

array::{ArrayRef, Int32Array, RecordBatch},

datatypes::{DataType, Field, Schema, SchemaRef},

};

use datafusion_execution::object_store::{

DefaultObjectStoreRegistry, ObjectStoreRegistry,

};

use object_store::{local::LocalFileSystem, path::Path};

use std::{collections::HashMap, ops::Not, sync::Arc};

use url::Url;

/// Return a RecordBatch with a single Int32 array with values (0..sz) in a field named "i"

pub fn make_partition(sz: i32) -> RecordBatch {

let seq_start = 0;

let seq_end = sz;

let values = (seq_start..seq_end).collect::<Vec<_>>();

let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));

let arr = Arc::new(Int32Array::from(values));

RecordBatch::try_new(schema, vec![arr as ArrayRef]).unwrap()

}

/// Get the schema for the aggregate_test_* csv files

pub fn aggr_test_schema() -> SchemaRef {

let mut f1 = Field::new("c1", DataType::Utf8, false);

f1.set_metadata(HashMap::from_iter(vec![("testing".into(), "test".into())]));

let schema = Schema::new(vec![

f1,

Field::new("c2", DataType::UInt32, false),

Field::new("c3", DataType::Int8, false),

Field::new("c4", DataType::Int16, false),

Field::new("c5", DataType::Int32, false),

Field::new("c6", DataType::Int64, false),

Field::new("c7", DataType::UInt8, false),

Field::new("c8", DataType::UInt16, false),

Field::new("c9", DataType::UInt32, false),

Field::new("c10", DataType::UInt64, false),

Field::new("c11", DataType::Float32, false),

Field::new("c12", DataType::Float64, false),

Field::new("c13", DataType::Utf8, false),

]);

Arc::new(schema)

}

#[test]

fn test_object_store_listing_url() {

let listing = ListingTableUrl::parse("file:///").unwrap();

let store = listing.object_store();

assert_eq!(store.as_str(), "file:///");

let listing = ListingTableUrl::parse("s3://bucket/").unwrap();

let store = listing.object_store();

assert_eq!(store.as_str(), "s3://bucket/");

}

#[test]

fn test_get_store_hdfs() {

let sut = DefaultObjectStoreRegistry::default();

let url = Url::parse("hdfs://localhost:8020").unwrap();

sut.register_store(&url, Arc::new(LocalFileSystem::new()));

let url = ListingTableUrl::parse("hdfs://localhost:8020/key").unwrap();

sut.get_store(url.as_ref()).unwrap();

}

#[test]

fn test_get_store_s3() {

let sut = DefaultObjectStoreRegistry::default();

let url = Url::parse("s3://bucket/key").unwrap();

sut.register_store(&url, Arc::new(LocalFileSystem::new()));

let url = ListingTableUrl::parse("s3://bucket/key").unwrap();

sut.get_store(url.as_ref()).unwrap();

}

#[test]

fn test_get_store_file() {

let sut = DefaultObjectStoreRegistry::default();

let url = ListingTableUrl::parse("file:///bucket/key").unwrap();

sut.get_store(url.as_ref()).unwrap();

}

#[test]

fn test_get_store_local() {

let sut = DefaultObjectStoreRegistry::default();

let url = ListingTableUrl::parse("../").unwrap();

sut.get_store(url.as_ref()).unwrap();

}

#[test]

fn test_with_statistics_appends_partition_column_stats() {

use crate::PartitionedFile;

use datafusion_common::stats::Precision;

use datafusion_common::{ColumnStatistics, ScalarValue, Statistics};

// Create a PartitionedFile with partition values

let mut pf = PartitionedFile::new(

"test.parquet",

100, // file size

);

pf.partition_values = vec![

ScalarValue::Date32(Some(20148)), // 2025-03-01

];

// Create file-only statistics (1 column for 'id')

let file_stats = Arc::new(Statistics {

num_rows: Precision::Exact(2),

total_byte_size: Precision::Exact(16),

column_statistics: vec![ColumnStatistics {

null_count: Precision::Exact(0),

max_value: Precision::Exact(ScalarValue::Int32(Some(4))),

min_value: Precision::Exact(ScalarValue::Int32(Some(3))),

sum_value: Precision::Absent,

distinct_count: Precision::Absent,

byte_size: Precision::Absent,

}],

});

// Call with_statistics - should append partition column stats

let pf = pf.with_statistics(file_stats);

// Verify the statistics now have 2 columns

let stats = pf.statistics.unwrap();

assert_eq!(

stats.column_statistics.len(),

2,

"Expected 2 columns (id + date partition)"

);

// Verify partition column statistics

let partition_col_stats = &stats.column_statistics[1];

assert_eq!(

partition_col_stats.null_count,

Precision::Exact(0),

"Partition column null_count should be Exact(0)"

);

assert_eq!(

partition_col_stats.min_value,

Precision::Exact(ScalarValue::Date32(Some(20148))),

"Partition column min should match partition value"

);

assert_eq!(

partition_col_stats.max_value,

Precision::Exact(ScalarValue::Date32(Some(20148))),

"Partition column max should match partition value"

);

assert_eq!(

partition_col_stats.distinct_count,

Precision::Exact(1),

"Partition column distinct_count should be Exact(1)"

);

}

#[test]

fn test_with_statistics_null_partition_value() {

use crate::PartitionedFile;

use datafusion_common::stats::Precision;

use datafusion_common::{ScalarValue, Statistics};

let mut pf = PartitionedFile::new("test.parquet", 100);

pf.partition_values = vec![ScalarValue::Date32(None)];

let file_stats = Arc::new(Statistics {

num_rows: Precision::Exact(2),

total_byte_size: Precision::Exact(16),

column_statistics: vec![],

});

let stats = pf.with_statistics(file_stats).statistics.unwrap();

let partition_col_stats = &stats.column_statistics[0];

assert_eq!(partition_col_stats.null_count, Precision::Exact(2));

assert_eq!(partition_col_stats.min_value, Precision::Absent);

assert_eq!(partition_col_stats.max_value, Precision::Absent);

assert_eq!(partition_col_stats.distinct_count, Precision::Exact(0));

}

#[test]

fn test_url_contains() {

let url = ListingTableUrl::parse("file:///var/data/mytable/").unwrap();

// standard case with default config

assert!(url.contains(

&Path::parse("/var/data/mytable/data.parquet").unwrap(),

true

));

// standard case with `ignore_subdirectory` set to false

assert!(url.contains(

&Path::parse("/var/data/mytable/data.parquet").unwrap(),

false

));

// as per documentation, when `ignore_subdirectory` is true, we should ignore files that aren't

// a direct child of the `url`

assert!(

url.contains(

&Path::parse("/var/data/mytable/mysubfolder/data.parquet").unwrap(),

true

)

.not()

);

// when we set `ignore_subdirectory` to false, we should not ignore the file

assert!(url.contains(

&Path::parse("/var/data/mytable/mysubfolder/data.parquet").unwrap(),

false

));

// as above, `ignore_subdirectory` is false, so we include the file

assert!(url.contains(

&Path::parse("/var/data/mytable/year=2024/data.parquet").unwrap(),

false

));

// in this case, we include the file even when `ignore_subdirectory` is true because the

// path segment is a hive partition which doesn't count as a subdirectory for the purposes

// of `Url::contains`

assert!(url.contains(

&Path::parse("/var/data/mytable/year=2024/data.parquet").unwrap(),

true

));

// testing an empty path with default config

assert!(url.contains(&Path::parse("/var/data/mytable/").unwrap(), true));

// testing an empty path with `ignore_subdirectory` set to false

assert!(url.contains(&Path::parse("/var/data/mytable/").unwrap(), false));

}

}