Fix up more doc tests

Author: friendlymatthew

datafusion-examples/examples/default_column_values.rs

@@ -262,12 +262,11 @@ impl TableProvider for DefaultValueTableProvider {
         .with_expr_adapter(Some(Arc::new(DefaultValuePhysicalExprAdapterFactory) as _))
         .build();
 
-        let parquet_source =
+        Ok(DataSourceExec::from_data_source(
             ParquetSource::new(TableParquetOptions::default(), config.clone())
                 .with_predicate(filter)
-                .with_pushdown_filters(true);
-
-        Ok(DataSourceExec::from_data_source(parquet_source))
+                .with_pushdown_filters(true),
+        ))
     }
 }
 

datafusion-examples/examples/json_shredding.rs

@@ -270,12 +270,11 @@ impl TableProvider for ExampleTableProvider {
                 .with_expr_adapter(Some(Arc::new(ShreddedJsonRewriterFactory) as _))
                 .build();
 
-        let parquet_source =
+        Ok(DataSourceExec::from_data_source(
             ParquetSource::new(TableParquetOptions::default(), config.clone())
                 .with_predicate(filter)
-                .with_pushdown_filters(true);
-
-        Ok(DataSourceExec::from_data_source(parquet_source))
+                .with_pushdown_filters(true),
+        ))
     }
 }
 

datafusion-examples/examples/parquet_embedded_index.rs

@@ -440,10 +440,10 @@ impl TableProvider for DistinctIndexTable {
             builder = builder.with_file(partitioned_file);
         }
 
-        let config = builder.build();
-        let source = ParquetSource::new(TableParquetOptions::default(), config.clone())
-            .with_enable_page_index(true);
-        Ok(DataSourceExec::from_data_source(source))
+        Ok(DataSourceExec::from_data_source(
+            ParquetSource::new(TableParquetOptions::default(), builder.build())
+                .with_enable_page_index(true),
+        ))
     }
 
     /// Tell DataFusion that we can handle filters on the "category" column

datafusion-examples/examples/parquet_index.rs

@@ -260,11 +260,13 @@ impl TableProvider for IndexTableProvider {
             );
         }
 
-        let config = file_scan_config_builder.build();
-        let source = ParquetSource::new(TableParquetOptions::default(), config.clone())
-            .with_predicate(predicate);
-
-        Ok(DataSourceExec::from_data_source(source))
+        Ok(DataSourceExec::from_data_source(
+            ParquetSource::new(
+                TableParquetOptions::default(),
+                file_scan_config_builder.build(),
+            )
+            .with_predicate(predicate),
+        ))
     }
 
     /// Tell DataFusion to push filters down to the scan method

datafusion/core/src/datasource/file_format/mod.rs

@@ -89,9 +89,7 @@ pub(crate) mod test_util {
                 .build();
 
         let source = format.file_source(config);
-        let exec = format.create_physical_plan(state, source).await?;
-
-        Ok(exec)
+        Ok(format.create_physical_plan(state, source).await?)
     }
 }
 

datafusion/core/src/datasource/listing/table.rs

@@ -1125,7 +1125,6 @@ impl ListingTable {
         if let Some(factory) = &self.schema_adapter_factory {
             source = source.with_schema_adapter_factory(Arc::clone(factory))?;
         }
-
         Ok(source)
     }
 
@@ -1249,12 +1248,13 @@ impl TableProvider for ListingTable {
                 .with_expr_adapter(self.expr_adapter_factory.clone())
                 .build();
 
-        let file_source = self.create_file_source_with_schema_adapter(conf)?;
-
         // create the execution plan
         self.options
             .format
-            .create_physical_plan(state, file_source)
+            .create_physical_plan(
+                state,
+                self.create_file_source_with_schema_adapter(conf)?,
+            )
             .await
     }
 

datafusion/datasource/src/file_scan_config.rs

@@ -62,6 +62,7 @@ use log::{debug, warn};
 /// # Example
 /// ```
 /// # use std::any::Any;
+/// # use std::fmt::{Debug, Formatter};
 /// # use std::sync::Arc;
 /// # use arrow::datatypes::{Field, Fields, DataType, Schema, SchemaRef};
 /// # use object_store::ObjectStore;
@@ -72,7 +73,7 @@ use log::{debug, warn};
 /// # use datafusion_datasource::PartitionedFile;
 /// # use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
 /// # use datafusion_datasource::file_stream::FileOpener;
-/// # use datafusion_datasource::source::DataSourceExec;
+/// # use datafusion_datasource::source::{DataSource, DataSourceExec};
 /// # use datafusion_execution::object_store::ObjectStoreUrl;
 /// # use datafusion_physical_plan::ExecutionPlan;
 /// # use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
@@ -90,15 +91,19 @@ use log::{debug, warn};
 /// #    schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>
 /// #    config: FileScanConfig,
 /// # };
-/// # impl FileSource for ParquetSource {
+/// # impl Debug for ParquetSource {fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { unimplemented!() }}
+///
+/// impl FileSource for ParquetSource {
 /// #  fn config(&self) -> &FileScanConfig { &self.config }
+/// #  fn with_config(&self, config: FileScanConfig) -> Arc<dyn FileSource> { unimplemented!() }
+/// #  fn as_data_source(&self) -> Arc<dyn DataSource> { Arc::new(self.clone()) }
 /// #  fn create_file_opener(&self, _: Arc<dyn ObjectStore>, _: usize) -> Arc<dyn FileOpener> { unimplemented!() }
 /// #  fn as_any(&self) -> &dyn Any { self  }
-/// #  fn metrics(&self) -> &ExecutionPlanMetricsSet { unimplemented!() }
+/// #  fn metrics_inner(&self) -> &ExecutionPlanMetricsSet { unimplemented!() }
 /// #  fn file_type(&self) -> &str { "parquet" }
 /// #  fn with_schema_adapter_factory(&self, factory: Arc<dyn SchemaAdapterFactory>) -> Result<Arc<dyn FileSource>> { Ok(Arc::new(Self {projected_statistics: self.projected_statistics.clone(), schema_adapter_factory: Some(factory), config: self.config.clone() } )) }
 /// #  fn schema_adapter_factory(&self) -> Option<Arc<dyn SchemaAdapterFactory>> { self.schema_adapter_factory.clone() }
-/// #  }
+/// # }
 /// # impl ParquetSource {
 /// #  fn new(config: FileScanConfig) -> Self { Self {projected_statistics: None, schema_adapter_factory: None, config} }
 /// # }
@@ -115,9 +120,8 @@ use log::{debug, warn};
 ///    PartitionedFile::new("file2.parquet", 56),
 ///    PartitionedFile::new("file3.parquet", 78),
 ///   ])).build();
-/// let file_source = Arc::new(ParquetSource::new(config.clone()));
 /// // create an execution plan from the config
-/// let plan: Arc<dyn ExecutionPlan> =DataSourceExec::from_data_source(source);
+/// let plan: Arc<dyn ExecutionPlan> =DataSourceExec::from_data_source(ParquetSource::new(config.clone()));
 /// ```
 #[derive(Clone)]
 pub struct FileScanConfig {