Merge branch 'apache:main' into dev

Author: comphead

.github/actions/setup-spark-builder/action.yaml

@@ -29,7 +29,7 @@ inputs:
   comet-version:
     description: 'The Comet version to use for Spark'
     required: true
-    default: '0.4.0-SNAPSHOT'
+    default: '0.5.0-SNAPSHOT'
 runs:
   using: "composite"
   steps:

.github/workflows/spark_sql_test.yml

@@ -71,7 +71,7 @@ jobs:
         with:
           spark-version: ${{ matrix.spark-version.full }}
           spark-short-version: ${{ matrix.spark-version.short }}
-          comet-version: '0.4.0-SNAPSHOT' # TODO: get this from pom.xml
+          comet-version: '0.5.0-SNAPSHOT' # TODO: get this from pom.xml
       - name: Run Spark tests
         run: |
           cd apache-spark

.github/workflows/spark_sql_test_ansi.yml

@@ -69,7 +69,7 @@ jobs:
         with:
           spark-version: ${{ matrix.spark-version.full }}
           spark-short-version: ${{ matrix.spark-version.short }}
-          comet-version: '0.4.0-SNAPSHOT' # TODO: get this from pom.xml
+          comet-version: '0.5.0-SNAPSHOT' # TODO: get this from pom.xml
       - name: Run Spark tests
         run: |
           cd apache-spark

README.md

