Merge branch 'main' into fix-scalar-val-num-ops

Author: Jefffrey

.github/workflows/rust.yml

@@ -697,7 +697,9 @@ jobs:
         with:
           rust-version: stable
       - name: Install taplo
-        run: cargo +stable install taplo-cli --version ^0.9 --locked
+        uses: taiki-e/install-action@43aecc8d72668fbcfe75c31400bc4f890f1c5853  # v2.83.2
+        with:
+          tool: taplo-cli@0.9
       # if you encounter an error, try running 'taplo format' to fix the formatting automatically.
       - name: Check Cargo.toml formatting
         run: taplo format --check

ci/scripts/check_no_cargo_install_in_workflows.sh

@@ -22,7 +22,7 @@ set -euo pipefail
 SCRIPT_NAME="$(basename "${BASH_SOURCE[0]}")"
 WORKFLOWS_DIR=".github/workflows"
 
-if grep -R -n --include='*.yml' --include='*.yaml' -- 'cargo install' "${WORKFLOWS_DIR}"; then
+if grep -R -E -w -n --include='*.yml' --include='*.yaml' -- 'cargo.*install' "${WORKFLOWS_DIR}"; then
   echo "[${SCRIPT_NAME}] Found workflow Rust tool installs that should use taiki-e/install-action instead." >&2
   exit 1
 fi

datafusion-examples/examples/data_io/parquet_embedded_index.rs

@@ -87,7 +87,7 @@
 //! 2. Read and deserialize the index.
 //!
 //! 3. Create a `TableProvider` that knows how to use the index to quickly find
-//!   the relevant files, row groups, data pages or rows based on on pushed down
+//!   the relevant files, row groups, data pages or rows based on pushed down
 //!   filters.
 //!
 //! # FAQ: Why do other Parquet readers skip over the custom index?

datafusion-examples/examples/execution_monitoring/memory_pool_execution_plan.rs

@@ -26,9 +26,9 @@
 //! - Handle memory pressure by spilling to disk
 //! - Release memory when done
 
+use arrow::array::record_batch;
 use arrow::record_batch::RecordBatch;
 use arrow_schema::SchemaRef;
-use datafusion::common::record_batch;
 use datafusion::common::{exec_datafusion_err, internal_err};
 use datafusion::datasource::{DefaultTableSource, memory::MemTable};
 use datafusion::error::Result;

datafusion-examples/examples/ffi/ffi_example_table_provider/src/lib.rs

@@ -17,9 +17,10 @@
 
 use std::sync::Arc;
 
-use arrow::array::RecordBatch;
+use arrow::array::{RecordBatch, record_batch};
+use arrow::datatypes as arrow_schema;
 use arrow::datatypes::{DataType, Field, Schema};
-use datafusion::{common::record_batch, datasource::MemTable};
+use datafusion::datasource::MemTable;
 use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use datafusion_ffi::table_provider::FFI_TableProvider;
 use ffi_module_interface::TableProviderModule;

datafusion/catalog/src/information_schema.rs

@@ -20,6 +20,7 @@
 //! [Information Schema]: https://en.wikipedia.org/wiki/Information_schema
 
 use crate::streaming::StreamingTable;
+use crate::table::TableFunction;
 use crate::{CatalogProviderList, SchemaProvider, TableProvider};
 use arrow::array::builder::{BooleanBuilder, UInt8Builder};
 use arrow::{
@@ -81,14 +82,28 @@ impl InformationSchemaProvider {
     /// Creates a new [`InformationSchemaProvider`] for the provided `catalog_list`
     pub fn new(catalog_list: Arc<dyn CatalogProviderList>) -> Self {
         Self {
-            config: InformationSchemaConfig { catalog_list },
+            config: InformationSchemaConfig {
+                catalog_list,
+                table_functions: HashMap::new(),
+            },
         }
     }
+
+    /// Attach the session's table (UDTF) functions so that they appear in
+    /// `information_schema.routines` / `SHOW FUNCTIONS`.
+    pub fn with_table_functions(
+        mut self,
+        table_functions: HashMap<String, Arc<TableFunction>>,
+    ) -> Self {
+        self.config.table_functions = table_functions;
+        self
+    }
 }
 
 #[derive(Clone, Debug)]
 struct InformationSchemaConfig {
     catalog_list: Arc<dyn CatalogProviderList>,
+    table_functions: HashMap<String, Arc<TableFunction>>,
 }
 
 impl InformationSchemaConfig {
@@ -301,6 +316,26 @@ impl InformationSchemaConfig {
                 )
             }
         }
+
+        // Table functions (UDTFs) don't have scalar signatures; their return
+        // type is always a table, so emit a single row per UDTF with
+        // routine_type = "FUNCTION", function_type = "TABLE" and
+        // data_type = "TABLE".
+        for name in self.table_functions.keys() {
+            builder.add_routine(
+                catalog_name,
+                schema_name,
+                name,
+                "FUNCTION",
+                // No signature is available for UDTFs; report deterministic
+                // = false to stay conservative.
+                false,
+                Some(&"TABLE"),
+                "TABLE",
+                None::<String>,
+                None::<String>,
+            )
+        }
         Ok(())
     }
 
@@ -400,6 +435,14 @@ impl InformationSchemaConfig {
             }
         }
 
+        // UDTFs deliberately do NOT appear in `information_schema.parameters`.
+        // A same-named scalar UDF (e.g. `generate_series` exists as both a
+        // scalar UDF in functions-nested and a UDTF in functions-table) would
+        // cross-join with a UDTF row keyed only by (name, rid) and produce
+        // spurious `TABLE`-typed variants of every scalar signature in
+        // SHOW FUNCTIONS. `show_functions_to_plan` sources UDTFs directly
+        // from `information_schema.routines` via a UNION branch instead.
+
         Ok(())
     }
 
