datafusion_datasource/

mod.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#![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`](crate::file_groups::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;
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>>;

/// Only scan a subset of Row Groups from the Parquet file whose data "midpoint"
/// lies within the [start, end) byte offsets. This option can be used to scan non-overlapping
/// sections of a Parquet file in parallel.
#[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` (partition values extracted from paths are never null)
/// - `distinct_count = 1` (single distinct value per file for each partition column)
///
/// 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` (partition values from paths are never null)
    /// - `distinct_count = 1` (all rows have the same partition value)
    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 col_stats = ColumnStatistics {
                    null_count: Precision::Exact(0),
                    max_value: Precision::Exact(partition_value.clone()),
                    min_value: Precision::Exact(partition_value.clone()),
                    distinct_count: Precision::Exact(1),
                    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.
///
/// Used by tests and benchmarks.
///
/// # 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]
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)]
}

// Helper function to verify that files within each group maintain sort order
/// Used by tests and benchmarks
pub fn verify_sort_integrity(file_groups: &[FileGroup]) -> bool {
    for group in file_groups {
        let files = group.iter().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_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));
    }
}