Merge branch 'main' into feature/add-string-split-support

Shekharrajak · web-flow · commit dbb34d527628 · 2025-11-17T12:14:41.000+05:30
diff --git a/.github/workflows/pr_markdown_format.yml b/.github/workflows/pr_markdown_format.yml
@@ -0,0 +1,49 @@
+# 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.
+
+name: Check Markdown Formatting
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}
+  cancel-in-progress: true
+
+on:
+  pull_request:
+    paths:
+      - '**.md'
+
+jobs:
+  prettier-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v6
+        with:
+          node-version: '24'
+
+      - name: Install prettier
+        run: npm install -g prettier
+
+      - name: Check markdown formatting
+        run: |
+          # if you encounter error, run prettier locally and commit changes using instructions at:
+          #
+          # https://datafusion.apache.org/comet/contributor-guide/development.html#how-to-format-md-document
+          #
+          prettier --check "**/*.md"
diff --git a/common/src/main/scala/org/apache/comet/CometConf.scala b/common/src/main/scala/org/apache/comet/CometConf.scala
@@ -666,14 +666,6 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
-  val COMET_EXPR_ALLOW_INCOMPATIBLE: ConfigEntry[Boolean] =
-    conf("spark.comet.expression.allowIncompatible")
-      .category(CATEGORY_EXEC)
-      .doc("Comet is not currently fully compatible with Spark for all expressions. " +
-        s"Set this config to true to allow them anyway. $COMPAT_GUIDE.")
-      .booleanConf
-      .createWithDefault(false)
-
   val COMET_EXEC_STRICT_FLOATING_POINT: ConfigEntry[Boolean] =
     conf("spark.comet.exec.strictFloatingPoint")
       .category(CATEGORY_EXEC)
diff --git a/dev/benchmarks/comet-tpcds.sh b/dev/benchmarks/comet-tpcds.sh
@@ -40,7 +40,7 @@ $SPARK_HOME/bin/spark-submit \
     --conf spark.executor.extraClassPath=$COMET_JAR \
     --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.expression.Cast.allowIncompatible=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 \
diff --git a/dev/benchmarks/comet-tpch.sh b/dev/benchmarks/comet-tpch.sh
@@ -41,7 +41,7 @@ $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.exec.replaceSortMergeJoin=true \
-    --conf spark.comet.expression.allowIncompatible=true \
+    --conf spark.comet.expression.Cast.allowIncompatible=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 \
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/configs.md b/docs/source/user-guide/latest/configs.md
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
@@ -31,9 +31,6 @@ of expressions that be disabled.
 Expressions that are not Spark-compatible will fall back to Spark by default and can be enabled by setting
 `spark.comet.expression.EXPRNAME.allowIncompatible=true`.
 
-It is also possible to specify `spark.comet.expression.allowIncompatible=true` to enable all
-incompatible expressions.
-
 ## Conditional Expressions
 
 | Expression | SQL                                         | Spark-Compatible? |
diff --git a/docs/source/user-guide/latest/kubernetes.md b/docs/source/user-guide/latest/kubernetes.md
@@ -79,7 +79,6 @@ spec:
     "spark.plugins": "org.apache.spark.CometPlugin"
     "spark.comet.enabled": "true"
     "spark.comet.exec.enabled": "true"
-    "spark.comet.expression.allowIncompatible": "true"
     "spark.comet.exec.shuffle.enabled": "true"
     "spark.comet.exec.shuffle.mode": "auto"
     "spark.shuffle.manager": "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager"
diff --git a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala
@@ -52,6 +52,7 @@ object GenerateDocs {
       w.write(s"${line.stripTrailing()}\n".getBytes)
       line match {
         case pattern(category) =>
+          w.write("<!-- prettier-ignore-start -->\n".getBytes)
           w.write("| Config | Description | Default Value |\n".getBytes)
           w.write("|--------|-------------|---------------|\n".getBytes)
           category match {
@@ -61,12 +62,14 @@ object GenerateDocs {
                 w.write(
                   s"| `$config` | Enable Comet acceleration for `$expr` | true |\n".getBytes)
               }
+              w.write("<!-- prettier-ignore-end -->\n".getBytes)
             case "enable_agg_expr" =>
               for (expr <- QueryPlanSerde.aggrSerdeMap.keys.map(_.getSimpleName).toList.sorted) {
                 val config = s"spark.comet.expression.$expr.enabled"
                 w.write(
                   s"| `$config` | Enable Comet acceleration for `$expr` | true |\n".getBytes)
               }
+              w.write("<!-- prettier-ignore-end -->\n".getBytes)
             case _ =>
               val urlPattern = """Comet\s+(Compatibility|Tuning|Tracing)\s+Guide\s+\(""".r
               val confs = publicConfigs.filter(_.category == category).toList.sortBy(_.key)
@@ -93,6 +96,7 @@ object GenerateDocs {
                   }
                 }
               }
+              w.write("<!-- prettier-ignore-end -->\n".getBytes)
           }
         case _ =>
       }