@@ -1522,6 +1565,7 @@ mod tests {
     async fn make_tables_uses_table_type() {
         let config = InformationSchemaConfig {
             catalog_list: Arc::new(Fixture),
+            table_functions: HashMap::new(),
         };
         let mut builder = InformationSchemaTablesBuilder {
             catalog_names: StringBuilder::new(),

datafusion/common/src/pruning.rs

@@ -305,7 +305,7 @@ impl PruningStatistics for PartitionPruningStatistics {
 /// that has statistics of its columns.
 ///
 /// It is up to the caller to decide what each container represents. For
-/// example, they can come from a file (e.g. [`PartitionedFile`]) or a set of of
+/// example, they can come from a file (e.g. [`PartitionedFile`]) or a set of
 /// files (e.g. [`FileGroup`])
 ///
 /// [`PartitionedFile`]: https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.PartitionedFile.html

datafusion/common/src/test_util.rs

@@ -174,7 +174,7 @@ macro_rules! assert_contains {
 }
 
 /// A macro to assert that one string is NOT contained within another with
-/// a nice error message if they are are.
+/// a nice error message if they are.
 ///
 /// Usage: `assert_not_contains!(actual, unexpected)`
 ///
@@ -364,15 +364,20 @@ macro_rules! create_array {
 /// Creates a record batch from literal slice of values, suitable for rapid
 /// testing and development.
 ///
+/// **Deprecated**: prefer the upstream macro from `arrow`,
+/// [`arrow::array::record_batch`], which now supports both the literal slice
+/// form shown below and a variable/expression form.
+///
 /// Example:
 /// ```
-/// use datafusion_common::record_batch;
+/// use arrow::array::record_batch;
 /// let batch = record_batch!(
 ///     ("a", Int32, vec![1, 2, 3]),
 ///     ("b", Float64, vec![Some(4.0), None, Some(5.0)]),
 ///     ("c", Utf8, vec!["alpha", "beta", "gamma"])
 /// );
 /// ```
+#[deprecated(since = "55.0.0", note = "Use `arrow::array::record_batch` instead")]
 #[macro_export]
 macro_rules! record_batch {
     ($(($name: expr, $type: ident, $values: expr)),*) => {
@@ -776,6 +781,10 @@ mod tests {
     }
 
     #[test]
+    #[expect(
+        deprecated,
+        reason = "testing the deprecated record_batch! macro itself"
+    )]
     fn test_create_record_batch() -> Result<()> {
         use arrow::array::Array;
 

datafusion/core/src/execution/session_state.rs

@@ -349,9 +349,10 @@ impl SessionState {
         let resolved_ref = self.resolve_table_ref(table_ref);
         if self.config.information_schema() && *resolved_ref.schema == *INFORMATION_SCHEMA
         {
-            return Ok(Arc::new(InformationSchemaProvider::new(Arc::clone(
-                &self.catalog_list,
-            ))));
+            return Ok(Arc::new(
+                InformationSchemaProvider::new(Arc::clone(&self.catalog_list))
+                    .with_table_functions(self.table_functions.clone()),
+            ));
         }
 
         self.catalog_list

datafusion/core/tests/dataframe/mod.rs

@@ -39,6 +39,7 @@ use datafusion_functions_aggregate::expr_fn::{
     array_agg, avg, avg_distinct, count, count_distinct, max, median, min, sum,
     sum_distinct,
 };
+use datafusion_functions_nested::expr_fn::{array_filter, array_transform, make_array};
 use datafusion_functions_nested::make_array::make_array_udf;
 use datafusion_functions_window::expr_fn::{first_value, lead, row_number};
 use insta::assert_snapshot;
@@ -78,8 +79,8 @@ use datafusion_expr::{
     CreateMemoryTable, CreateView, DdlStatement, Expr, ExprFunctionExt, ExprSchemable,
     LogicalPlan, LogicalPlanBuilder, ScalarFunctionImplementation, SortExpr, TableType,
     WindowFrame, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition, cast, col,
-    create_udf, exists, in_subquery, lit, out_ref_col, placeholder, scalar_subquery,
-    when, wildcard,
+    create_udf, exists, in_subquery, lambda, lambda_var, lit, out_ref_col, placeholder,
+    scalar_subquery, when, wildcard,
 };
 use datafusion_physical_expr::Partitioning;
 use datafusion_physical_expr::aggregate::AggregateExprBuilder;
@@ -90,7 +91,9 @@ use datafusion_physical_plan::aggregates::{
     AggregateExec, AggregateMode, PhysicalGroupBy,
 };
 use datafusion_physical_plan::empty::EmptyExec;
-use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties, displayable};
+use datafusion_physical_plan::{
+    ExecutionPlan, ExecutionPlanProperties, collect, displayable,
+};
 
 use datafusion::error::Result as DataFusionResult;
 use datafusion::execution::options::JsonReadOptions;
@@ -7238,3 +7241,45 @@ async fn test_grouping_with_alias() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_unresolved_lambda_variable() -> Result<()> {
+    let plan = table_with_mixed_lists()
+        .await?
+        .with_column(
+            "c",
+            array_transform(
+                make_array(vec![col("list")]),
+                lambda(
+                    ["x"],
+                    array_filter(
+                        lambda_var("x"),
+                        lambda(["y"], lambda_var("y").gt_eq(lit(2))),
+                    ),
+                ),
+            ),
+        )?
+        .select_columns(&["list", "c"])?
+        .into_unoptimized_plan()
+        .resolve_lambda_variables()?
+        .data;
+
+    let session = SessionContext::new();
+    let exec = session.state().create_physical_plan(&plan).await?;
+    let context = session.task_ctx();
+    let results = collect(exec, context).await?;
+
+    let expected = [
+        "+-----------+----------+",
+        "| list      | c        |",
+        "+-----------+----------+",
+        "| [1, 2, 3] | [[2, 3]] |",
+        "|           | []       |",
+        "| []        | [[]]     |",
+        "|           | []       |",
+        "+-----------+----------+",
+    ];
+    assert_batches_eq!(expected, &results);
+
+    Ok(())
+}