@@ -46,7 +46,7 @@ The following chart shows the time it takes to run the 22 TPC-H queries against
 using a single executor with 8 cores. See the [Comet Benchmarking Guide](https://datafusion.apache.org/comet/contributor-guide/benchmarking.html)
 for details of the environment used for these benchmarks.
 
-When using Comet, the overall run time is reduced from 616 seconds to 374 seconds, a 1.6x speedup, with query 1
+When using Comet, the overall run time is reduced from 615 seconds to 364 seconds, a 1.7x speedup, with query 1
 running 9x faster than Spark.
 
 Running the same queries with DataFusion standalone (without Spark) using the same number of cores results in a 3.6x 
@@ -55,21 +55,21 @@ speedup compared to Spark.
 Comet is not yet achieving full DataFusion speeds in all cases, but with future work we aim to provide a 2x-4x speedup 
 for a broader set of queries.
 
-![](docs/source/_static/images/benchmark-results/0.3.0/tpch_allqueries.png)
+![](docs/source/_static/images/benchmark-results/0.4.0/tpch_allqueries.png)
 
 Here is a breakdown showing relative performance of Spark, Comet, and DataFusion for each TPC-H query.
 
-![](docs/source/_static/images/benchmark-results/0.3.0/tpch_queries_compare.png)
+![](docs/source/_static/images/benchmark-results/0.4.0/tpch_queries_compare.png)
 
 The following charts shows how much Comet currently accelerates each query from the benchmark.
 
 ### Relative speedup
 
-![](docs/source/_static/images/benchmark-results/0.3.0/tpch_queries_speedup_rel.png)
+![](docs/source/_static/images/benchmark-results/0.4.0/tpch_queries_speedup_rel.png)
 
 ### Absolute speedup
 
-![](docs/source/_static/images/benchmark-results/0.3.0/tpch_queries_speedup_abs.png)
+![](docs/source/_static/images/benchmark-results/0.4.0/tpch_queries_speedup_abs.png)
 
 These benchmarks can be reproduced in any environment using the documentation in the 
 [Comet Benchmarking Guide](https://datafusion.apache.org/comet/contributor-guide/benchmarking.html). We encourage 
@@ -80,7 +80,7 @@ Results for our benchmark derived from TPC-DS are available in the [benchmarking
 ## Use Commodity Hardware
 
 Comet leverages commodity hardware, eliminating the need for costly hardware upgrades or
-specialized hardware accelerators, such as GPUs or FGPA. By maximizing the utilization of commodity hardware, Comet 
+specialized hardware accelerators, such as GPUs or FPGA. By maximizing the utilization of commodity hardware, Comet 
 ensures cost-effectiveness and scalability for your Spark deployments.
 
 ## Spark Compatibility

benchmarks/README.md

@@ -62,7 +62,7 @@ docker push localhost:32000/apache/datafusion-comet-tpcbench:latest
 export SPARK_MASTER=k8s://https://127.0.0.1:16443
 export COMET_DOCKER_IMAGE=localhost:32000/apache/datafusion-comet-tpcbench:latest
 # Location of Comet JAR within the Docker image
-export COMET_JAR=/opt/spark/jars/comet-spark-spark3.4_2.12-0.2.0-SNAPSHOT.jar
+export COMET_JAR=/opt/spark/jars/comet-spark-spark3.4_2.12-0.5.0-SNAPSHOT.jar
 
 $SPARK_HOME/bin/spark-submit \
     --master $SPARK_MASTER \

common/pom.xml

@@ -26,7 +26,7 @@ under the License.
   <parent>
     <groupId>org.apache.datafusion</groupId>
     <artifactId>comet-parent-spark${spark.version.short}_${scala.binary.version}</artifactId>
-    <version>0.4.0-SNAPSHOT</version>
+    <version>0.5.0-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -73,7 +73,7 @@ object CometConf extends ShimCometConf {
       "Whether to enable native scans. When this is turned on, Spark will use Comet to " +
         "read supported data sources (currently only Parquet is supported natively). Note " +
         "that to enable native vectorized execution, both this config and " +
-        "'spark.comet.exec.enabled' need to be enabled. By default, this config is true.")
+        "'spark.comet.exec.enabled' need to be enabled.")
     .booleanConf
     .createWithDefault(true)
 
@@ -82,7 +82,7 @@ object CometConf extends ShimCometConf {
       .doc(
         "Whether to enable Comet's parallel reader for Parquet files. The parallel reader reads " +
           "ranges of consecutive data in a  file in parallel. It is faster for large files and " +
-          "row groups but uses more resources. The parallel reader is enabled by default.")
+          "row groups but uses more resources.")
       .booleanConf
       .createWithDefault(true)
 
@@ -98,7 +98,7 @@ object CometConf extends ShimCometConf {
       .doc(
         "When enabled the parallel reader will try to merge ranges of data that are separated " +
           "by less than 'comet.parquet.read.io.mergeRanges.delta' bytes. Longer continuous reads " +
-          "are faster on cloud storage. The default behavior is to merge consecutive ranges.")
+          "are faster on cloud storage.")
       .booleanConf
       .createWithDefault(true)
 
@@ -115,7 +115,7 @@ object CometConf extends ShimCometConf {
       .doc("In the parallel reader, if the read ranges submitted are skewed in sizes, this " +
         "option will cause the reader to break up larger read ranges into smaller ranges to " +
         "reduce the skew. This will result in a slightly larger number of connections opened to " +
-        "the file system but may give improved performance. The option is off by default.")
+        "the file system but may give improved performance.")
       .booleanConf
       .createWithDefault(false)
 
@@ -153,7 +153,7 @@ object CometConf extends ShimCometConf {
         "native space. Note: each operator is associated with a separate config in the " +
         "format of 'spark.comet.exec.<operator_name>.enabled' at the moment, and both the " +
         "config and this need to be turned on, in order for the operator to be executed in " +
-        "native. By default, this config is true.")
+        "native.")
     .booleanConf
     .createWithDefault(true)
 
@@ -215,7 +215,7 @@ object CometConf extends ShimCometConf {
     "spark.comet.memory.overhead.factor")
     .doc(
       "Fraction of executor memory to be allocated as additional non-heap memory per executor " +
-        "process for Comet. Default value is 0.2.")
+        "process for Comet.")
     .doubleConf
     .checkValue(
       factor => factor > 0,
@@ -247,8 +247,7 @@ object CometConf extends ShimCometConf {
       "is enabled. Available modes are 'native', 'jvm', and 'auto'. " +
       "'native' is for native shuffle which has best performance in general. " +
       "'jvm' is for jvm-based columnar shuffle which has higher coverage than native shuffle. " +
-      "'auto' is for Comet to choose the best shuffle mode based on the query plan. " +
-      "By default, this config is 'auto'.")
+      "'auto' is for Comet to choose the best shuffle mode based on the query plan.")
     .internal()
     .stringConf
     .transform(_.toLowerCase(Locale.ROOT))
@@ -258,8 +257,8 @@ object CometConf extends ShimCometConf {
   val COMET_EXEC_BROADCAST_FORCE_ENABLED: ConfigEntry[Boolean] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.broadcast.enabled")
       .doc(
-        "Whether to force enabling broadcasting for Comet native operators. By default, " +
-          "this config is false. Comet broadcast feature will be enabled automatically by " +
+        "Whether to force enabling broadcasting for Comet native operators. " +
+          "Comet broadcast feature will be enabled automatically by " +
           "Comet extension. But for unit tests, we need this feature to force enabling it " +
           "for invalid cases. So this config is only used for unit test.")
       .internal()
@@ -273,34 +272,41 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
-  val COMET_EXEC_SHUFFLE_CODEC: ConfigEntry[String] = conf(
-    s"$COMET_EXEC_CONFIG_PREFIX.shuffle.codec")
+  val COMET_EXEC_SHUFFLE_COMPRESSION_CODEC: ConfigEntry[String] = conf(
+    s"$COMET_EXEC_CONFIG_PREFIX.shuffle.compression.codec")
     .doc(
-      "The codec of Comet native shuffle used to compress shuffle data. Only zstd is supported.")
+      "The codec of Comet native shuffle used to compress shuffle data. Only zstd is supported. " +
+        "Compression can be disabled by setting spark.shuffle.compress=false.")
     .stringConf
