Merge remote-tracking branch 'apache/main' into alamb/improve_boolean…

…_handling

datafusion-cli/src/functions.rs

@@ -214,6 +214,7 @@ pub fn display_all_functions() -> Result<()> {
 }
 
 /// PARQUET_META table function
+#[derive(Debug)]
 struct ParquetMetadataTable {
     schema: SchemaRef,
     batch: RecordBatch,

datafusion-examples/examples/simple_udtf.rs

@@ -73,6 +73,7 @@ async fn main() -> Result<()> {
 /// Usage: `read_csv(filename, [limit])`
 ///
 /// [`read_csv`]: https://duckdb.org/docs/data/csv/overview.html
+#[derive(Debug)]
 struct LocalCsvTable {
     schema: SchemaRef,
     limit: Option<usize>,

datafusion/catalog/src/table.rs

@@ -17,6 +17,7 @@
 
 use std::any::Any;
 use std::borrow::Cow;
+use std::fmt::Debug;
 use std::sync::Arc;
 
 use crate::session::Session;
@@ -31,7 +32,7 @@ use datafusion_physical_plan::ExecutionPlan;
 
 /// Source table
 #[async_trait]
-pub trait TableProvider: Sync + Send {
+pub trait TableProvider: Debug + Sync + Send {
     /// Returns the table provider as [`Any`](std::any::Any) so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;
@@ -193,6 +194,7 @@ pub trait TableProvider: Sync + Send {
     /// # use datafusion_expr::{Expr, TableProviderFilterPushDown, TableType};
     /// # use datafusion_physical_plan::ExecutionPlan;
     /// // Define a struct that implements the TableProvider trait
+    /// #[derive(Debug)]
     /// struct TestDataSource {}
     ///
     /// #[async_trait]
@@ -212,7 +214,7 @@ pub trait TableProvider: Sync + Send {
     ///             // This example only supports a between expr with a single column named "c1".
     ///             Expr::Between(between_expr) => {
     ///                 between_expr.expr
-    ///                 .try_into_col()
+    ///                 .try_as_col()
     ///                 .map(|column| {
     ///                     if column.name == "c1" {
     ///                         TableProviderFilterPushDown::Exact
@@ -283,7 +285,7 @@ pub trait TableProvider: Sync + Send {
 /// For example, this can be used to create a table "on the fly"
 /// from a directory of files only when that name is referenced.
 #[async_trait]
-pub trait TableProviderFactory: Sync + Send {
+pub trait TableProviderFactory: Debug + Sync + Send {
     /// Create a TableProvider with the given url
     async fn create(
         &self,

datafusion/core/src/catalog_common/information_schema.rs

@@ -19,15 +19,15 @@
 //!
 //! [Information Schema]: https://en.wikipedia.org/wiki/Information_schema
 
-use async_trait::async_trait;
-use datafusion_common::DataFusionError;
-use std::{any::Any, sync::Arc};
-
 use arrow::{
     array::{StringBuilder, UInt64Builder},
     datatypes::{DataType, Field, Schema, SchemaRef},
     record_batch::RecordBatch,
 };
+use async_trait::async_trait;
+use datafusion_common::DataFusionError;
+use std::fmt::{Debug, Formatter};
+use std::{any::Any, sync::Arc};
 
 use crate::catalog::{CatalogProviderList, SchemaProvider, TableProvider};
 use crate::datasource::streaming::StreamingTable;
@@ -75,6 +75,15 @@ struct InformationSchemaConfig {
     catalog_list: Arc<dyn CatalogProviderList>,
 }
 
+impl Debug for InformationSchemaConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("InformationSchemaConfig")
+            // TODO it would be great to print the catalog list here
+            // but that would require CatalogProviderList to implement Debug
+            .finish_non_exhaustive()
+    }
+}
+
 impl InformationSchemaConfig {
     /// Construct the `information_schema.tables` virtual table
     async fn make_tables(
@@ -246,6 +255,7 @@ impl SchemaProvider for InformationSchemaProvider {
     }
 }
 
+#[derive(Debug)]
 struct InformationSchemaTables {
     schema: SchemaRef,
     config: InformationSchemaConfig,
@@ -337,6 +347,7 @@ impl InformationSchemaTablesBuilder {
     }
 }
 
+#[derive(Debug)]
 struct InformationSchemaViews {
     schema: SchemaRef,
     config: InformationSchemaConfig,
@@ -424,6 +435,7 @@ impl InformationSchemaViewBuilder {
     }
 }
 
+#[derive(Debug)]
 struct InformationSchemaColumns {
     schema: SchemaRef,
     config: InformationSchemaConfig,
@@ -640,6 +652,7 @@ impl InformationSchemaColumnsBuilder {
     }
 }
 
+#[derive(Debug)]
 struct InformationSchemata {
     schema: SchemaRef,
     config: InformationSchemaConfig,
@@ -741,6 +754,7 @@ impl PartitionStream for InformationSchemata {
     }
 }
 
+#[derive(Debug)]
 struct InformationSchemaDfSettings {
     schema: SchemaRef,
     config: InformationSchemaConfig,

datafusion/core/src/dataframe/mod.rs

@@ -1873,6 +1873,7 @@ impl DataFrame {
     }
 }
 
+#[derive(Debug)]
 struct DataFrameTableProvider {
     plan: LogicalPlan,
 }
@@ -2028,6 +2029,43 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_coalesce_schema() -> Result<()> {
+        let ctx = SessionContext::new();
+
+        let query = r#"SELECT COALESCE(null, 5)"#;
+
+        let result = ctx.sql(query).await?;
+        assert_logical_expr_schema_eq_physical_expr_schema(result).await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_coalesce_from_values_schema() -> Result<()> {
+        let ctx = SessionContext::new();
+
+        let query = r#"SELECT COALESCE(column1, column2) FROM VALUES (null, 1.2)"#;
+
+        let result = ctx.sql(query).await?;
+        assert_logical_expr_schema_eq_physical_expr_schema(result).await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_coalesce_from_values_schema_multiple_rows() -> Result<()> {
+        let ctx = SessionContext::new();
+
+        let query = r#"SELECT COALESCE(column1, column2)
+        FROM VALUES
+        (null, 1.2),
+        (1.1, null),
+        (2, 5);"#;
+
+        let result = ctx.sql(query).await?;
+        assert_logical_expr_schema_eq_physical_expr_schema(result).await?;
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_array_agg_schema() -> Result<()> {
         let ctx = SessionContext::new();

datafusion/core/src/datasource/cte_worktable.rs

@@ -36,6 +36,7 @@ use crate::datasource::{TableProvider, TableType};
 /// The temporary working table where the previous iteration of a recursive query is stored
 /// Naming is based on PostgreSQL's implementation.
 /// See here for more details: www.postgresql.org/docs/11/queries-with.html#id-1.5.6.12.5.4
+#[derive(Debug)]
 pub struct CteWorkTable {
     /// The name of the CTE work table
     // WIP, see https://github.com/apache/datafusion/issues/462

datafusion/core/src/datasource/empty.rs

@@ -32,6 +32,7 @@ use crate::physical_plan::{empty::EmptyExec, ExecutionPlan};
 
 /// An empty plan that is useful for testing and generating plans
 /// without mapping them to actual data.
+#[derive(Debug)]
 pub struct EmptyTable {
     schema: SchemaRef,
     partitions: usize,

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

@@ -646,6 +646,7 @@ impl ListingOptions {
 /// # Ok(())
 /// # }
 /// ```
+#[derive(Debug)]
 pub struct ListingTable {
     table_paths: Vec<ListingTableUrl>,
     /// File fields only

datafusion/core/src/datasource/stream.rs

@@ -293,6 +293,7 @@ impl StreamConfig {
 ///
 /// [Hadoop]: https://hadoop.apache.org/
 /// [`ListingTable`]: crate::datasource::listing::ListingTable
+#[derive(Debug)]
 pub struct StreamTable(Arc<StreamConfig>);
 
 impl StreamTable {
@@ -370,6 +371,7 @@ impl TableProvider for StreamTable {
     }
 }
 
+#[derive(Debug)]
 struct StreamRead(Arc<StreamConfig>);
 
 impl PartitionStream for StreamRead {

datafusion/core/src/datasource/streaming.rs

@@ -32,6 +32,7 @@ use datafusion_expr::{Expr, TableType};
 use log::debug;
 
 /// A [`TableProvider`] that streams a set of [`PartitionStream`]
+#[derive(Debug)]
 pub struct StreamingTable {
     schema: SchemaRef,
     partitions: Vec<Arc<dyn PartitionStream>>,

datafusion/core/src/datasource/view.rs

@@ -36,6 +36,7 @@ use datafusion_optimizer::Analyzer;
 use crate::datasource::{TableProvider, TableType};
 
 /// An implementation of `TableProvider` that uses another logical plan.
+#[derive(Debug)]
 pub struct ViewTable {
     /// LogicalPlan of the view
     logical_plan: LogicalPlan,

datafusion/core/src/physical_optimizer/coalesce_batches.rs

@@ -34,7 +34,7 @@ use datafusion_physical_optimizer::PhysicalOptimizerRule;
 
 /// Optimizer rule that introduces CoalesceBatchesExec to avoid overhead with small batches that
 /// are produced by highly selective filters
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct CoalesceBatches {}
 
 impl CoalesceBatches {

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -176,7 +176,7 @@ use itertools::izip;
 ///
 /// This rule only chooses the exact match and satisfies the Distribution(a, b, c)
 /// by a HashPartition(a, b, c).
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct EnforceDistribution {}
 
 impl EnforceDistribution {

datafusion/core/src/physical_optimizer/enforce_sorting.rs

@@ -72,7 +72,7 @@ use itertools::izip;
 
 /// This rule inspects [`SortExec`]'s in the given physical plan and removes the
 /// ones it can prove unnecessary.
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct EnforceSorting {}
 
 impl EnforceSorting {

datafusion/core/src/physical_optimizer/join_selection.rs

@@ -46,7 +46,7 @@ use datafusion_physical_optimizer::PhysicalOptimizerRule;
 /// The [`JoinSelection`] rule tries to modify a given plan so that it can
 /// accommodate infinite sources and optimize joins in the plan according to
 /// available statistical information, if there is any.
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct JoinSelection {}
 
 impl JoinSelection {

datafusion/core/src/physical_optimizer/optimizer.rs

@@ -35,7 +35,7 @@ use crate::physical_optimizer::sanity_checker::SanityCheckPlan;
 use crate::physical_optimizer::topk_aggregation::TopKAggregation;
 
 /// A rule-based physical optimizer.
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct PhysicalOptimizer {
     /// All rules to apply
     pub rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,

datafusion/core/src/physical_optimizer/projection_pushdown.rs

@@ -60,7 +60,7 @@ use itertools::Itertools;
 
 /// This rule inspects [`ProjectionExec`]'s in the given physical plan and tries to
 /// remove or swap with its child.
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct ProjectionPushdown {}
 
 impl ProjectionPushdown {
@@ -1831,6 +1831,7 @@ mod tests {
 
     #[test]
     fn test_streaming_table_after_projection() -> Result<()> {
+        #[derive(Debug)]
         struct DummyStreamPartition {
             schema: SchemaRef,
         }

datafusion/core/src/physical_optimizer/sanity_checker.rs

@@ -33,6 +33,7 @@ use datafusion_physical_expr::intervals::utils::{check_support, is_datatype_supp
 use datafusion_physical_plan::joins::SymmetricHashJoinExec;
 use datafusion_physical_plan::{get_plan_string, ExecutionPlanProperties};
 
+use datafusion_physical_expr_common::sort_expr::format_physical_sort_requirement_list;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use itertools::izip;
 
@@ -41,7 +42,7 @@ use itertools::izip;
 ///    are not satisfied by their children.
 /// 2. Plans that use pipeline-breaking operators on infinite input(s),
 ///    it is impossible to execute such queries (they will never generate output nor finish)
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct SanityCheckPlan {}
 
 impl SanityCheckPlan {
@@ -130,9 +131,9 @@ pub fn check_plan_sanity(
             if !child_eq_props.ordering_satisfy_requirement(sort_req) {
                 let plan_str = get_plan_string(&plan);
                 return plan_err!(
-                    "Plan: {:?} does not satisfy order requirements: {:?}. Child-{} order: {:?}",
+                    "Plan: {:?} does not satisfy order requirements: {}. Child-{} order: {}",
                     plan_str,
-                    sort_req,
+                    format_physical_sort_requirement_list(sort_req),
                     idx,
                     child_eq_props.oeq_class
                 );
@@ -145,7 +146,7 @@ pub fn check_plan_sanity(
         {
             let plan_str = get_plan_string(&plan);
             return plan_err!(
-                "Plan: {:?} does not satisfy distribution requirements: {:?}. Child-{} output partitioning: {:?}",
+                "Plan: {:?} does not satisfy distribution requirements: {}. Child-{} output partitioning: {}",
                 plan_str,
                 dist_req,
                 idx,

datafusion/core/src/physical_optimizer/update_aggr_exprs.rs

@@ -48,7 +48,7 @@ use datafusion_physical_plan::{
 /// This rule analyzes aggregate expressions of type `Beneficial` to see whether
 /// their input ordering requirements are satisfied. If this is the case, the
 /// aggregators are modified to run in a more efficient mode.
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct OptimizeAggregateOrder {}
 
 impl OptimizeAggregateOrder {

datafusion/core/src/test/mod.rs

@@ -297,6 +297,7 @@ pub fn csv_exec_sorted(
 }
 
 // construct a stream partition for test purposes
+#[derive(Debug)]
 pub(crate) struct TestStreamPartition {
     pub schema: SchemaRef,
 }

datafusion/core/src/test_util/mod.rs

@@ -174,6 +174,7 @@ pub fn populate_csv_partitions(
 }
 
 /// TableFactory for tests
+#[derive(Default, Debug)]
 pub struct TestTableFactory {}
 
 #[async_trait]
@@ -191,6 +192,7 @@ impl TableProviderFactory for TestTableFactory {
 }
 
 /// TableProvider for testing purposes
+#[derive(Debug)]
 pub struct TestTableProvider {
     /// URL of table files or folder
     pub url: String,

datafusion/core/tests/custom_sources_cases/mod.rs

@@ -71,7 +71,9 @@ macro_rules! TEST_CUSTOM_RECORD_BATCH {
 
 //--- Custom source dataframe tests ---//
 
+#[derive(Debug)]
 struct CustomTableProvider;
+
 #[derive(Debug, Clone)]
 struct CustomExecutionPlan {
     projection: Option<Vec<usize>>,

datafusion/core/tests/custom_sources_cases/provider_filter_pushdown.rs

@@ -142,7 +142,7 @@ impl ExecutionPlan for CustomPlan {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 struct CustomProvider {
     zero_batch: RecordBatch,
     one_batch: RecordBatch,