fix: Mark WindowsExec as incompatible (#2748)

Author: andygrove

docs/source/user-guide/latest/compatibility.md

@@ -70,7 +70,7 @@ this can be overridden by setting `spark.comet.regexp.allowIncompatible=true`.
 
 ## Window Functions
 
-Comet's support for window functions is incomplete and known to be incorrect. It is disabled by default and 
+Comet's support for window functions is incomplete and known to be incorrect. It is disabled by default and
 should not be used in production. The feature will be enabled in a future release. Tracking issue: [#2721](https://github.com/apache/datafusion-comet/issues/2721).
 
 ## Cast

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -71,7 +71,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
       classOf[ShuffledHashJoinExec] -> CometShuffleHashJoin,
       classOf[SortMergeJoinExec] -> CometSortMergeJoin,
       classOf[SortExec] -> CometSort,
-      classOf[LocalTableScanExec] -> CometLocalTableScan)
+      classOf[LocalTableScanExec] -> CometLocalTableScan,
+      classOf[WindowExec] -> CometWindow)
 
   private val arrayExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = Map(
     classOf[ArrayAppend] -> CometArrayAppend,
@@ -929,7 +930,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
    *   converted to a native operator.
    */
   def operator2Proto(op: SparkPlan, childOp: Operator*): Option[Operator] = {
-    val conf = op.conf
     val builder = OperatorOuterClass.Operator.newBuilder().setPlanId(op.id)
     childOp.foreach(builder.addChildren)
 
@@ -953,10 +953,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
       case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
         CometNativeScan.convert(scan, builder, childOp: _*)
 
-      case _: WindowExec if CometConf.COMET_EXEC_WINDOW_ENABLED.get(conf) =>
-        withInfo(op, "Window expressions are not supported")
-        None
-
       case op if isCometSink(op) =>
         val supportedTypes =
           op.output.forall(a => supportedDataType(a.dataType, allowComplex = true))
@@ -1084,7 +1080,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
       case _: CollectLimitExec => true
       case _: UnionExec => true
       case _: TakeOrderedAndProjectExec => true
-      case _: WindowExec => true
       case _ => false
     }
   }

spark/src/main/scala/org/apache/comet/serde/operator/CometWindow.scala

@@ -27,7 +27,7 @@ import org.apache.spark.sql.execution.window.WindowExec
 
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.withInfo
-import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
+import org.apache.comet.serde.{CometOperatorSerde, Incompatible, OperatorOuterClass, SupportLevel}
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.QueryPlanSerde.{exprToProto, windowExprToProto}
 
@@ -36,6 +36,10 @@ object CometWindow extends CometOperatorSerde[WindowExec] {
   override def enabledConfig: Option[ConfigEntry[Boolean]] = Some(
     CometConf.COMET_EXEC_WINDOW_ENABLED)
 
+  override def getSupportLevel(op: WindowExec): SupportLevel = {
+    Incompatible(Some("Native WindowExec has known correctness issues"))
+  }
+
   override def convert(
       op: WindowExec,
       builder: Operator.Builder,

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q12.native_iceberg_compat/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q12/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q20.native_iceberg_compat/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q20/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q36.native_iceberg_compat/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q36/extended.txt

@@ -1,6 +1,6 @@
  TakeOrderedAndProject [COMET: ]
 +- Project
-   +-  Window [COMET: Window expressions are not supported]
+   +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       +- CometColumnarToRow
          +- CometSort
             +- CometExchange

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark3_5/q47.native_iceberg_compat/extended.txt

@@ -7,7 +7,7 @@
       :     :  +- Filter
       :     :     +- Window
       :     :        +- Filter
-      :     :           +-  Window [COMET: Window expressions are not supported]
+      :     :           +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       :     :              +- CometColumnarToRow
       :     :                 +- CometSort
       :     :                    +- CometExchange
@@ -39,7 +39,7 @@
       :     :                                            +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store
       :     +- BroadcastExchange
       :        +- Project
-      :           +-  Window [COMET: Window expressions are not supported]
+      :           +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
       :              +- CometColumnarToRow
       :                 +- CometSort
       :                    +- CometExchange
@@ -71,7 +71,7 @@
       :                                            +- CometScan [native_iceberg_compat] parquet spark_catalog.default.store
       +- BroadcastExchange
          +- Project
-            +-  Window [COMET: Window expressions are not supported]
+            +-  Window [COMET: WindowExec is not fully compatible with Spark (Native WindowExec has known correctness issues). To enable it anyway, set spark.comet.operator.WindowExec.allowIncompatible=true. For more information, refer to the Comet Compatibility Guide (https://datafusion.apache.org/comet/user-guide/compatibility.html).]
                +- CometColumnarToRow
                   +- CometSort
                      +- CometExchange