Support for metadata columns (location, size, last_modified) in ListingTableProvider (#74)

UPSTREAM NOTE: This PR was attempted to be upstreamed but was not accepted. Needs to be applied manually
apache#15181

Author: phillipleblanc

Cargo.lock

@@ -495,7 +495,7 @@ impl TableProvider for IndexTableProvider {
             ParquetSource::default()
                 // provide the predicate so the DataSourceExec can try and prune
                 // row groups internally
-                .with_predicate(predicate)
+                .with_predicate(Arc::clone(&schema), predicate)
                 // provide the factory to create parquet reader without re-reading metadata
                 .with_parquet_file_reader_factory(Arc::new(reader_factory)),
         );

datafusion-examples/examples/advanced_parquet_index.rs

@@ -21,7 +21,6 @@ use arrow::{
     array::{AsArray, RecordBatch, StringArray, UInt8Array},
     datatypes::{DataType, Field, Schema, SchemaRef, UInt64Type},
 };
-use datafusion::physical_expr::LexRequirement;
 use datafusion::{
     catalog::Session,
     common::{GetExt, Statistics},
@@ -42,6 +41,7 @@ use datafusion::{
     physical_plan::ExecutionPlan,
     prelude::SessionContext,
 };
+use datafusion::{physical_expr::LexRequirement, physical_plan::PhysicalExpr};
 use object_store::{ObjectMeta, ObjectStore};
 use tempfile::tempdir;
 
@@ -111,8 +111,11 @@ impl FileFormat for TSVFileFormat {
         &self,
         state: &dyn Session,
         conf: FileScanConfig,
+        filters: Option<&Arc<dyn PhysicalExpr>>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        self.csv_file_format.create_physical_plan(state, conf).await
+        self.csv_file_format
+            .create_physical_plan(state, conf, filters)
+            .await
     }
 
     async fn create_writer_physical_plan(

datafusion-examples/examples/custom_file_format.rs

@@ -243,7 +243,8 @@ impl TableProvider for IndexTableProvider {
         let files = self.index.get_files(predicate.clone())?;
 
         let object_store_url = ObjectStoreUrl::parse("file://")?;
-        let source = Arc::new(ParquetSource::default().with_predicate(predicate));
+        let source =
+            Arc::new(ParquetSource::default().with_predicate(self.schema(), predicate));
         let mut file_scan_config_builder =
             FileScanConfigBuilder::new(object_store_url, self.schema(), source)
                 .with_projection(projection.cloned())

datafusion-examples/examples/parquet_index.rs

@@ -48,6 +48,8 @@ object_store = { workspace = true }
 tokio = { workspace = true }
 
 [dev-dependencies]
+chrono = { workspace = true }
+tempfile = { workspace = true }
 
 [lints]
 workspace = true