Merge remote-tracking branch 'apache/main' into upgrade-plugins

andygrove · andygrove · commit 4f46d039e791 · 2025-09-29T21:00:59.000-06:00
diff --git a/.github/workflows/benchmark-tpcds.yml b/.github/workflows/benchmark-tpcds.yml
@@ -63,7 +63,8 @@ jobs:
           key: ${{ runner.os }}-java-maven-${{ hashFiles('**/pom.xml') }}
           restore-keys: |
             ${{ runner.os }}-java-maven-
-
+      - name: Build Comet
+        run: make release
       - name: Cache TPC-DS generated data
         id: cache-tpcds-sf-1
         uses: actions/cache@v4
@@ -76,17 +77,6 @@ jobs:
         with:
           repository: databricks/tpcds-kit
           path: ./tpcds-kit
-      - name: Build Comet
-        run: make release
-      - name: Upload Comet native lib
-        uses: actions/upload-artifact@v4
-        with:
-          name: libcomet-${{ github.run_id }}
-          path: |
-            native/target/release/libcomet.so
-            native/target/release/libcomet.dylib
-          retention-days: 1 # remove the artifact after 1 day, only valid for this workflow
-          overwrite: true
       - name: Build tpcds-kit
         if: steps.cache-tpcds-sf-1.outputs.cache-hit != 'true'
         run: |
@@ -132,11 +122,8 @@ jobs:
           path: ./tpcds-sf-1
           key: tpcds-${{ hashFiles('.github/workflows/benchmark.yml') }}
           fail-on-cache-miss: true # it's always be cached as it should be generated by pre-step if not existed
-      - name: Download Comet native lib
-        uses: actions/download-artifact@v5
-        with:
-          name: libcomet-${{ github.run_id }}
-          path: native/target/release
+      - name: Build Comet
+        run: make release
       - name: Run TPC-DS queries (Sort merge join)
         if: matrix.join == 'sort_merge'
         run: |
diff --git a/.github/workflows/benchmark-tpch.yml b/.github/workflows/benchmark-tpch.yml
@@ -71,15 +71,6 @@ jobs:
           key: tpch-${{ hashFiles('.github/workflows/benchmark-tpch.yml') }}
       - name: Build Comet
         run: make release
-      - name: Upload Comet native lib
-        uses: actions/upload-artifact@v4
-        with:
-          name: libcomet-${{ github.run_id }}
-          path: |
-            native/target/release/libcomet.so
-            native/target/release/libcomet.dylib
-          retention-days: 1 # remove the artifact after 1 day, only valid for this workflow
-          overwrite: true
       - name: Generate TPC-H (SF=1) table data
         if: steps.cache-tpch-sf-1.outputs.cache-hit != 'true'
         run: |
@@ -115,11 +106,8 @@ jobs:
           path: ./tpch
           key: tpch-${{ hashFiles('.github/workflows/benchmark-tpch.yml') }}
           fail-on-cache-miss: true # it's always be cached as it should be generated by pre-step if not existed
-      - name: Download Comet native lib
-        uses: actions/download-artifact@v5
-        with:
-          name: libcomet-${{ github.run_id }}
-          path: native/target/release
+      - name: Build Comet
+        run: make release
       - name: Run TPC-H queries
         run: |
           SPARK_HOME=`pwd` SPARK_TPCH_DATA=`pwd`/tpch/sf1_parquet ./mvnw -B -Prelease -Dsuites=org.apache.spark.sql.CometTPCHQuerySuite test
diff --git a/dev/benchmarks/README.md b/dev/benchmarks/README.md
@@ -22,6 +22,10 @@ under the License.
 This directory contains scripts used for generating benchmark results that are published in this repository and in 
 the Comet documentation.
 
+For full instructions on running these benchmarks on an EC2 instance, see the [Comet Benchmarking on EC2 Guide].
+
+[Comet Benchmarking on EC2 Guide]: https://datafusion.apache.org/comet/contributor-guide/benchmarking_aws_ec2.html
+
 ## Example usage
 
 Set Spark environment variables:
@@ -50,7 +54,7 @@ Run Comet benchmark:
 
 ```shell
 export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
-export COMET_JAR=/opt/comet/comet-spark-spark3.5_2.12-0.9.0.jar
+export COMET_JAR=/opt/comet/comet-spark-spark3.5_2.12-0.10.0.jar
 sudo ./drop-caches.sh
 ./comet-tpch.sh
 ```
diff --git a/dev/benchmarks/blaze-tpcds.sh b/dev/benchmarks/blaze-tpcds.sh
@@ -42,6 +42,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.shuffle.manager=org.apache.spark.sql.execution.blaze.shuffle.BlazeShuffleManager \
     --conf spark.blaze.enable=true \
     --conf spark.blaze.forceShuffledHashJoin=true \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name blaze \
     --benchmark tpcds \