+    .checkValues(Set("zstd"))
     .createWithDefault("zstd")
 
-  val COMET_COLUMNAR_SHUFFLE_ASYNC_ENABLED: ConfigEntry[Boolean] = conf(
-    "spark.comet.columnar.shuffle.async.enabled")
-    .doc(
-      "Whether to enable asynchronous shuffle for Arrow-based shuffle. By default, this config " +
-        "is false.")
-    .booleanConf
-    .createWithDefault(false)
+  val COMET_EXEC_SHUFFLE_COMPRESSION_LEVEL: ConfigEntry[Int] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.compression.level")
+      .doc("The compression level to use when compression shuffle files.")
+      .intConf
+      .createWithDefault(1)
+
+  val COMET_COLUMNAR_SHUFFLE_ASYNC_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.columnar.shuffle.async.enabled")
+      .doc("Whether to enable asynchronous shuffle for Arrow-based shuffle.")
+      .booleanConf
+      .createWithDefault(false)
 
   val COMET_COLUMNAR_SHUFFLE_ASYNC_THREAD_NUM: ConfigEntry[Int] =
     conf("spark.comet.columnar.shuffle.async.thread.num")
-      .doc("Number of threads used for Comet async columnar shuffle per shuffle task. " +
-        "By default, this config is 3. Note that more threads means more memory requirement to " +
-        "buffer shuffle data before flushing to disk. Also, more threads may not always " +
-        "improve performance, and should be set based on the number of cores available.")
+      .doc(
+        "Number of threads used for Comet async columnar shuffle per shuffle task. " +
+          "Note that more threads means more memory requirement to " +
+          "buffer shuffle data before flushing to disk. Also, more threads may not always " +
+          "improve performance, and should be set based on the number of cores available.")
       .intConf
       .createWithDefault(3)
 
   val COMET_COLUMNAR_SHUFFLE_ASYNC_MAX_THREAD_NUM: ConfigEntry[Int] = {
     conf("spark.comet.columnar.shuffle.async.max.thread.num")
       .doc("Maximum number of threads on an executor used for Comet async columnar shuffle. " +
-        "By default, this config is 100. This is the upper bound of total number of shuffle " +
+        "This is the upper bound of total number of shuffle " +
         "threads per executor. In other words, if the number of cores * the number of shuffle " +
         "threads per task `spark.comet.columnar.shuffle.async.thread.num` is larger than " +
         "this config. Comet will use this config as the number of shuffle threads per " +
@@ -317,16 +323,17 @@ object CometConf extends ShimCometConf {
           "Higher value means more memory requirement to buffer shuffle data before " +
           "flushing to disk. As Comet uses columnar shuffle which is columnar format, " +
           "higher value usually helps to improve shuffle data compression ratio. This is " +
-          "internal config for testing purpose or advanced tuning. By default, " +
-          "this config is Int.Max.")
+          "internal config for testing purpose or advanced tuning.")
       .internal()
       .intConf
       .createWithDefault(Int.MaxValue)
 
   val COMET_COLUMNAR_SHUFFLE_MEMORY_SIZE: OptionalConfigEntry[Long] =
     conf("spark.comet.columnar.shuffle.memorySize")
+      .internal()
       .doc(
-        "The optional maximum size of the memory used for Comet columnar shuffle, in MiB. " +
+        "Test-only config. This is only used to test Comet shuffle with Spark tests. " +
+          "The optional maximum size of the memory used for Comet columnar shuffle, in MiB. " +
           "Note that this config is only used when `spark.comet.exec.shuffle.mode` is " +
           "`jvm`. Once allocated memory size reaches this config, the current batch will be " +
           "flushed to disk immediately. If this is not configured, Comet will use " +
@@ -338,17 +345,29 @@ object CometConf extends ShimCometConf {
 
   val COMET_COLUMNAR_SHUFFLE_MEMORY_FACTOR: ConfigEntry[Double] =
     conf("spark.comet.columnar.shuffle.memory.factor")
+      .internal()
       .doc(
-        "Fraction of Comet memory to be allocated per executor process for Comet shuffle. " +
+        "Test-only config. This is only used to test Comet shuffle with Spark tests. " +
+          "Fraction of Comet memory to be allocated per executor process for Comet shuffle. " +
           "Comet memory size is specified by `spark.comet.memoryOverhead` or " +
-          "calculated by `spark.comet.memory.overhead.factor` * `spark.executor.memory`. " +
-          "By default, this config is 1.0.")
+          "calculated by `spark.comet.memory.overhead.factor` * `spark.executor.memory`.")
       .doubleConf
       .checkValue(
         factor => factor > 0,
         "Ensure that Comet shuffle memory overhead factor is a double greater than 0")
       .createWithDefault(1.0)
 
+  val COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST: ConfigEntry[Boolean] =
+    conf("spark.comet.columnar.shuffle.unifiedMemoryAllocatorTest")
+      .doc("Whether to use Spark unified memory allocator for Comet columnar shuffle in tests." +
+        "If not configured, Comet will use a test-only memory allocator for Comet columnar " +
+        "shuffle when Spark test env detected. The test-ony allocator is proposed to run with " +
+        "Spark tests as these tests require on-heap memory configuration. " +
+        "By default, this config is false.")
+      .internal()
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_COLUMNAR_SHUFFLE_BATCH_SIZE: ConfigEntry[Int] =
     conf("spark.comet.columnar.shuffle.batch.size")
       .internal()
@@ -360,11 +379,12 @@ object CometConf extends ShimCometConf {
 
   val COMET_SHUFFLE_PREFER_DICTIONARY_RATIO: ConfigEntry[Double] = conf(
     "spark.comet.shuffle.preferDictionary.ratio")
-    .doc("The ratio of total values to distinct values in a string column to decide whether to " +
-      "prefer dictionary encoding when shuffling the column. If the ratio is higher than " +
-      "this config, dictionary encoding will be used on shuffling string column. This config " +
-      "is effective if it is higher than 1.0. By default, this config is 10.0. Note that this " +
-      "config is only used when `spark.comet.exec.shuffle.mode` is `jvm`.")
+    .doc(
+      "The ratio of total values to distinct values in a string column to decide whether to " +
+        "prefer dictionary encoding when shuffling the column. If the ratio is higher than " +
+        "this config, dictionary encoding will be used on shuffling string column. This config " +
+        "is effective if it is higher than 1.0. Note that this " +
+        "config is only used when `spark.comet.exec.shuffle.mode` is `jvm`.")
     .doubleConf
     .createWithDefault(10.0)
 
@@ -377,7 +397,7 @@ object CometConf extends ShimCometConf {
   val COMET_DEBUG_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.debug.enabled")
       .doc(
-        "Whether to enable debug mode for Comet. By default, this config is false. " +
+        "Whether to enable debug mode for Comet. " +
           "When enabled, Comet will do additional checks for debugging purpose. For example, " +
           "validating array when importing arrays from JVM at native side. Note that these " +
           "checks may be expensive in performance and should only be enabled for debugging " +
@@ -437,27 +457,27 @@ object CometConf extends ShimCometConf {
       "The fraction of memory from Comet memory overhead that the native memory " +
         "manager can use for execution. The purpose of this config is to set aside memory for " +
         "untracked data structures, as well as imprecise size estimation during memory " +
-        "acquisition. Default value is 0.7.")
+        "acquisition.")
     .doubleConf
     .createWithDefault(0.7)
 
-  val COMET_PARQUET_ENABLE_DIRECT_BUFFER: ConfigEntry[Boolean] = conf(
-    "spark.comet.parquet.enable.directBuffer")
-    .doc("Whether to use Java direct byte buffer when reading Parquet. By default, this is false")
-    .booleanConf
-    .createWithDefault(false)
+  val COMET_PARQUET_ENABLE_DIRECT_BUFFER: ConfigEntry[Boolean] =
+    conf("spark.comet.parquet.enable.directBuffer")
+      .doc("Whether to use Java direct byte buffer when reading Parquet.")
+      .booleanConf
+      .createWithDefault(false)
 
   val COMET_SCAN_PREFETCH_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.scan.preFetch.enabled")
-      .doc("Whether to enable pre-fetching feature of CometScan. By default is disabled.")
+      .doc("Whether to enable pre-fetching feature of CometScan.")
       .booleanConf
       .createWithDefault(false)
 
   val COMET_SCAN_PREFETCH_THREAD_NUM: ConfigEntry[Int] =
     conf("spark.comet.scan.preFetch.threadNum")
       .doc(
         "The number of threads running pre-fetching for CometScan. Effective if " +
-          s"${COMET_SCAN_PREFETCH_ENABLED.key} is enabled. By default it is 2. Note that more " +
+          s"${COMET_SCAN_PREFETCH_ENABLED.key} is enabled. Note that more " +
           "pre-fetching threads means more memory requirement to store pre-fetched row groups.")
       .intConf
       .createWithDefault(2)