@@ -106,6 +110,7 @@ object GenerateDocs {
     for (line <- lines) {
       w.write(s"${line.stripTrailing()}\n".getBytes)
       if (line.trim == "<!--BEGIN:COMPAT_CAST_TABLE-->") {
+        w.write("<!-- prettier-ignore-start -->\n".getBytes)
         w.write("| From Type | To Type | Notes |\n".getBytes)
         w.write("|-|-|-|\n".getBytes)
         for (fromType <- CometCast.supportedTypes) {
@@ -123,7 +128,9 @@ object GenerateDocs {
             }
           }
         }
+        w.write("<!-- prettier-ignore-end -->\n".getBytes)
       } else if (line.trim == "<!--BEGIN:INCOMPAT_CAST_TABLE-->") {
+        w.write("<!-- prettier-ignore-start -->\n".getBytes)
         w.write("| From Type | To Type | Notes |\n".getBytes)
         w.write("|-|-|-|\n".getBytes)
         for (fromType <- CometCast.supportedTypes) {
@@ -140,6 +147,7 @@ object GenerateDocs {
             }
           }
         }
+        w.write("<!-- prettier-ignore-end -->\n".getBytes)
       }
     }
     w.close()
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -93,10 +93,9 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
         castBuilder.setDatatype(dataType)
         castBuilder.setEvalMode(evalModeToProto(evalMode))
         castBuilder.setAllowIncompat(
-          CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.get() ||
-            SQLConf.get
-              .getConfString(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]), "false")
-              .toBoolean)
+          SQLConf.get
+            .getConfString(CometConf.getExprAllowIncompatConfigKey(classOf[Cast]), "false")
+            .toBoolean)
         castBuilder.setTimezone(timeZoneId.getOrElse("UTC"))
         Some(
           ExprOuterClass.Expr
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -405,10 +405,11 @@ object QueryPlanSerde extends Logging with CometExprShim {
             None
           case Incompatible(notes) =>
             val exprAllowIncompat = CometConf.isExprAllowIncompat(exprConfName)
-            if (exprAllowIncompat || CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.get()) {
+            if (exprAllowIncompat) {
               if (notes.isDefined) {
                 logWarning(
-                  s"Comet supports $fn when ${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true " +
+                  s"Comet supports $fn when " +
+                    s"${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true " +
                     s"but has notes: ${notes.get}")
               }
               aggHandler.convert(aggExpr, fn, inputs, binding, conf)
@@ -417,9 +418,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
               withInfo(
                 fn,
                 s"$fn is not fully compatible with Spark$optionalNotes. To enable it anyway, " +
-                  s"set ${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true, or set " +
-                  s"${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true to enable all " +
-                  s"incompatible expressions. ${CometConf.COMPAT_GUIDE}.")
+                  s"set ${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true. " +
+                  s"${CometConf.COMPAT_GUIDE}.")
               None
             }
           case Compatible(notes) =>
@@ -510,10 +510,11 @@ object QueryPlanSerde extends Logging with CometExprShim {
           None
         case Incompatible(notes) =>
           val exprAllowIncompat = CometConf.isExprAllowIncompat(exprConfName)
-          if (exprAllowIncompat || CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.get()) {
+          if (exprAllowIncompat) {
             if (notes.isDefined) {
               logWarning(
-                s"Comet supports $expr when ${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true " +
+                s"Comet supports $expr when " +
+                  s"${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true " +
                   s"but has notes: ${notes.get}")
             }
             handler.convert(expr, inputs, binding)
@@ -522,9 +523,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
             withInfo(
               expr,
               s"$expr is not fully compatible with Spark$optionalNotes. To enable it anyway, " +
-                s"set ${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true, or set " +
-                s"${CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key}=true to enable all " +
-                s"incompatible expressions. ${CometConf.COMPAT_GUIDE}.")
+                s"set ${CometConf.getExprAllowIncompatConfigKey(exprConfName)}=true. " +
+                s"${CometConf.COMPAT_GUIDE}.")
             None
           }
         case Compatible(notes) =>
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -23,6 +23,7 @@ import scala.util.Random
 
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.{CometTestBase, DataFrame, Row}
+import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.catalyst.optimizer.EliminateSorts
 import org.apache.spark.sql.comet.CometHashAggregateExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -1073,7 +1074,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
         withSQLConf(
           CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> nativeShuffleEnabled.toString,
           CometConf.COMET_SHUFFLE_MODE.key -> "native",
-          CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key -> "true") {
+          CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true") {
           withTempDir { dir =>
             val path = new Path(dir.toURI.toString, "test")
             makeParquetFile(path, 1000, 20, dictionaryEnabled)
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala
@@ -28,7 +28,8 @@ import org.apache.commons.io.FileUtils
 import org.apache.spark.SparkContext
 import org.apache.spark.internal.config.{MEMORY_OFFHEAP_ENABLED, MEMORY_OFFHEAP_SIZE}
 import org.apache.spark.sql.TPCDSBase
-import org.apache.spark.sql.catalyst.expressions.AttributeSet
+import org.apache.spark.sql.catalyst.expressions.{AttributeSet, Cast}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{Average, Sum}
 import org.apache.spark.sql.catalyst.util.resourceToString
 import org.apache.spark.sql.execution.{FormattedMode, ReusedSubqueryExec, SparkPlan, SubqueryBroadcastExec, SubqueryExec}
 import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
@@ -223,9 +224,11 @@ trait CometPlanStabilitySuite extends DisableAdaptiveExecutionSuite with TPCDSBa
       CometConf.COMET_DPP_FALLBACK_ENABLED.key -> "false",
       CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
       CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key -> "true",
-      // COMET_EXPR_ALLOW_INCOMPATIBLE is needed for Spark 4.0.0 / ANSI support
+      // Allow Incompatible is needed for Sum + Average for Spark 4.0.0 / ANSI support
+      CometConf.getExprAllowIncompatConfigKey(classOf[Average]) -> "true",
+      CometConf.getExprAllowIncompatConfigKey(classOf[Sum]) -> "true",
       // as well as for v1.4/q9, v1.4/q44, v2.7.0/q6, v2.7.0/q64
-      CometConf.COMET_EXPR_ALLOW_INCOMPATIBLE.key -> "true",
+      CometConf.getExprAllowIncompatConfigKey(classOf[Cast]) -> "true",
       SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
       val qe = sql(queryString).queryExecution
       val plan = qe.executedPlan

Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,7 @@ object GenerateDocs {`
`52`	`52`	`w.write(s"${line.stripTrailing()}\n".getBytes)`
`53`	`53`	`line match {`
`54`	`54`	`case pattern(category) =>`
	`55`	`+ w.write("<!-- prettier-ignore-start -->\n".getBytes)`
`55`	`56`	`w.write("\| Config \| Description \| Default Value \|\n".getBytes)`
`56`	`57`	`w.write("\|--------\|-------------\|---------------\|\n".getBytes)`
`57`	`58`	`category match {`
`@@ -61,12 +62,14 @@ object GenerateDocs {`
`61`	`62`	`w.write(`
`62`	`63`	s"\| `$config` \| Enable Comet acceleration for `$expr` \| true \|\n".getBytes)
`63`	`64`	`}`
	`65`	`+ w.write("<!-- prettier-ignore-end -->\n".getBytes)`
`64`	`66`	`case "enable_agg_expr" =>`
`65`	`67`	`for (expr <- QueryPlanSerde.aggrSerdeMap.keys.map(_.getSimpleName).toList.sorted) {`
`66`	`68`	`val config = s"spark.comet.expression.$expr.enabled"`
`67`	`69`	`w.write(`
`68`	`70`	s"\| `$config` \| Enable Comet acceleration for `$expr` \| true \|\n".getBytes)
`69`	`71`	`}`
	`72`	`+ w.write("<!-- prettier-ignore-end -->\n".getBytes)`
`70`	`73`	`case _ =>`
`71`	`74`	`val urlPattern = """Comet\s+(Compatibility\|Tuning\|Tracing)\s+Guide\s+\(""".r`
`72`	`75`	`val confs = publicConfigs.filter(_.category == category).toList.sortBy(_.key)`
`@@ -93,6 +96,7 @@ object GenerateDocs {`
`93`	`96`	`}`
`94`	`97`	`}`
`95`	`98`	`}`
	`99`	`+ w.write("<!-- prettier-ignore-end -->\n".getBytes)`
`96`	`100`	`}`
`97`	`101`	`case _ =>`
`98`	`102`	`}`
`@@ -106,6 +110,7 @@ object GenerateDocs {`
`106`	`110`	`for (line <- lines) {`
`107`	`111`	`w.write(s"${line.stripTrailing()}\n".getBytes)`
`108`	`112`	`if (line.trim == "<!--BEGIN:COMPAT_CAST_TABLE-->") {`
	`113`	`+ w.write("<!-- prettier-ignore-start -->\n".getBytes)`
`109`	`114`	`w.write("\| From Type \| To Type \| Notes \|\n".getBytes)`
`110`	`115`	`w.write("\|-\|-\|-\|\n".getBytes)`
`111`	`116`	`for (fromType <- CometCast.supportedTypes) {`
`@@ -123,7 +128,9 @@ object GenerateDocs {`
`123`	`128`	`}`
`124`	`129`	`}`
`125`	`130`	`}`
	`131`	`+ w.write("<!-- prettier-ignore-end -->\n".getBytes)`
`126`	`132`	`} else if (line.trim == "<!--BEGIN:INCOMPAT_CAST_TABLE-->") {`
	`133`	`+ w.write("<!-- prettier-ignore-start -->\n".getBytes)`
`127`	`134`	`w.write("\| From Type \| To Type \| Notes \|\n".getBytes)`
`128`	`135`	`w.write("\|-\|-\|-\|\n".getBytes)`
`129`	`136`	`for (fromType <- CometCast.supportedTypes) {`
`@@ -140,6 +147,7 @@ object GenerateDocs {`
`140`	`147`	`}`
`141`	`148`	`}`
`142`	`149`	`}`
	`150`	`+ w.write("<!-- prettier-ignore-end -->\n".getBytes)`
`143`	`151`	`}`
`144`	`152`	`}`
`145`	`153`	`w.close()`