Fix PlanStabilitySuite

Author: manuzhang

dev/regenerate-golden-files.sh

@@ -22,7 +22,7 @@
 # Usage: ./dev/regenerate-golden-files.sh [--spark-version <version>]
 #
 # Options:
-#   --spark-version <version>  Only regenerate for specified Spark version (3.4, 3.5, or 4.0)
+#   --spark-version <version>  Only regenerate for specified Spark version (3.4, 3.5, 4.0 or 4.1)
 #                              If not specified, regenerates for all versions.
 #
 # Examples:
@@ -119,7 +119,7 @@ main() {
                 echo "Usage: $0 [--spark-version <version>]"
                 echo ""
                 echo "Options:"
-                echo "  --spark-version <version>  Only regenerate for specified Spark version (3.4, 3.5, or 4.0)"
+                echo "  --spark-version <version>  Only regenerate for specified Spark version (3.4, 3.5, 4.0 or 4.1)"
                 echo "                             If not specified, regenerates for all versions."
                 exit 0
                 ;;
@@ -133,9 +133,9 @@ main() {
 
     # Validate target version if specified
     if [ -n "$target_version" ]; then
-        if [[ ! "$target_version" =~ ^(3\.4|3\.5|4\.0)$ ]]; then
+        if [[ ! "$target_version" =~ ^(3\.4|3\.5|4\.0|4\.1)$ ]]; then
             echo "[ERROR] Invalid Spark version: $target_version"
-            echo "[ERROR] Supported versions: 3.4, 3.5, 4.0"
+            echo "[ERROR] Supported versions: 3.4, 3.5, 4.0, 4.1"
             exit 1
         fi
     fi
@@ -155,7 +155,7 @@ main() {
     if [ -n "$target_version" ]; then
         versions=("$target_version")
     else
-        versions=("3.4" "3.5" "4.0")
+        versions=("3.4" "3.5" "4.0", "4.1")
     fi
 
     # Install and regenerate for each version

spark/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_1/q1.native_datafusion/explain.txt

@@ -0,0 +1,291 @@
+== Physical Plan ==
+TakeOrderedAndProject (44)
++- * Project (43)
+   +- * BroadcastHashJoin Inner BuildRight (42)
+      :- * Project (36)
+      :  +- * BroadcastHashJoin Inner BuildRight (35)
+      :     :- * Project (29)
+      :     :  +- * BroadcastHashJoin Inner BuildRight (28)
+      :     :     :- * Filter (11)
+      :     :     :  +- * HashAggregate (10)
+      :     :     :     +- * CometColumnarToRow (9)
+      :     :     :        +- CometColumnarExchange (8)
+      :     :     :           +- * HashAggregate (7)
+      :     :     :              +- * Project (6)
+      :     :     :                 +- * BroadcastHashJoin Inner BuildRight (5)
+      :     :     :                    :- * Filter (3)
+      :     :     :                    :  +- * ColumnarToRow (2)
+      :     :     :                    :     +- Scan parquet spark_catalog.default.store_returns (1)
+      :     :     :                    +- ReusedExchange (4)
+      :     :     +- BroadcastExchange (27)
+      :     :        +- * Filter (26)
+      :     :           +- * HashAggregate (25)
+      :     :              +- * CometColumnarToRow (24)
+      :     :                 +- CometColumnarExchange (23)
+      :     :                    +- * HashAggregate (22)
+      :     :                       +- * HashAggregate (21)
+      :     :                          +- * CometColumnarToRow (20)
+      :     :                             +- CometColumnarExchange (19)
+      :     :                                +- * HashAggregate (18)
+      :     :                                   +- * Project (17)
+      :     :                                      +- * BroadcastHashJoin Inner BuildRight (16)
+      :     :                                         :- * Filter (14)
+      :     :                                         :  +- * ColumnarToRow (13)
+      :     :                                         :     +- Scan parquet spark_catalog.default.store_returns (12)
+      :     :                                         +- ReusedExchange (15)
+      :     +- BroadcastExchange (34)
+      :        +- * CometColumnarToRow (33)
+      :           +- CometProject (32)
+      :              +- CometFilter (31)
+      :                 +- CometNativeScan parquet spark_catalog.default.store (30)
+      +- BroadcastExchange (41)
+         +- * CometColumnarToRow (40)
+            +- CometProject (39)
+               +- CometFilter (38)
+                  +- CometNativeScan parquet spark_catalog.default.customer (37)
+
+
+(1) Scan parquet spark_catalog.default.store_returns
+Output [4]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3, sr_returned_date_sk#4]
+Batched: true
+Location: InMemoryFileIndex []
+PartitionFilters: [isnotnull(sr_returned_date_sk#4), dynamicpruningexpression(sr_returned_date_sk#4 IN dynamicpruning#5)]
+PushedFilters: [IsNotNull(sr_store_sk), IsNotNull(sr_customer_sk)]
+ReadSchema: struct<sr_customer_sk:int,sr_store_sk:int,sr_return_amt:decimal(7,2)>
+
+(2) ColumnarToRow [codegen id : 2]
+Input [4]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3, sr_returned_date_sk#4]
+
+(3) Filter [codegen id : 2]
+Input [4]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3, sr_returned_date_sk#4]
+Condition : (isnotnull(sr_store_sk#2) AND isnotnull(sr_customer_sk#1))
+
+(4) ReusedExchange [Reuses operator id: 49]
+Output [1]: [d_date_sk#6]
+
+(5) BroadcastHashJoin [codegen id : 2]
+Left keys [1]: [sr_returned_date_sk#4]
+Right keys [1]: [d_date_sk#6]
+Join type: Inner
+Join condition: None
+
+(6) Project [codegen id : 2]
+Output [3]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3]
+Input [5]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3, sr_returned_date_sk#4, d_date_sk#6]
+
+(7) HashAggregate [codegen id : 2]
+Input [3]: [sr_customer_sk#1, sr_store_sk#2, sr_return_amt#3]
+Keys [2]: [sr_customer_sk#1, sr_store_sk#2]
+Functions [1]: [partial_sum(UnscaledValue(sr_return_amt#3))]
+Aggregate Attributes [1]: [sum#7]
+Results [3]: [sr_customer_sk#1, sr_store_sk#2, sum#8]
+
+(8) CometColumnarExchange
+Input [3]: [sr_customer_sk#1, sr_store_sk#2, sum#8]
+Arguments: hashpartitioning(sr_customer_sk#1, sr_store_sk#2, 5), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=1]
+
+(9) CometColumnarToRow [codegen id : 9]
+Input [3]: [sr_customer_sk#1, sr_store_sk#2, sum#8]
+
+(10) HashAggregate [codegen id : 9]
+Input [3]: [sr_customer_sk#1, sr_store_sk#2, sum#8]
+Keys [2]: [sr_customer_sk#1, sr_store_sk#2]
+Functions [1]: [sum(UnscaledValue(sr_return_amt#3))]
+Aggregate Attributes [1]: [sum(UnscaledValue(sr_return_amt#3))#9]
+Results [3]: [sr_customer_sk#1 AS ctr_customer_sk#10, sr_store_sk#2 AS ctr_store_sk#11, MakeDecimal(sum(UnscaledValue(sr_return_amt#3))#9,17,2) AS ctr_total_return#12]
+
+(11) Filter [codegen id : 9]
+Input [3]: [ctr_customer_sk#10, ctr_store_sk#11, ctr_total_return#12]
+Condition : isnotnull(ctr_total_return#12)
+
+(12) Scan parquet spark_catalog.default.store_returns
+Output [4]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15, sr_returned_date_sk#16]
+Batched: true
+Location: InMemoryFileIndex []
+PartitionFilters: [isnotnull(sr_returned_date_sk#16), dynamicpruningexpression(sr_returned_date_sk#16 IN dynamicpruning#5)]
+PushedFilters: [IsNotNull(sr_store_sk)]
+ReadSchema: struct<sr_customer_sk:int,sr_store_sk:int,sr_return_amt:decimal(7,2)>
+
+(13) ColumnarToRow [codegen id : 4]
+Input [4]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15, sr_returned_date_sk#16]
+
+(14) Filter [codegen id : 4]
+Input [4]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15, sr_returned_date_sk#16]
+Condition : isnotnull(sr_store_sk#14)
+
+(15) ReusedExchange [Reuses operator id: 49]
+Output [1]: [d_date_sk#17]
+
+(16) BroadcastHashJoin [codegen id : 4]
+Left keys [1]: [sr_returned_date_sk#16]
+Right keys [1]: [d_date_sk#17]
+Join type: Inner
+Join condition: None
+
+(17) Project [codegen id : 4]
+Output [3]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15]
+Input [5]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15, sr_returned_date_sk#16, d_date_sk#17]
+
+(18) HashAggregate [codegen id : 4]
+Input [3]: [sr_customer_sk#13, sr_store_sk#14, sr_return_amt#15]
+Keys [2]: [sr_customer_sk#13, sr_store_sk#14]
+Functions [1]: [partial_sum(UnscaledValue(sr_return_amt#15))]
+Aggregate Attributes [1]: [sum#18]
+Results [3]: [sr_customer_sk#13, sr_store_sk#14, sum#19]
+
+(19) CometColumnarExchange
+Input [3]: [sr_customer_sk#13, sr_store_sk#14, sum#19]
+Arguments: hashpartitioning(sr_customer_sk#13, sr_store_sk#14, 5), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=2]
+
+(20) CometColumnarToRow [codegen id : 5]
+Input [3]: [sr_customer_sk#13, sr_store_sk#14, sum#19]
+
+(21) HashAggregate [codegen id : 5]
+Input [3]: [sr_customer_sk#13, sr_store_sk#14, sum#19]
+Keys [2]: [sr_customer_sk#13, sr_store_sk#14]
+Functions [1]: [sum(UnscaledValue(sr_return_amt#15))]
+Aggregate Attributes [1]: [sum(UnscaledValue(sr_return_amt#15))#9]
+Results [2]: [sr_store_sk#14 AS ctr_store_sk#20, MakeDecimal(sum(UnscaledValue(sr_return_amt#15))#9,17,2) AS ctr_total_return#21]
+
+(22) HashAggregate [codegen id : 5]
+Input [2]: [ctr_store_sk#20, ctr_total_return#21]
+Keys [1]: [ctr_store_sk#20]
+Functions [1]: [partial_avg(ctr_total_return#21)]
+Aggregate Attributes [2]: [sum#22, count#23]
+Results [3]: [ctr_store_sk#20, sum#24, count#25]
+
+(23) CometColumnarExchange
+Input [3]: [ctr_store_sk#20, sum#24, count#25]
+Arguments: hashpartitioning(ctr_store_sk#20, 5), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=3]
+
+(24) CometColumnarToRow [codegen id : 6]
+Input [3]: [ctr_store_sk#20, sum#24, count#25]
+
+(25) HashAggregate [codegen id : 6]
+Input [3]: [ctr_store_sk#20, sum#24, count#25]
+Keys [1]: [ctr_store_sk#20]
+Functions [1]: [avg(ctr_total_return#21)]
+Aggregate Attributes [1]: [avg(ctr_total_return#21)#26]
+Results [2]: [(avg(ctr_total_return#21)#26 * 1.2) AS (avg(ctr_total_return) * 1.2)#27, ctr_store_sk#20]
+
+(26) Filter [codegen id : 6]
+Input [2]: [(avg(ctr_total_return) * 1.2)#27, ctr_store_sk#20]
+Condition : isnotnull((avg(ctr_total_return) * 1.2)#27)
+
+(27) BroadcastExchange
+Input [2]: [(avg(ctr_total_return) * 1.2)#27, ctr_store_sk#20]
+Arguments: HashedRelationBroadcastMode(List(cast(input[1, int, true] as bigint)),false), [plan_id=4]
+
+(28) BroadcastHashJoin [codegen id : 9]
+Left keys [1]: [ctr_store_sk#11]
+Right keys [1]: [ctr_store_sk#20]
+Join type: Inner
+Join condition: (cast(ctr_total_return#12 as decimal(24,7)) > (avg(ctr_total_return) * 1.2)#27)
+
+(29) Project [codegen id : 9]
+Output [2]: [ctr_customer_sk#10, ctr_store_sk#11]
+Input [5]: [ctr_customer_sk#10, ctr_store_sk#11, ctr_total_return#12, (avg(ctr_total_return) * 1.2)#27, ctr_store_sk#20]
+
+(30) CometNativeScan parquet spark_catalog.default.store
+Output [2]: [s_store_sk#28, s_state#29]
+Batched: true
+Location [not included in comparison]/{warehouse_dir}/store]
+PushedFilters: [IsNotNull(s_state), IsNotNull(s_store_sk)]
+ReadSchema: struct<s_store_sk:int,s_state:string>
+
+(31) CometFilter
+Input [2]: [s_store_sk#28, s_state#29]
+Condition : ((isnotnull(s_state#29) AND (static_invoke(CharVarcharCodegenUtils.readSidePadding(s_state#29, 2)) = TN)) AND isnotnull(s_store_sk#28))
+
+(32) CometProject
+Input [2]: [s_store_sk#28, s_state#29]
+Arguments: [s_store_sk#28], [s_store_sk#28]
+
+(33) CometColumnarToRow [codegen id : 7]
+Input [1]: [s_store_sk#28]
+
+(34) BroadcastExchange
+Input [1]: [s_store_sk#28]
+Arguments: HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [plan_id=5]
+
+(35) BroadcastHashJoin [codegen id : 9]
+Left keys [1]: [ctr_store_sk#11]
+Right keys [1]: [s_store_sk#28]
+Join type: Inner
+Join condition: None
+
+(36) Project [codegen id : 9]
+Output [1]: [ctr_customer_sk#10]
+Input [3]: [ctr_customer_sk#10, ctr_store_sk#11, s_store_sk#28]
+
+(37) CometNativeScan parquet spark_catalog.default.customer
+Output [2]: [c_customer_sk#30, c_customer_id#31]
+Batched: true
+Location [not included in comparison]/{warehouse_dir}/customer]
+PushedFilters: [IsNotNull(c_customer_sk)]
+ReadSchema: struct<c_customer_sk:int,c_customer_id:string>
+
+(38) CometFilter
+Input [2]: [c_customer_sk#30, c_customer_id#31]
+Condition : isnotnull(c_customer_sk#30)
+
+(39) CometProject
+Input [2]: [c_customer_sk#30, c_customer_id#31]
+Arguments: [c_customer_sk#30, c_customer_id#32], [c_customer_sk#30, static_invoke(CharVarcharCodegenUtils.readSidePadding(c_customer_id#31, 16)) AS c_customer_id#32]
+
+(40) CometColumnarToRow [codegen id : 8]
+Input [2]: [c_customer_sk#30, c_customer_id#32]
+
+(41) BroadcastExchange
+Input [2]: [c_customer_sk#30, c_customer_id#32]
+Arguments: HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [plan_id=6]
+
+(42) BroadcastHashJoin [codegen id : 9]
+Left keys [1]: [ctr_customer_sk#10]
+Right keys [1]: [c_customer_sk#30]
+Join type: Inner
+Join condition: None
+
+(43) Project [codegen id : 9]
+Output [1]: [c_customer_id#32]
+Input [3]: [ctr_customer_sk#10, c_customer_sk#30, c_customer_id#32]
+
+(44) TakeOrderedAndProject
+Input [1]: [c_customer_id#32]
+Arguments: 100, [c_customer_id#32 ASC NULLS FIRST], [c_customer_id#32]
+
+===== Subqueries =====
+
+Subquery:1 Hosting operator id = 1 Hosting Expression = sr_returned_date_sk#4 IN dynamicpruning#5
+BroadcastExchange (49)
++- * CometColumnarToRow (48)
+   +- CometProject (47)
+      +- CometFilter (46)
+         +- CometNativeScan parquet spark_catalog.default.date_dim (45)
+
+
+(45) CometNativeScan parquet spark_catalog.default.date_dim
+Output [2]: [d_date_sk#6, d_year#33]
+Batched: true
+Location [not included in comparison]/{warehouse_dir}/date_dim]
+PushedFilters: [IsNotNull(d_year), EqualTo(d_year,2000), IsNotNull(d_date_sk)]
+ReadSchema: struct<d_date_sk:int,d_year:int>
+
+(46) CometFilter
+Input [2]: [d_date_sk#6, d_year#33]
+Condition : ((isnotnull(d_year#33) AND (d_year#33 = 2000)) AND isnotnull(d_date_sk#6))
+
+(47) CometProject
+Input [2]: [d_date_sk#6, d_year#33]
+Arguments: [d_date_sk#6], [d_date_sk#6]
+
+(48) CometColumnarToRow [codegen id : 1]
+Input [1]: [d_date_sk#6]
+
+(49) BroadcastExchange
+Input [1]: [d_date_sk#6]
+Arguments: HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [plan_id=7]
+
+Subquery:2 Hosting operator id = 12 Hosting Expression = sr_returned_date_sk#16 IN dynamicpruning#5
+
+

spark/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4-spark4_1/q1.native_datafusion/extended.txt

@@ -0,0 +1,61 @@
+TakeOrderedAndProject
++- Project
+   +- BroadcastHashJoin
+      :- Project
+      :  +- BroadcastHashJoin
+      :     :- Project
+      :     :  +- BroadcastHashJoin
+      :     :     :- Filter
+      :     :     :  +- HashAggregate
+      :     :     :     +- CometColumnarToRow
+      :     :     :        +- CometColumnarExchange
+      :     :     :           +- HashAggregate
+      :     :     :              +- Project
+      :     :     :                 +- BroadcastHashJoin
+      :     :     :                    :- Filter
+      :     :     :                    :  +- ColumnarToRow
+      :     :     :                    :     +-  Scan parquet spark_catalog.default.store_returns [COMET: Native DataFusion scan does not support subqueries/dynamic pruning]
+      :     :     :                    :           +- SubqueryBroadcast
+      :     :     :                    :              +- BroadcastExchange
+      :     :     :                    :                 +- CometColumnarToRow
+      :     :     :                    :                    +- CometProject
+      :     :     :                    :                       +- CometFilter
+      :     :     :                    :                          +- CometNativeScan parquet spark_catalog.default.date_dim
+      :     :     :                    +- BroadcastExchange
+      :     :     :                       +- CometColumnarToRow
+      :     :     :                          +- CometProject
+      :     :     :                             +- CometFilter
+      :     :     :                                +- CometNativeScan parquet spark_catalog.default.date_dim
+      :     :     +- BroadcastExchange
+      :     :        +- Filter
+      :     :           +- HashAggregate
+      :     :              +- CometColumnarToRow
+      :     :                 +- CometColumnarExchange
+      :     :                    +- HashAggregate
+      :     :                       +- HashAggregate
+      :     :                          +- CometColumnarToRow
+      :     :                             +- CometColumnarExchange
+      :     :                                +- HashAggregate
+      :     :                                   +- Project
+      :     :                                      +- BroadcastHashJoin
+      :     :                                         :- Filter
+      :     :                                         :  +- ColumnarToRow
+      :     :                                         :     +-  Scan parquet spark_catalog.default.store_returns [COMET: Native DataFusion scan does not support subqueries/dynamic pruning]
+      :     :                                         :           +- ReusedSubquery
+      :     :                                         +- BroadcastExchange
+      :     :                                            +- CometColumnarToRow
+      :     :                                               +- CometProject
+      :     :                                                  +- CometFilter
+      :     :                                                     +- CometNativeScan parquet spark_catalog.default.date_dim
+      :     +- BroadcastExchange
+      :        +- CometColumnarToRow
+      :           +- CometProject
+      :              +- CometFilter
+      :                 +- CometNativeScan parquet spark_catalog.default.store
+      +- BroadcastExchange
+         +- CometColumnarToRow
+            +- CometProject
+               +- CometFilter
+                  +- CometNativeScan parquet spark_catalog.default.customer
+
+Comet accelerated 18 out of 49 eligible operators (36%). Final plan contains 10 transitions between Spark and Comet.