Expression explain tags leak between unrelated plans via the shared Literal.TrueLiteral singleton

[andygrove]

Describe the bug

CometExecRule.rollUpInfoMessages lifts EXTENSION_INFO and CODEGEN_DISPATCH_EXPRS tags off every expression node of an operator:

https://github.com/apache/datafusion-comet/blob/main/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala#L751-L767

val allExprs = op.expressions.flatMap(_.collect { case e: Expression => e })

val infos =
  op.getTagValue(CometExplainInfo.EXTENSION_INFO).getOrElse(Set.empty[String]) ++
    allExprs.flatMap(_.getTagValue(CometExplainInfo.EXTENSION_INFO)).flatten
infos.foreach(msg => withInfo(exec, msg))

Some of the nodes in allExprs are process-wide singletons, most importantly Literal.TrueLiteral. Catalyst's TreeNode.copyTagsFrom copies a rewritten node's tags onto its replacement whenever the replacement has no tags of its own (SPARK-32753). So the first time Comet tags an expression that Catalyst later rewrites into Literal.TrueLiteral, the tag is welded onto that singleton for the lifetime of the JVM.

Every plan planned afterwards in the same JVM that contains TrueLiteral then inherits the tag. dynamicpruningexpression(true), which CleanupDynamicPruningFilters puts on partitioned scans, is the common carrier. The result is a [COMET-INFO: ...] message attached to an operator that has nothing to do with the message, in an unrelated query, in an unrelated session.

The write side already guards against structural nodes (QueryPlanSerde.isStructuralExpr excludes Attribute, BoundReference, Literal, Alias), but the read side above does not, so a tag that arrived on a literal by copying is read back as if the operator itself had produced it.

Steps to reproduce

Plant a tag on the singleton to stand in for whatever earlier query contaminated it, then plan an unrelated query whose scan carries a cleaned-up DPP filter:

class TagLeakSuite extends CometTestBase {
  test("EXTENSION_INFO on the TrueLiteral singleton leaks into unrelated plans") {
    Literal.TrueLiteral.setTagValue(CometExplainInfo.EXTENSION_INFO, Set("PLANTED_LEAK"))

    withSQLConf(
      CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key ->
        CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE,
      "spark.sql.optimizer.dynamicPartitionPruning.enabled" -> "true",
      "spark.sql.autoBroadcastJoinThreshold" -> "-1",
      "spark.sql.adaptive.enabled" -> "false") {
      withTable("fact", "dim") {
        sql("create table fact (v int, p int) using parquet partitioned by (p)")
        sql("insert into fact values (1, 1), (2, 2)")
        sql("create table dim (k int, s string) using parquet")
        sql("insert into dim values (1, 'a'), (2, 'b')")
        val plan = sql("select * from fact join dim on fact.p = dim.k where dim.s = 'a'")
          .queryExecution.executedPlan
        println(new ExtendedExplainInfo().generateExtendedInfo(plan))
      }
    }
  }
}

Output, on the fact scan, which never saw the tagged expression:

CometColumnarToRow
+- CometSortMergeJoin
   :- CometSort
   :  +- CometExchange
   :     +-  CometNativeScan parquet spark_catalog.default.fact [COMET-INFO: PLANTED_LEAK]
   +- CometSort
      +- CometExchange
         +- CometFilter
            +- CometNativeScan parquet spark_catalog.default.dim

The contamination step is not hypothetical. It was found while investigating a CI-only plan diff on #5201, which adds a NATIVE_EXPRS tag read through this same code path. CometJoinSuite's BroadcastNestedLoopJoin ... with inequality tests tag their GreaterThanOrEqual conditions; a later test in that suite plans a join whose condition is TrueLiteral, and from that point on every TPC-DS plan in the JVM reported a greaterthanorequal that the query does not contain:

op=CometScanExec exprs=ss_item_sk | ss_ext_sales_price | ss_sold_date_sk
    | isnotnull(ss_sold_date_sk) | dynamicpruningexpression(true) | isnotnull(ss_item_sk)
  carrier=Literal: true

That is why CometTPCDSV1_4_PlanStabilitySuite produces different output when it runs alone (as dev/regenerate-golden-files.sh runs it) than when it runs inside CI's 40-suite [exec] batch, where CometJoinSuite has already polluted the singleton.

Expected behavior

A [COMET-INFO: ...] message should appear only on the operator whose own expressions produced it, and explain output for a query should not depend on which queries were planned earlier in the same JVM.

Additional context

Suggested fix: withInfo, withCodegenDispatchExpr (and withNativeExpr on #5201) only ever tag a node with a value derived from that node. When reading, keep only values the carrier could have produced, e.g. filter to exprDisplayName(e) for the name-valued tags, or skip structural nodes on the read side to mirror isStructuralExpr on the write side. Filtering by carrier is the stronger of the two: it holds for any shared node, not just literals.

Worth checking whether FALLBACK_REASONS is exposed to the same copying, since it is also set on expressions.

Impact today is confined to explain output; no query results are affected. It does mean [COMET-INFO: ...] segments in an EXPLAIN can be misleading, and it makes any golden file that captures extended explain sensitive to suite ordering.