fix: Unknown operator id when explain with formatted mode (#410)

* Fix unkown id for explain with formatted mode

* update plan stability results

spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala

@@ -142,7 +142,10 @@ case class CometScanExec(
   override lazy val metadata: Map[String, String] =
     if (wrapped == null) Map.empty else wrapped.metadata
 
-  override def verboseStringWithOperatorId(): String = wrapped.verboseStringWithOperatorId()
+  override def verboseStringWithOperatorId(): String = {
+    getTagValue(QueryPlan.OP_ID_TAG).foreach(id => wrapped.setTagValue(QueryPlan.OP_ID_TAG, id))
+    wrapped.verboseStringWithOperatorId()
+  }
 
   lazy val inputRDD: RDD[InternalRow] = {
     val options = relation.options +

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

@@ -41,7 +41,7 @@ TakeOrderedAndProject (40)
                +- CometScan parquet spark_catalog.default.customer (34)
 
 
-(unknown) Scan parquet spark_catalog.default.store_returns
+(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 []
@@ -91,7 +91,7 @@ Results [3]: [sr_customer_sk#1 AS ctr_customer_sk#10, sr_store_sk#2 AS ctr_store
 Input [3]: [ctr_customer_sk#10, ctr_store_sk#11, ctr_total_return#12]
 Condition : isnotnull(ctr_total_return#12)
 
-(unknown) Scan parquet spark_catalog.default.store_returns
+(11) 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 []
@@ -173,7 +173,7 @@ Join condition: (cast(ctr_total_return#12 as decimal(24,7)) > (avg(ctr_total_ret
 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)#21, ctr_store_sk#11#22]
 
-(unknown) Scan parquet spark_catalog.default.store
+(27) Scan parquet spark_catalog.default.store
 Output [2]: [s_store_sk#23, s_state#24]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/store]
@@ -205,7 +205,7 @@ Join condition: None
 Output [1]: [ctr_customer_sk#10]
 Input [3]: [ctr_customer_sk#10, ctr_store_sk#11, s_store_sk#23]
 
-(unknown) Scan parquet spark_catalog.default.customer
+(34) Scan parquet spark_catalog.default.customer
 Output [2]: [c_customer_sk#25, c_customer_id#26]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -247,7 +247,7 @@ BroadcastExchange (45)
          +- CometScan parquet spark_catalog.default.date_dim (41)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(41) Scan parquet spark_catalog.default.date_dim
 Output [2]: [d_date_sk#6, d_year#27]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q10/explain.txt

@@ -44,7 +44,7 @@ TakeOrderedAndProject (43)
                         +- CometScan parquet spark_catalog.default.customer_demographics (34)
 
 
-(unknown) Scan parquet spark_catalog.default.customer
+(1) Scan parquet spark_catalog.default.customer
 Output [3]: [c_customer_sk#3, c_current_cdemo_sk#4, c_current_addr_sk#5]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -58,7 +58,7 @@ Condition : (isnotnull(c_current_addr_sk#5) AND isnotnull(c_current_cdemo_sk#4))
 (3) ColumnarToRow [codegen id : 9]
 Input [3]: [c_customer_sk#3, c_current_cdemo_sk#4, c_current_addr_sk#5]
 
-(unknown) Scan parquet spark_catalog.default.store_sales
+(4) Scan parquet spark_catalog.default.store_sales
 Output [2]: [ss_customer_sk#6, ss_sold_date_sk#7]
 Batched: true
 Location: InMemoryFileIndex []
@@ -91,7 +91,7 @@ Right keys [1]: [ss_customer_sk#6]
 Join type: LeftSemi
 Join condition: None
 
-(unknown) Scan parquet spark_catalog.default.web_sales
+(11) Scan parquet spark_catalog.default.web_sales
 Output [2]: [ws_bill_customer_sk#10, ws_sold_date_sk#11]
 Batched: true
 Location: InMemoryFileIndex []
@@ -124,7 +124,7 @@ Right keys [1]: [ws_bill_customer_sk#10]
 Join type: ExistenceJoin(exists#2)
 Join condition: None
 
-(unknown) Scan parquet spark_catalog.default.catalog_sales
+(18) Scan parquet spark_catalog.default.catalog_sales
 Output [2]: [cs_ship_customer_sk#14, cs_sold_date_sk#15]
 Batched: true
 Location: InMemoryFileIndex []
@@ -165,7 +165,7 @@ Condition : (exists#2 OR exists#1)
 Output [2]: [c_current_cdemo_sk#4, c_current_addr_sk#5]
 Input [5]: [c_customer_sk#3, c_current_cdemo_sk#4, c_current_addr_sk#5, exists#2, exists#1]
 
-(unknown) Scan parquet spark_catalog.default.customer_address
+(27) Scan parquet spark_catalog.default.customer_address
 Output [2]: [ca_address_sk#18, ca_county#19]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer_address]
@@ -197,7 +197,7 @@ Join condition: None
 Output [1]: [c_current_cdemo_sk#4]
 Input [3]: [c_current_cdemo_sk#4, c_current_addr_sk#5, ca_address_sk#18]
 
-(unknown) Scan parquet spark_catalog.default.customer_demographics
+(34) Scan parquet spark_catalog.default.customer_demographics
 Output [9]: [cd_demo_sk#20, cd_gender#21, cd_marital_status#22, cd_education_status#23, cd_purchase_estimate#24, cd_credit_rating#25, cd_dep_count#26, cd_dep_employed_count#27, cd_dep_college_count#28]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer_demographics]
@@ -257,7 +257,7 @@ BroadcastExchange (48)
          +- CometScan parquet spark_catalog.default.date_dim (44)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(44) Scan parquet spark_catalog.default.date_dim
 Output [3]: [d_date_sk#9, d_year#38, d_moy#39]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q11/explain.txt

@@ -73,7 +73,7 @@ TakeOrderedAndProject (72)
                         +- ReusedExchange (63)
 
 
-(unknown) Scan parquet spark_catalog.default.customer
+(1) Scan parquet spark_catalog.default.customer
 Output [8]: [c_customer_sk#1, c_customer_id#2, c_first_name#3, c_last_name#4, c_preferred_cust_flag#5, c_birth_country#6, c_login#7, c_email_address#8]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -87,7 +87,7 @@ Condition : (isnotnull(c_customer_sk#1) AND isnotnull(c_customer_id#2))
 (3) ColumnarToRow [codegen id : 3]
 Input [8]: [c_customer_sk#1, c_customer_id#2, c_first_name#3, c_last_name#4, c_preferred_cust_flag#5, c_birth_country#6, c_login#7, c_email_address#8]
 
-(unknown) Scan parquet spark_catalog.default.store_sales
+(4) Scan parquet spark_catalog.default.store_sales
 Output [4]: [ss_customer_sk#9, ss_ext_discount_amt#10, ss_ext_list_price#11, ss_sold_date_sk#12]
 Batched: true
 Location: InMemoryFileIndex []
@@ -151,7 +151,7 @@ Results [2]: [c_customer_id#2 AS customer_id#19, MakeDecimal(sum(UnscaledValue((
 Input [2]: [customer_id#19, year_total#20]
 Condition : (isnotnull(year_total#20) AND (year_total#20 > 0.00))
 
-(unknown) Scan parquet spark_catalog.default.customer
+(17) Scan parquet spark_catalog.default.customer
 Output [8]: [c_customer_sk#21, c_customer_id#22, c_first_name#23, c_last_name#24, c_preferred_cust_flag#25, c_birth_country#26, c_login#27, c_email_address#28]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -165,7 +165,7 @@ Condition : (isnotnull(c_customer_sk#21) AND isnotnull(c_customer_id#22))
 (19) ColumnarToRow [codegen id : 6]
 Input [8]: [c_customer_sk#21, c_customer_id#22, c_first_name#23, c_last_name#24, c_preferred_cust_flag#25, c_birth_country#26, c_login#27, c_email_address#28]
 
-(unknown) Scan parquet spark_catalog.default.store_sales
+(20) Scan parquet spark_catalog.default.store_sales
 Output [4]: [ss_customer_sk#29, ss_ext_discount_amt#30, ss_ext_list_price#31, ss_sold_date_sk#32]
 Batched: true
 Location: InMemoryFileIndex []
@@ -239,7 +239,7 @@ Join condition: None
 Output [4]: [customer_id#19, year_total#20, customer_preferred_cust_flag#39, year_total#40]
 Input [5]: [customer_id#19, year_total#20, customer_id#38, customer_preferred_cust_flag#39, year_total#40]
 
-(unknown) Scan parquet spark_catalog.default.customer
+(35) Scan parquet spark_catalog.default.customer
 Output [8]: [c_customer_sk#41, c_customer_id#42, c_first_name#43, c_last_name#44, c_preferred_cust_flag#45, c_birth_country#46, c_login#47, c_email_address#48]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -253,7 +253,7 @@ Condition : (isnotnull(c_customer_sk#41) AND isnotnull(c_customer_id#42))
 (37) ColumnarToRow [codegen id : 10]
 Input [8]: [c_customer_sk#41, c_customer_id#42, c_first_name#43, c_last_name#44, c_preferred_cust_flag#45, c_birth_country#46, c_login#47, c_email_address#48]
 
-(unknown) Scan parquet spark_catalog.default.web_sales
+(38) Scan parquet spark_catalog.default.web_sales
 Output [4]: [ws_bill_customer_sk#49, ws_ext_discount_amt#50, ws_ext_list_price#51, ws_sold_date_sk#52]
 Batched: true
 Location: InMemoryFileIndex []
@@ -331,7 +331,7 @@ Join condition: None
 Output [5]: [customer_id#19, year_total#20, customer_preferred_cust_flag#39, year_total#40, year_total#60]
 Input [6]: [customer_id#19, year_total#20, customer_preferred_cust_flag#39, year_total#40, customer_id#59, year_total#60]
 
-(unknown) Scan parquet spark_catalog.default.customer
+(54) Scan parquet spark_catalog.default.customer
 Output [8]: [c_customer_sk#61, c_customer_id#62, c_first_name#63, c_last_name#64, c_preferred_cust_flag#65, c_birth_country#66, c_login#67, c_email_address#68]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer]
@@ -345,7 +345,7 @@ Condition : (isnotnull(c_customer_sk#61) AND isnotnull(c_customer_id#62))
 (56) ColumnarToRow [codegen id : 14]
 Input [8]: [c_customer_sk#61, c_customer_id#62, c_first_name#63, c_last_name#64, c_preferred_cust_flag#65, c_birth_country#66, c_login#67, c_email_address#68]
 
-(unknown) Scan parquet spark_catalog.default.web_sales
+(57) Scan parquet spark_catalog.default.web_sales
 Output [4]: [ws_bill_customer_sk#69, ws_ext_discount_amt#70, ws_ext_list_price#71, ws_sold_date_sk#72]
 Batched: true
 Location: InMemoryFileIndex []
@@ -432,7 +432,7 @@ BroadcastExchange (76)
       +- CometScan parquet spark_catalog.default.date_dim (73)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(73) Scan parquet spark_catalog.default.date_dim
 Output [2]: [d_date_sk#14, d_year#15]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]
@@ -457,7 +457,7 @@ BroadcastExchange (80)
       +- CometScan parquet spark_catalog.default.date_dim (77)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(77) Scan parquet spark_catalog.default.date_dim
 Output [2]: [d_date_sk#34, d_year#35]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]

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

@@ -21,7 +21,7 @@ TakeOrderedAndProject (20)
                            +- ReusedExchange (10)
 
 
-(unknown) Scan parquet spark_catalog.default.web_sales
+(1) Scan parquet spark_catalog.default.web_sales
 Output [3]: [ws_item_sk#1, ws_ext_sales_price#2, ws_sold_date_sk#3]
 Batched: true
 Location: InMemoryFileIndex []
@@ -36,7 +36,7 @@ Condition : isnotnull(ws_item_sk#1)
 (3) ColumnarToRow [codegen id : 3]
 Input [3]: [ws_item_sk#1, ws_ext_sales_price#2, ws_sold_date_sk#3]
 
-(unknown) Scan parquet spark_catalog.default.item
+(4) Scan parquet spark_catalog.default.item
 Output [6]: [i_item_sk#5, i_item_id#6, i_item_desc#7, i_current_price#8, i_class#9, i_category#10]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/item]
@@ -125,7 +125,7 @@ BroadcastExchange (25)
          +- CometScan parquet spark_catalog.default.date_dim (21)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(21) Scan parquet spark_catalog.default.date_dim
 Output [2]: [d_date_sk#11, d_date#19]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]

spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q13/explain.txt

@@ -35,7 +35,7 @@
                      +- CometScan parquet spark_catalog.default.household_demographics (26)
 
 
-(unknown) Scan parquet spark_catalog.default.store_sales
+(1) Scan parquet spark_catalog.default.store_sales
 Output [10]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_addr_sk#3, ss_store_sk#4, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, ss_net_profit#9, ss_sold_date_sk#10]
 Batched: true
 Location: InMemoryFileIndex []
@@ -50,7 +50,7 @@ Condition : (((((isnotnull(ss_store_sk#4) AND isnotnull(ss_addr_sk#3)) AND isnot
 (3) ColumnarToRow [codegen id : 6]
 Input [10]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_addr_sk#3, ss_store_sk#4, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, ss_net_profit#9, ss_sold_date_sk#10]
 
-(unknown) Scan parquet spark_catalog.default.store
+(4) Scan parquet spark_catalog.default.store
 Output [1]: [s_store_sk#12]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/store]
@@ -78,7 +78,7 @@ Join condition: None
 Output [9]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_addr_sk#3, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, ss_net_profit#9, ss_sold_date_sk#10]
 Input [11]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_addr_sk#3, ss_store_sk#4, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, ss_net_profit#9, ss_sold_date_sk#10, s_store_sk#12]
 
-(unknown) Scan parquet spark_catalog.default.customer_address
+(10) Scan parquet spark_catalog.default.customer_address
 Output [3]: [ca_address_sk#13, ca_state#14, ca_country#15]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer_address]
@@ -123,7 +123,7 @@ Join condition: None
 Output [6]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8]
 Input [8]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, ss_sold_date_sk#10, d_date_sk#16]
 
-(unknown) Scan parquet spark_catalog.default.customer_demographics
+(20) Scan parquet spark_catalog.default.customer_demographics
 Output [3]: [cd_demo_sk#17, cd_marital_status#18, cd_education_status#19]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/customer_demographics]
@@ -151,7 +151,7 @@ Join condition: ((((((cd_marital_status#18 = M) AND (cd_education_status#19 = Ad
 Output [7]: [ss_hdemo_sk#2, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, cd_marital_status#18, cd_education_status#19]
 Input [9]: [ss_cdemo_sk#1, ss_hdemo_sk#2, ss_quantity#5, ss_sales_price#6, ss_ext_sales_price#7, ss_ext_wholesale_cost#8, cd_demo_sk#17, cd_marital_status#18, cd_education_status#19]
 
-(unknown) Scan parquet spark_catalog.default.household_demographics
+(26) Scan parquet spark_catalog.default.household_demographics
 Output [2]: [hd_demo_sk#20, hd_dep_count#21]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/household_demographics]
@@ -207,7 +207,7 @@ BroadcastExchange (39)
          +- CometScan parquet spark_catalog.default.date_dim (35)
 
 
-(unknown) Scan parquet spark_catalog.default.date_dim
+(35) Scan parquet spark_catalog.default.date_dim
 Output [2]: [d_date_sk#16, d_year#44]
 Batched: true
 Location [not included in comparison]/{warehouse_dir}/date_dim]