Merge branch 'main' into implement_groups_accumulator_count_distinct_primitive_types

Author: coderfender

datafusion-cli/src/catalog.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::sync::{Arc, Weak};
 
 use crate::object_storage::{AwsOptions, GcpOptions, get_object_store};
@@ -50,10 +49,6 @@ impl DynamicObjectStoreCatalog {
 }
 
 impl CatalogProviderList for DynamicObjectStoreCatalog {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn register_catalog(
         &self,
         name: String,
@@ -91,10 +86,6 @@ impl DynamicObjectStoreCatalogProvider {
 }
 
 impl CatalogProvider for DynamicObjectStoreCatalogProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema_names(&self) -> Vec<String> {
         self.inner.schema_names()
     }
@@ -134,10 +125,6 @@ impl DynamicObjectStoreSchemaProvider {
 
 #[async_trait]
 impl SchemaProvider for DynamicObjectStoreSchemaProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn table_names(&self) -> Vec<String> {
         self.inner.table_names()
     }

datafusion-cli/src/functions.rs

@@ -229,10 +229,6 @@ struct ParquetMetadataTable {
 
 #[async_trait]
 impl TableProvider for ParquetMetadataTable {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
@@ -479,10 +475,6 @@ struct MetadataCacheTable {
 
 #[async_trait]
 impl TableProvider for MetadataCacheTable {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
@@ -598,10 +590,6 @@ struct StatisticsCacheTable {
 
 #[async_trait]
 impl TableProvider for StatisticsCacheTable {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }
@@ -734,10 +722,6 @@ struct ListFilesCacheTable {
 
 #[async_trait]
 impl TableProvider for ListFilesCacheTable {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

datafusion-examples/examples/custom_data_source/custom_datasource.rs

@@ -17,7 +17,6 @@
 
 //! See `main.rs` for how to run it.
 
-use std::any::Any;
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::{self, Debug, Formatter};
 use std::sync::{Arc, Mutex};
@@ -162,10 +161,6 @@ impl Default for CustomDataSource {
 
 #[async_trait]
 impl TableProvider for CustomDataSource {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         SchemaRef::new(Schema::new(vec![
             Field::new("id", DataType::UInt8, false),

datafusion-examples/examples/custom_data_source/custom_file_casts.rs

@@ -33,7 +33,7 @@ use datafusion::execution::context::SessionContext;
 use datafusion::execution::object_store::ObjectStoreUrl;
 use datafusion::parquet::arrow::ArrowWriter;
 use datafusion::physical_expr::PhysicalExpr;
-use datafusion::physical_expr::expressions::{CastColumnExpr, CastExpr};
+use datafusion::physical_expr::expressions::CastExpr;
 use datafusion::prelude::SessionConfig;
 use datafusion_physical_expr_adapter::{
     DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
@@ -43,9 +43,10 @@ use object_store::path::Path;
 use object_store::{ObjectStore, ObjectStoreExt, PutPayload};
 
 // Example showing how to implement custom casting rules to adapt file schemas.
-// This example enforces that casts must be strictly widening: if the file type is Int64 and the table type is Int32, it will error
-// before even reading the data.
-// Without this custom cast rule DataFusion would happily do the narrowing cast, potentially erroring only if it found a row with data it could not cast.
+// This example enforces strictly widening casts: if the file type is Int64 and
+// the table type is Int32, it errors before reading the data. Without this
+// custom cast rule DataFusion would apply the narrowing cast and might only
+// error after reading a row that it could not cast.
 pub async fn custom_file_casts() -> Result<()> {
     println!("=== Creating example data ===");
 
@@ -139,7 +140,7 @@ async fn write_data(
     Ok(())
 }
 
-/// Factory for creating DefaultValuePhysicalExprAdapter instances
+/// Factory for creating custom cast physical expression adapters
 #[derive(Debug)]
 struct CustomCastPhysicalExprAdapterFactory {
     inner: Arc<dyn PhysicalExprAdapterFactory>,
@@ -167,8 +168,8 @@ impl PhysicalExprAdapterFactory for CustomCastPhysicalExprAdapterFactory {
     }
 }
 
-/// Custom PhysicalExprAdapter that handles missing columns with default values from metadata
-/// and wraps DefaultPhysicalExprAdapter for standard schema adaptation
+/// Custom `PhysicalExprAdapter` that wraps the default adapter and rejects
+/// narrowing file-schema casts.
 #[derive(Debug, Clone)]
 struct CustomCastsPhysicalExprAdapter {
     physical_file_schema: SchemaRef,
@@ -177,34 +178,23 @@ struct CustomCastsPhysicalExprAdapter {
 
 impl PhysicalExprAdapter for CustomCastsPhysicalExprAdapter {
     fn rewrite(&self, mut expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
-        // First delegate to the inner adapter to handle missing columns and discover any necessary casts
+        // First delegate to the inner adapter to handle standard schema adaptation
+        // and discover any necessary casts.
         expr = self.inner.rewrite(expr)?;
-        // Now we can apply custom casting rules or even swap out all CastExprs for a custom cast kernel / expression
-        // For example, [DataFusion Comet](https://github.com/apache/datafusion-comet) has a [custom cast kernel](https://github.com/apache/datafusion-comet/blob/b4ac876ab420ed403ac7fc8e1b29f42f1f442566/native/spark-expr/src/conversion_funcs/cast.rs#L133-L138).
+        // Now apply custom casting rules or swap CastExprs for a custom cast
+        // kernel / expression. For example, DataFusion Comet has a custom cast
+        // kernel in its native Spark expression implementation.
         expr.transform(|expr| {
             if let Some(cast) = expr.as_any().downcast_ref::<CastExpr>() {
                 let input_data_type =
                     cast.expr().data_type(&self.physical_file_schema)?;
-                let output_data_type = cast.data_type(&self.physical_file_schema)?;
+                let output_data_type = cast.target_field().data_type();
                 if !cast.is_bigger_cast(&input_data_type) {
                     return not_impl_err!(
                         "Unsupported CAST from {input_data_type} to {output_data_type}"
                     );
                 }
             }
-            if let Some(cast) = expr.as_any().downcast_ref::<CastColumnExpr>() {
-                let input_data_type =
-                    cast.expr().data_type(&self.physical_file_schema)?;
-                let output_data_type = cast.data_type(&self.physical_file_schema)?;
-                if !CastExpr::check_bigger_cast(
-                    cast.target_field().data_type(),
-                    &input_data_type,
-                ) {
-                    return not_impl_err!(
-                        "Unsupported CAST from {input_data_type} to {output_data_type}"
-                    );
-                }
-            }
             Ok(Transformed::no(expr))
         })
         .data()

datafusion-examples/examples/custom_data_source/default_column_values.rs

@@ -17,7 +17,6 @@
 
 //! See `main.rs` for how to run it.
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -201,10 +200,6 @@ impl DefaultValueTableProvider {
 
 #[async_trait]
 impl TableProvider for DefaultValueTableProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }

datafusion-examples/examples/custom_data_source/main.rs

@@ -21,7 +21,7 @@
 //!
 //! ## Usage
 //! ```bash
-//! cargo run --example custom_data_source -- [all|csv_json_opener|csv_sql_streaming|custom_datasource|custom_file_casts|custom_file_format|default_column_values|file_stream_provider]
+//! cargo run --example custom_data_source -- [all|adapter_serialization|csv_json_opener|csv_sql_streaming|custom_datasource|custom_file_casts|custom_file_format|default_column_values|file_stream_provider]
 //! ```
 //!
 //! Each subcommand runs a corresponding example:

datafusion-examples/examples/data_io/catalog.rs

@@ -32,7 +32,7 @@ use datafusion::{
     prelude::SessionContext,
 };
 use std::sync::RwLock;
-use std::{any::Any, collections::HashMap, path::Path, sync::Arc};
+use std::{collections::HashMap, path::Path, sync::Arc};
 use std::{fs::File, io::Write};
 use tempfile::TempDir;
 
@@ -178,10 +178,6 @@ impl DirSchema {
 
 #[async_trait]
 impl SchemaProvider for DirSchema {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn table_names(&self) -> Vec<String> {
         let tables = self.tables.read().unwrap();
         tables.keys().cloned().collect::<Vec<_>>()
@@ -231,10 +227,6 @@ impl DirCatalog {
 }
 
 impl CatalogProvider for DirCatalog {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn register_schema(
         &self,
         name: &str,
@@ -277,10 +269,6 @@ impl CustomCatalogProviderList {
 }
 
 impl CatalogProviderList for CustomCatalogProviderList {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn register_catalog(
         &self,
         name: String,

datafusion-examples/examples/data_io/parquet_advanced_index.rs

@@ -17,7 +17,6 @@
 
 //! See `main.rs` for how to run it.
 
-use std::any::Any;
 use std::collections::{HashMap, HashSet};
 use std::fs::File;
 use std::ops::Range;
@@ -451,10 +450,6 @@ impl IndexedFile {
 /// so that we can query it as a table.
 #[async_trait]
 impl TableProvider for IndexTableProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.indexed_file.schema)
     }

datafusion-examples/examples/data_io/parquet_embedded_index.rs

@@ -393,9 +393,6 @@ fn get_key_value<'a>(file_meta_data: &'a FileMetaData, key: &'_ str) -> Option<&
 /// Implement TableProvider for DistinctIndexTable, using the distinct index to prune files
 #[async_trait]
 impl TableProvider for DistinctIndexTable {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

datafusion-examples/examples/data_io/parquet_index.rs

@@ -45,7 +45,6 @@ use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_optimizer::pruning::PruningPredicate;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::*;
-use std::any::Any;
 use std::collections::HashSet;
 use std::fmt::Display;
 use std::fs;
@@ -208,10 +207,6 @@ impl IndexTableProvider {
 
 #[async_trait]
 impl TableProvider for IndexTableProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.index.schema().clone()
     }