Merge branch 'apache:main' into fix/support-duplicate-column-names-6543

Author: RafaelHerrero

datafusion-examples/README.md

@@ -88,18 +88,19 @@ cargo run --example dataframe -- dataframe
 
 #### Category: Single Process
 
-| Subcommand           | File Path                                                                                 | Description                                            |
-| -------------------- | ----------------------------------------------------------------------------------------- | ------------------------------------------------------ |
-| catalog              | [`data_io/catalog.rs`](examples/data_io/catalog.rs)                                       | Register tables into a custom catalog                  |
-| json_shredding       | [`data_io/json_shredding.rs`](examples/data_io/json_shredding.rs)                         | Implement filter rewriting for JSON shredding          |
-| parquet_adv_idx      | [`data_io/parquet_advanced_index.rs`](examples/data_io/parquet_advanced_index.rs)         | Create a secondary index across multiple parquet files |
-| parquet_emb_idx      | [`data_io/parquet_embedded_index.rs`](examples/data_io/parquet_embedded_index.rs)         | Store a custom index inside Parquet files              |
-| parquet_enc          | [`data_io/parquet_encrypted.rs`](examples/data_io/parquet_encrypted.rs)                   | Read & write encrypted Parquet files                   |
-| parquet_enc_with_kms | [`data_io/parquet_encrypted_with_kms.rs`](examples/data_io/parquet_encrypted_with_kms.rs) | Encrypted Parquet I/O using a KMS-backed factory       |
-| parquet_exec_visitor | [`data_io/parquet_exec_visitor.rs`](examples/data_io/parquet_exec_visitor.rs)             | Extract statistics by visiting an ExecutionPlan        |
-| parquet_idx          | [`data_io/parquet_index.rs`](examples/data_io/parquet_index.rs)                           | Create a secondary index                               |
-| query_http_csv       | [`data_io/query_http_csv.rs`](examples/data_io/query_http_csv.rs)                         | Query CSV files via HTTP                               |
-| remote_catalog       | [`data_io/remote_catalog.rs`](examples/data_io/remote_catalog.rs)                         | Interact with a remote catalog                         |
+| Subcommand             | File Path                                                                                 | Description                                                               |
+| ---------------------- | ----------------------------------------------------------------------------------------- | ------------------------------------------------------------------------- |
+| catalog                | [`data_io/catalog.rs`](examples/data_io/catalog.rs)                                       | Register tables into a custom catalog                                     |
+| in_memory_object_store | [`data_io/in_memory_object_store.rs`](examples/data_io/in_memory_object_store.rs)         | Read CSV from an in-memory object store (pattern applies to JSON/Parquet) |
+| json_shredding         | [`data_io/json_shredding.rs`](examples/data_io/json_shredding.rs)                         | Implement filter rewriting for JSON shredding                             |
+| parquet_adv_idx        | [`data_io/parquet_advanced_index.rs`](examples/data_io/parquet_advanced_index.rs)         | Create a secondary index across multiple parquet files                    |
+| parquet_emb_idx        | [`data_io/parquet_embedded_index.rs`](examples/data_io/parquet_embedded_index.rs)         | Store a custom index inside Parquet files                                 |
+| parquet_enc            | [`data_io/parquet_encrypted.rs`](examples/data_io/parquet_encrypted.rs)                   | Read & write encrypted Parquet files                                      |
+| parquet_enc_with_kms   | [`data_io/parquet_encrypted_with_kms.rs`](examples/data_io/parquet_encrypted_with_kms.rs) | Encrypted Parquet I/O using a KMS-backed factory                          |
+| parquet_exec_visitor   | [`data_io/parquet_exec_visitor.rs`](examples/data_io/parquet_exec_visitor.rs)             | Extract statistics by visiting an ExecutionPlan                           |
+| parquet_idx            | [`data_io/parquet_index.rs`](examples/data_io/parquet_index.rs)                           | Create a secondary index                                                  |
+| query_http_csv         | [`data_io/query_http_csv.rs`](examples/data_io/query_http_csv.rs)                         | Query CSV files via HTTP                                                  |
+| remote_catalog         | [`data_io/remote_catalog.rs`](examples/data_io/remote_catalog.rs)                         | Interact with a remote catalog                                            |
 
 ## DataFrame Examples
 

datafusion-examples/examples/data_io/in_memory_object_store.rs