diff --git a/dev/benchmarks/blaze-tpch.sh b/dev/benchmarks/blaze-tpch.sh
@@ -42,6 +42,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.shuffle.manager=org.apache.spark.sql.execution.blaze.shuffle.BlazeShuffleManager \
     --conf spark.blaze.enable=true \
     --conf spark.blaze.forceShuffledHashJoin=true \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name blaze \
     --benchmark tpch \
diff --git a/dev/benchmarks/comet-tpcds.sh b/dev/benchmarks/comet-tpcds.sh
@@ -41,7 +41,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.plugins=org.apache.spark.CometPlugin \
     --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
     --conf spark.comet.expression.allowIncompatible=true \
-    --conf spark.comet.scan.impl=native_datafusion \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name comet \
     --benchmark tpcds \
diff --git a/dev/benchmarks/comet-tpch.sh b/dev/benchmarks/comet-tpch.sh
@@ -42,7 +42,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
     --conf spark.comet.exec.replaceSortMergeJoin=true \
     --conf spark.comet.expression.allowIncompatible=true \
-    --conf spark.comet.scan.impl=native_datafusion \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name comet \
     --benchmark tpch \
diff --git a/dev/benchmarks/gluten-tpcds.sh b/dev/benchmarks/gluten-tpcds.sh
@@ -42,6 +42,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.gluten.sql.columnar.forceShuffledHashJoin=true \
     --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
     --conf spark.sql.session.timeZone=UTC \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name gluten \
     --benchmark tpcds \
diff --git a/dev/benchmarks/gluten-tpch.sh b/dev/benchmarks/gluten-tpch.sh
@@ -42,6 +42,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.gluten.sql.columnar.forceShuffledHashJoin=true \
     --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
     --conf spark.sql.session.timeZone=UTC \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name gluten \
     --benchmark tpch \
diff --git a/dev/benchmarks/spark-tpcds.sh b/dev/benchmarks/spark-tpcds.sh
@@ -34,6 +34,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.memory.offHeap.enabled=true \
     --conf spark.memory.offHeap.size=16g \
     --conf spark.eventLog.enabled=true \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name spark \
     --benchmark tpcds \
diff --git a/dev/benchmarks/spark-tpch.sh b/dev/benchmarks/spark-tpch.sh
@@ -34,6 +34,8 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.memory.offHeap.enabled=true \
     --conf spark.memory.offHeap.size=16g \
     --conf spark.eventLog.enabled=true \
+    --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
+    --conf spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain \
     tpcbench.py \
     --name spark \
     --benchmark tpch \
diff --git a/dev/benchmarks/tpcbench.py b/dev/benchmarks/tpcbench.py
@@ -21,7 +21,7 @@
 from pyspark.sql import SparkSession
 import time
 
-def main(benchmark: str, data_path: str, query_path: str, iterations: int, output: str, name: str):
+def main(benchmark: str, data_path: str, query_path: str, iterations: int, output: str, name: str, query_num: int = None):
 
     # Initialize a SparkSession
     spark = SparkSession.builder \
@@ -59,9 +59,17 @@ def main(benchmark: str, data_path: str, query_path: str, iterations: int, outpu
 
     for iteration in range(0, iterations):
         print(f"Starting iteration {iteration} of {iterations}")
-        iter_start_time = time.time()
 
-        for query in range(1, num_queries+1):
+        # Determine which queries to run
+        if query_num is not None:
+            # Validate query number
+            if query_num < 1 or query_num > num_queries:
+                raise ValueError(f"Query number {query_num} is out of range. Valid range is 1-{num_queries} for {benchmark}")
+            queries_to_run = [query_num]
+        else:
+            queries_to_run = range(1, num_queries+1)
+
+        for query in queries_to_run:
             spark.sparkContext.setJobDescription(f"{benchmark} q{query}")
 
             # read text file
@@ -105,8 +113,6 @@ def main(benchmark: str, data_path: str, query_path: str, iterations: int, outpu
     # Stop the SparkSession
     spark.stop()
 
-    #print(str)
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="DataFusion benchmark derived from TPC-H / TPC-DS")
     parser.add_argument("--benchmark", required=True, help="Benchmark to run (tpch or tpcds)")
@@ -115,6 +121,7 @@ def main(benchmark: str, data_path: str, query_path: str, iterations: int, outpu
     parser.add_argument("--iterations", required=False, default="1", help="How many iterations to run")
     parser.add_argument("--output", required=True, help="Path to write output")
     parser.add_argument("--name", required=True, help="Prefix for result file e.g. spark/comet/gluten")
+    parser.add_argument("--query", required=False, type=int, help="Specific query number to run (1-based). If not specified, all queries will be run.")
     args = parser.parse_args()
 
-    main(args.benchmark, args.data, args.queries, int(args.iterations), args.output, args.name)
+    main(args.benchmark, args.data, args.queries, int(args.iterations), args.output, args.name, args.query)
diff --git a/docs/source/contributor-guide/benchmarking_aws_ec2.md b/docs/source/contributor-guide/benchmarking_aws_ec2.md
diff --git a/docs/source/user-guide/latest/compatibility.md b/docs/source/user-guide/latest/compatibility.md
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala
