Merge branch 'apache:main' into copy_to_preserve_ordering

Author: bert-beyondloops

.github/workflows/rust.yml

@@ -218,6 +218,8 @@ jobs:
         run: cargo check --profile ci --no-default-features -p datafusion --features=string_expressions
       - name: Check datafusion (unicode_expressions)
         run: cargo check --profile ci --no-default-features -p datafusion --features=unicode_expressions
+      - name: Check parquet encryption (parquet_encryption)
+        run: cargo check --profile ci --no-default-features -p datafusion --features=parquet_encryption
 
   # Check datafusion-functions crate features
   #
@@ -312,18 +314,6 @@ jobs:
           fetch-depth: 1
       - name: Setup Rust toolchain
         run: rustup toolchain install stable
-      - name: Setup Minio - S3-compatible storage
-        run: |
-          docker run -d --name minio-container \
-            -p 9000:9000 \
-            -e MINIO_ROOT_USER=TEST-DataFusionLogin -e MINIO_ROOT_PASSWORD=TEST-DataFusionPassword \
-            -v $(pwd)/datafusion/core/tests/data:/source quay.io/minio/minio \
-            server /data
-          docker exec minio-container /bin/sh -c "\
-            mc ready local
-            mc alias set localminio http://localhost:9000 TEST-DataFusionLogin TEST-DataFusionPassword && \
-            mc mb localminio/data && \
-            mc cp -r /source/* localminio/data"
       - name: Run tests (excluding doctests)
         env:
           RUST_BACKTRACE: 1
@@ -335,9 +325,6 @@ jobs:
         run: cargo test --profile ci -p datafusion-cli  --lib --tests --bins
       - name: Verify Working Directory Clean
         run: git diff --exit-code
-      - name: Minio Output
-        if: ${{ !cancelled() }}
-        run: docker logs minio-container
 
 
   linux-test-example:
@@ -769,10 +756,15 @@ jobs:
           # `rust-version` key of `Cargo.toml`.
           #
           # To reproduce:
-          # 1. Install the version of Rust that is failing. Example:
-          #    rustup install 1.80.1
-          # 2. Run the command that failed with that version. Example:
-          #    cargo +1.80.1 check -p datafusion
+          # 1. Install the version of Rust that is failing.
+          # 2. Run the command that failed with that version.
+          #
+          # Example:
+          #    # MSRV looks like "1.80.0" and is specified in Cargo.toml. We can read the value with the following command:
+          #    msrv="$(cargo metadata --format-version=1 | jq '.packages[] | select( .name == "datafusion" ) | .rust_version' -r)"
+          #    echo "MSRV: ${msrv}"
+          #    rustup install "${msrv}"
+          #    cargo "+${msrv}" check
           #
           # To resolve, either:
           # 1. Change your code to use older Rust features,

Cargo.lock

@@ -150,6 +150,7 @@ env_logger = "0.11"
 futures = "0.3"
 half = { version = "2.6.0", default-features = false }
 hashbrown = { version = "0.14.5", features = ["raw"] }
+hex = { version = "0.4.3" }
 indexmap = "2.10.0"
 itertools = "0.14"
 log = "^0.4"
@@ -173,6 +174,8 @@ rstest = "0.25.0"
 serde_json = "1"
 sqlparser = { version = "0.55.0", default-features = false, features = ["std", "visitor"] }
 tempfile = "3"
+testcontainers = { version = "0.24", features = ["default"] }
+testcontainers-modules = { version = "0.12" }
 tokio = { version = "1.46", features = ["macros", "rt", "sync"] }
 url = "2.5.4"
 

Cargo.toml

@@ -120,6 +120,7 @@ Default features:
 - `datetime_expressions`: date and time functions such as `to_timestamp`
 - `encoding_expressions`: `encode` and `decode` functions
 - `parquet`: support for reading the [Apache Parquet] format
+- `parquet_encryption`: support for using [Parquet Modular Encryption]
 - `regex_expressions`: regular expression functions, such as `regexp_match`
 - `unicode_expressions`: Include unicode aware functions such as `character_length`
 - `unparser`: enables support to reverse LogicalPlans back into SQL
@@ -134,6 +135,7 @@ Optional features:
 
 [apache avro]: https://avro.apache.org/
 [apache parquet]: https://parquet.apache.org/
+[parquet modular encryption]: https://parquet.apache.org/docs/file-format/data-pages/encryption/
 
 ## DataFusion API Evolution and Deprecation Guidelines
 

README.md

@@ -95,7 +95,8 @@ external_aggr:          External aggregation benchmark on TPC-H dataset (SF=1)
 
 # ClickBench Benchmarks
 clickbench_1:           ClickBench queries against a single parquet file
-clickbench_partitioned: ClickBench queries against a partitioned (100 files) parquet
+clickbench_partitioned: ClickBench queries against partitioned (100 files) parquet
+clickbench_pushdown:    ClickBench queries against partitioned (100 files) parquet w/ filter_pushdown enabled
 clickbench_extended:    ClickBench \"inspired\" queries against a single parquet (DataFusion specific)
 
 # H2O.ai Benchmarks (Group By, Join, Window)
@@ -207,6 +208,9 @@ main() {
                 clickbench_partitioned)
                     data_clickbench_partitioned
                     ;;
+                clickbench_pushdown)
+                    data_clickbench_partitioned # same data as clickbench_partitioned
+                    ;;
                 clickbench_extended)
                     data_clickbench_1
                     ;;
@@ -303,6 +307,7 @@ main() {
                     run_cancellation
                     run_clickbench_1
                     run_clickbench_partitioned
+                    run_clickbench_pushdown
                     run_clickbench_extended
                     run_h2o "SMALL" "PARQUET" "groupby"
                     run_h2o "MEDIUM" "PARQUET" "groupby"
@@ -340,6 +345,9 @@ main() {
                 clickbench_partitioned)
                     run_clickbench_partitioned
                     ;;
+                clickbench_pushdown)
+                    run_clickbench_pushdown
+                    ;;
                 clickbench_extended)
                     run_clickbench_extended
                     ;;
@@ -572,14 +580,24 @@ run_clickbench_1() {
     debug_run $CARGO_COMMAND --bin dfbench -- clickbench  --iterations 5 --path "${DATA_DIR}/hits.parquet"  --queries-path "${SCRIPT_DIR}/queries/clickbench/queries" -o "${RESULTS_FILE}" ${QUERY_ARG}
 }
 
- # Runs the clickbench benchmark with the partitioned parquet files
+ # Runs the clickbench benchmark with the partitioned parquet dataset (100 files)
 run_clickbench_partitioned() {
     RESULTS_FILE="${RESULTS_DIR}/clickbench_partitioned.json"
     echo "RESULTS_FILE: ${RESULTS_FILE}"
     echo "Running clickbench (partitioned, 100 files) benchmark..."
     debug_run $CARGO_COMMAND --bin dfbench -- clickbench  --iterations 5 --path "${DATA_DIR}/hits_partitioned" --queries-path "${SCRIPT_DIR}/queries/clickbench/queries" -o "${RESULTS_FILE}" ${QUERY_ARG}
 }
 
+
+ # Runs the clickbench benchmark with the partitioned parquet files and filter_pushdown enabled
+run_clickbench_pushdown() {
+    RESULTS_FILE="${RESULTS_DIR}/clickbench_pushdown.json"
+    echo "RESULTS_FILE: ${RESULTS_FILE}"
+    echo "Running clickbench (partitioned, 100 files) benchmark with pushdown_filters=true, reorder_filters=true..."
+    debug_run $CARGO_COMMAND --bin dfbench -- clickbench --pushdown --iterations 5 --path "${DATA_DIR}/hits_partitioned" --queries-path "${SCRIPT_DIR}/queries/clickbench/queries" -o "${RESULTS_FILE}" ${QUERY_ARG}
+}
+
+
 # Runs the clickbench "extended" benchmark with a single large parquet file
 run_clickbench_extended() {
     RESULTS_FILE="${RESULTS_DIR}/clickbench_extended.json"

benchmarks/bench.sh

@@ -29,7 +29,7 @@ use datafusion_common::exec_datafusion_err;
 use datafusion_common::instant::Instant;
 use structopt::StructOpt;
 
-/// Run the clickbench benchmark
+/// Driver program to run the ClickBench benchmark
 ///
 /// The ClickBench[1] benchmarks are widely cited in the industry and
 /// focus on grouping / aggregation / filtering. This runner uses the
@@ -44,6 +44,14 @@ pub struct RunOpt {
     #[structopt(short, long)]
     query: Option<usize>,
 
+    /// If specified, enables Parquet Filter Pushdown.
+    ///
+    /// Specifically, it enables:
+    /// * `pushdown_filters = true`
+    /// * `reorder_filters = true`
+    #[structopt(long = "pushdown")]
+    pushdown: bool,
+
     /// Common options
     #[structopt(flatten)]
     common: CommonOpt,
@@ -122,6 +130,12 @@ impl RunOpt {
             // The hits_partitioned dataset specifies string columns
             // as binary due to how it was written. Force it to strings
             parquet_options.binary_as_string = true;
+
+            // Turn on Parquet filter pushdown if requested
+            if self.pushdown {
+                parquet_options.pushdown_filters = true;
+                parquet_options.reorder_filters = true;
+            }
         }
 
         let rt_builder = self.common.runtime_env_builder()?;

benchmarks/src/clickbench.rs

@@ -29,47 +29,26 @@ cargo test
 
 ## Running Storage Integration Tests
 
-By default, storage integration tests are not run. To run them you will need to set `TEST_STORAGE_INTEGRATION=1` and
-then provide the necessary configuration for that object store.
+By default, storage integration tests are not run. These test use the `testcontainers` crate to start up a local MinIO server using docker on port 9000.
 
-For some of the tests, [snapshots](https://datafusion.apache.org/contributor-guide/testing.html#snapshot-testing) are used.
-
-### AWS
-
-To test the S3 integration against [Minio](https://github.com/minio/minio)
-
-First start up a container with Minio and load test files.
+To run them you will need to set `TEST_STORAGE_INTEGRATION`:
 
 ```shell
-docker run -d \
-  --name datafusion-test-minio \
-  -p 9000:9000 \
-  -e MINIO_ROOT_USER=TEST-DataFusionLogin \
-  -e MINIO_ROOT_PASSWORD=TEST-DataFusionPassword \
-  -v $(pwd)/../datafusion/core/tests/data:/source \
-  quay.io/minio/minio server /data
-
-docker exec datafusion-test-minio /bin/sh -c "\
-  mc ready local
-  mc alias set localminio http://localhost:9000 TEST-DataFusionLogin TEST-DataFusionPassword && \
-  mc mb localminio/data && \
-  mc cp -r /source/* localminio/data"
+TEST_STORAGE_INTEGRATION=1 cargo test
 ```
 
-Setup environment
+For some of the tests, [snapshots](https://datafusion.apache.org/contributor-guide/testing.html#snapshot-testing) are used.
 
-```shell
-export TEST_STORAGE_INTEGRATION=1
-export AWS_ACCESS_KEY_ID=TEST-DataFusionLogin
-export AWS_SECRET_ACCESS_KEY=TEST-DataFusionPassword
-export AWS_ENDPOINT=http://127.0.0.1:9000
-export AWS_ALLOW_HTTP=true
-```
+### AWS
 
-Note that `AWS_ENDPOINT` is set without slash at the end.
+S3 integration is tested against [Minio](https://github.com/minio/minio) with [TestContainers](https://github.com/testcontainers/testcontainers-rs)
+This requires Docker to be running on your machine and port 9000 to be free.
 
-Run tests
+If you see an error mentioning "failed to load IMDS session token" such as
 
-```shell
-cargo test
-```
+> ---- object_storage::tests::s3_object_store_builder_resolves_region_when_none_provided stdout ----
+> Error: ObjectStore(Generic { store: "S3", source: "Error getting credentials from provider: an error occurred while loading credentials: failed to load IMDS session token" })
+
+You my need to disable trying to fetch S3 credentials from the environment using the `AWS_EC2_METADATA_DISABLED`, for example:
+
+> $ AWS_EC2_METADATA_DISABLED=true TEST_STORAGE_INTEGRATION=1 cargo test

datafusion-cli/CONTRIBUTING.md

@@ -72,3 +72,5 @@ insta = { workspace = true }
 insta-cmd = "0.6.0"
 predicates = "3.0"
 rstest = { workspace = true }
+testcontainers = { workspace = true }
+testcontainers-modules = { workspace = true, features = ["minio"] }