@@ -0,0 +1,81 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This follows the recommended approach: implement the `ObjectStore` trait
+//! (or use an existing implementation), register it with DataFusion, and then
+//! read a URL "path" from that store.
+//! See the in-memory reference implementation:
+//! https://docs.rs/object_store/latest/object_store/memory/struct.InMemory.html
+
+use std::sync::Arc;
+
+use arrow::datatypes::{DataType, Field, Schema};
+use datafusion::assert_batches_eq;
+use datafusion::common::Result;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::prelude::{CsvReadOptions, SessionContext};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, ObjectStoreExt, PutPayload};
+
+/// Demonstrates reading CSV data from an in-memory object store.
+///
+/// The same pattern applies to JSON/Parquet: register a store for a URL
+/// prefix, write bytes into the store, then read via that URL.
+pub async fn in_memory_object_store() -> Result<()> {
+    let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+    let ctx = SessionContext::new();
+    let object_store_url = ObjectStoreUrl::parse("memory://")?;
+    // Register a URL prefix to route reads through this object store.
+    ctx.register_object_store(object_store_url.as_ref(), Arc::clone(&store));
+
+    let schema = Schema::new(vec![
+        Field::new("id", DataType::Int64, false),
+        Field::new("name", DataType::Utf8, false),
+    ]);
+
+    println!("=== CSV from memory ===");
+    let csv_path = Path::from("/people.csv");
+    let csv_data = b"id,name\n1,Alice\n2,Bob\n";
+    // Write bytes into the in-memory object store.
+    store
+        .put(&csv_path, PutPayload::from_static(csv_data))
+        .await?;
+    // Read using the URL that matches the registered prefix.
+    let csv = ctx
+        .read_csv(
+            "memory:///people.csv",
+            CsvReadOptions::new().schema(&schema),
+        )
+        .await?
+        .collect()
+        .await?;
+    #[rustfmt::skip]
+    let expected = [
+        "+----+-------+",
+        "| id | name  |",
+        "+----+-------+",
+        "| 1  | Alice |",
+        "| 2  | Bob   |",
+        "+----+-------+",
+    ];
+    assert_batches_eq!(expected, &csv);
+
+    Ok(())
+}

datafusion-examples/examples/data_io/main.rs

@@ -21,7 +21,7 @@
 //!
 //! ## Usage
 //! ```bash
-//! cargo run --example data_io -- [all|catalog|json_shredding|parquet_adv_idx|parquet_emb_idx|parquet_enc_with_kms|parquet_enc|parquet_exec_visitor|parquet_idx|query_http_csv|remote_catalog]
+//! cargo run --example data_io -- [all|catalog|in_memory_object_store|json_shredding|parquet_adv_idx|parquet_emb_idx|parquet_enc_with_kms|parquet_enc|parquet_exec_visitor|parquet_idx|query_http_csv|remote_catalog]
 //! ```
 //!
 //! Each subcommand runs a corresponding example:
@@ -30,6 +30,9 @@
 //! - `catalog`
 //!   (file: catalog.rs, desc: Register tables into a custom catalog)
 //!
+//! - `in_memory_object_store`
+//!   (file: in_memory_object_store.rs, desc: Read CSV from an in-memory object store (pattern applies to JSON/Parquet))
+//!
 //! - `json_shredding`
 //!   (file: json_shredding.rs, desc: Implement filter rewriting for JSON shredding)
 //!
@@ -58,6 +61,7 @@
 //!   (file: remote_catalog.rs, desc: Interact with a remote catalog)
 
 mod catalog;
+mod in_memory_object_store;
 mod json_shredding;
 mod parquet_advanced_index;
 mod parquet_embedded_index;
@@ -77,6 +81,7 @@ use strum_macros::{Display, EnumIter, EnumString, VariantNames};
 enum ExampleKind {
     All,
     Catalog,
+    InMemoryObjectStore,
     JsonShredding,
     ParquetAdvIdx,
     ParquetEmbIdx,
@@ -104,6 +109,9 @@ impl ExampleKind {
                 }
             }
             ExampleKind::Catalog => catalog::catalog().await?,
+            ExampleKind::InMemoryObjectStore => {
+                in_memory_object_store::in_memory_object_store().await?
+            }
             ExampleKind::JsonShredding => json_shredding::json_shredding().await?,
             ExampleKind::ParquetAdvIdx => {
                 parquet_advanced_index::parquet_advanced_index().await?

datafusion-examples/examples/query_planning/expr_api.rs

@@ -175,9 +175,10 @@ fn simplify_demo() -> Result<()> {
     // the ExecutionProps carries information needed to simplify
     // expressions, such as the current time (to evaluate `now()`
     // correctly)
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(schema)
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
 
     // And then call the simplify_expr function:
@@ -192,9 +193,10 @@ fn simplify_demo() -> Result<()> {
 
     // here are some other examples of what DataFusion is capable of
     let schema = Schema::new(vec![make_field("i", DataType::Int64)]).to_dfschema_ref()?;
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(Arc::clone(&schema))
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
 
     // basic arithmetic simplification
@@ -554,9 +556,10 @@ fn type_coercion_demo() -> Result<()> {
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 2. Type coercion with `ExprSimplifier::coerce`.
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(Arc::new(df_schema.clone()))
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
     let coerced_expr = simplifier.coerce(expr.clone(), &df_schema)?;
     let physical_expr = datafusion::physical_expr::create_physical_expr(

datafusion/catalog/src/table.rs

@@ -84,10 +84,10 @@ pub trait TableProvider: Debug + Sync + Send {
         None
     }
 
-    /// Create an [`ExecutionPlan`] for scanning the table with optionally
-    /// specified `projection`, `filter` and `limit`, described below.
+    /// Create an [`ExecutionPlan`] for scanning the table with optional
+    /// `projection`, `filter`, and `limit`, described below.
     ///
-    /// The `ExecutionPlan` is responsible scanning the datasource's
+    /// The returned `ExecutionPlan` is responsible for scanning the datasource's
     /// partitions in a streaming, parallelized fashion.
     ///
     /// # Projection
@@ -96,33 +96,30 @@ pub trait TableProvider: Debug + Sync + Send {
     /// specified. The projection is a set of indexes of the fields in
     /// [`Self::schema`].
     ///
-    /// DataFusion provides the projection to scan only the columns actually
-    /// used in the query to improve performance, an optimization  called
-    /// "Projection Pushdown". Some datasources, such as Parquet, can use this
-    /// information to go significantly faster when only a subset of columns is
-    /// required.
+    /// DataFusion provides the projection so the scan reads only the columns
+    /// actually used in the query, an optimization called "Projection
+    /// Pushdown". Some datasources, such as Parquet, can use this information
+    /// to go significantly faster when only a subset of columns is required.
     ///
     /// # Filters
     ///
     /// A list of boolean filter [`Expr`]s to evaluate *during* the scan, in the
     /// manner specified by [`Self::supports_filters_pushdown`]. Only rows for
-    /// which *all* of the `Expr`s evaluate to `true` must be returned (aka the
-    /// expressions are `AND`ed together).
+    /// which *all* of the `Expr`s evaluate to `true` must be returned (that is,
+    /// the expressions are `AND`ed together).
     ///
-    /// To enable filter pushdown you must override
-    /// [`Self::supports_filters_pushdown`] as the default implementation does
-    /// not and `filters` will be empty.
+    /// To enable filter pushdown, override
+    /// [`Self::supports_filters_pushdown`]. The default implementation does not
+    /// push down filters, and `filters` will be empty.
     ///
-    /// DataFusion pushes filtering into the scans whenever possible
-    /// ("Filter Pushdown"), and depending on the format and the
-    /// implementation of the format, evaluating the predicate during the scan
-    /// can increase performance significantly.
+    /// DataFusion pushes filters into scans whenever possible ("Filter
+    /// Pushdown"). Depending on the data format and implementation, evaluating
+    /// predicates during the scan can significantly improve performance.
     ///
     /// ## Note: Some columns may appear *only* in Filters
     ///
-    /// In certain cases, a query may only use a certain column in a Filter that
-    /// has been completely pushed down to the scan. In this case, the
-    /// projection will not contain all the columns found in the filter
+    /// In some cases, a query may use a column only in a filter and the
+    /// projection will not contain all columns referenced by the filter
     /// expressions.
     ///
     /// For example, given the query `SELECT t.a FROM t WHERE t.b > 5`,
@@ -154,15 +151,40 @@ pub trait TableProvider: Debug + Sync + Send {
     ///
     /// # Limit
     ///
-    /// If `limit` is specified,  must only produce *at least* this many rows,
-    /// (though it may return more).  Like Projection Pushdown and Filter
-    /// Pushdown, DataFusion pushes `LIMIT`s  as far down in the plan as
-    /// possible, called "Limit Pushdown" as some sources can use this
-    /// information to improve their performance. Note that if there are any
-    /// Inexact filters pushed down, the LIMIT cannot be pushed down. This is
-    /// because inexact filters do not guarantee that every filtered row is
-    /// removed, so applying the limit could lead to too few rows being available
-    /// to return as a final result.
+    /// If `limit` is specified, the scan must produce *at least* this many
+    /// rows, though it may return more. Like Projection Pushdown and Filter
+    /// Pushdown, DataFusion pushes `LIMIT`s as far down in the plan as
+    /// possible. This is called "Limit Pushdown", and some sources can use the
+    /// information to improve performance.
+    ///
+    /// Note: If any pushed-down filters are `Inexact`, the `LIMIT` cannot be
+    /// pushed down. Inexact filters do not guarantee that every filtered row is
+    /// removed, so applying the limit could leave too few rows to return in the
+    /// final result.
+    ///
+    /// # Evaluation Order
+    ///
+    /// The logical evaluation order is `filters`, then `limit`, then
+    /// `projection`.
+    ///
+    /// Note that `limit` applies to the filtered result, not to the unfiltered
+    /// input, and `projection` affects only which columns are returned, not
+    /// which rows qualify.
+    ///
+    /// For example, if a scan receives:
+    ///
+    /// - `projection = [a]`
+    /// - `filters = [b > 5]`
+    /// - `limit = Some(3)`
+    ///
+    /// It must logically produce results equivalent to:
+    ///
+    /// ```text
+    /// PROJECTION a (LIMIT 3 (SCAN WHERE b > 5))
+    /// ```
+    ///
+    /// As noted above, columns referenced only by pushed-down filters may be
+    /// absent from `projection`.
     async fn scan(
         &self,
         state: &dyn Session,