fix: adjust CometNativeScan's doCanonicalize and hashCode for AQE, use DataSourceScanExec trait (#1578)

## Which issue does this PR close?

Addresses another failure in #1441.

## Rationale for this change

`CometExecSuite.explain native plan` fails with `native_datafusion` experimental scan. It's an interesting query that does a self-join of two columns from the same table. The root case is that when AQE is enabled, it would reuse the shuffle output from one scan as the output of the other scan:

```
+- == Initial Plan ==
   CometProject [_1#6], [_1#6]
   +- CometSortMergeJoin [_1#6], [_2#11], Inner
      :- CometSort [_1#6], [_1#6 ASC NULLS FIRST]
      :  +- CometExchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=304]
      :     +- CometFilter [_1#6], isnotnull(_1#6)
      :        +- CometNativeScan:  [_1#6]
      +- CometSort [_2#11], [_2#11 ASC NULLS FIRST]
         +- CometExchange hashpartitioning(_2#11, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=308]
            +- CometFilter [_2#11], isnotnull(_2#11)
               +- CometNativeScan:  [_2#11]
```
AQE incorrectly adds a `ReusedExchange` on the left side with the same `plan_id` as the right side of the join.
```
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(1) CometColumnarToRow
   +- CometProject [_1#6], [_1#6]
      +- CometBroadcastHashJoin [_1#6], [_2#11], Inner, BuildRight
         :- AQEShuffleRead coalesced
         :  +- ShuffleQueryStage 0
         :     +- CometExchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=304]
         :        +- CometFilter [_1#6], isnotnull(_1#6)
         :           +- CometNativeScan:  [_1#6]
         +- BroadcastQueryStage 2
            +- CometBroadcastExchange [_2#11]
               +- AQEShuffleRead local
                  +- ShuffleQueryStage 1
                     +- ReusedExchange [_2#11], CometExchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=304]
```

The reason is that `hashCode()` for `CometNativeScan` is only defined as the output of the node, so the `TrieMap` used in AQE (which hashes the `SparkPlan`) resulted in the stages having the same hash value (after canonicalization), making AQE think that one stage could be reused for the other.

## What changes are included in this PR?

- Expand `hashCode` to include the original `FileSourceScanExec` and `serializedPlanOpt` which has better info about the node. I'd like to understand if this is hashing too much information, and may make stages that could be reused appear to distinct, but need to dig into AQE behavior more.
- Expand `equals` to check more than just the plan output.
- Expand `doCanonicalize` based on behavior seen in `CometScan` node. Similar to above: I'd like to understand if this is canonicalizing the right information, but need to dig into AQE behavior more.
- `CometNativeScan` now uses the `DataSourceScanExec` trait. The benefit here is that we get more detailed information in the Spark plan. For example, explain before (note the `CometNativeScan`):
```
CometProject [_1#6], [_1#6]
+- CometSortMergeJoin [_1#6], [_2#11], Inner
   :- CometSort [_1#6], [_1#6 ASC NULLS FIRST]
   :  +- CometExchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=304]
   :     +- CometFilter [_1#6], isnotnull(_1#6)
   :        +- CometNativeScan:  [_1#6]
   +- CometSort [_2#11], [_2#11 ASC NULLS FIRST]
      +- CometExchange hashpartitioning(_2#11, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=308]
         +- CometFilter [_2#11], isnotnull(_2#11)
            +- CometNativeScan:  [_2#11]
```
and explain now (note the `CometNativeScan`):
```
CometProject [_1#6], [_1#6]
+- CometSortMergeJoin [_1#6], [_2#11], Inner
   :- CometSort [_1#6], [_1#6 ASC NULLS FIRST]
   :  +- CometExchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=91]
   :     +- CometFilter [_1#6], isnotnull(_1#6)
   :        +- CometNativeScan parquet [_1#6] Batched: true, DataFilters: [isnotnull(_1#6)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/12/4pf3d5zn72n7q2_0ks3bkh7c0000gn/T/spark-8f..., PartitionFilters: [], PushedFilters: [IsNotNull(_1)], ReadSchema: struct<_1:int>
   +- CometSort [_2#11], [_2#11 ASC NULLS FIRST]
      +- CometExchange hashpartitioning(_2#11, 10), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=95]
         +- CometFilter [_2#11], isnotnull(_2#11)
            +- CometNativeScan parquet [_2#11] Batched: true, DataFilters: [isnotnull(_2#11)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/12/4pf3d5zn72n7q2_0ks3bkh7c0000gn/T/spark-8f..., PartitionFilters: [], PushedFilters: [IsNotNull(_2)], ReadSchema: struct<_2:int>
```
This better represents a corresponding Spark plan with its `FileScan` node:
```
Project [_1#6]
+- SortMergeJoin [_1#6], [_2#11], Inner
   :- Sort [_1#6 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(_1#6, 10), ENSURE_REQUIREMENTS, [plan_id=126]
   :     +- Filter isnotnull(_1#6)
   :        +- FileScan parquet [_1#6] Batched: true, DataFilters: [isnotnull(_1#6)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/12/4pf3d5zn72n7q2_0ks3bkh7c0000gn/T/spark-8f..., PartitionFilters: [], PushedFilters: [IsNotNull(_1)], ReadSchema: struct<_1:int>
   +- Sort [_2#11 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(_2#11, 10), ENSURE_REQUIREMENTS, [plan_id=127]
         +- Filter isnotnull(_2#11)
            +- FileScan parquet [_2#11] Batched: true, DataFilters: [isnotnull(_2#11)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/12/4pf3d5zn72n7q2_0ks3bkh7c0000gn/T/spark-8f..., PartitionFilters: [], PushedFilters: [IsNotNull(_2)], ReadSchema: struct<_2:int>
```
- `doCanonicalize` reused a method from `CometScanExec` so I moved it to a new common `CometScanUtils`.

## How are these changes tested?

Existing tests. Enabled one previously skipped test for `native_datafusion`.

Author: mbutrovich

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

@@ -21,9 +21,11 @@ package org.apache.spark.sql.comet
 
 import scala.reflect.ClassTag
 
+import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, UnknownPartitioning}
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.datasources._
@@ -53,29 +55,50 @@ case class CometNativeScanExec(
     disableBucketedScan: Boolean = false,
     originalPlan: FileSourceScanExec,
     override val serializedPlanOpt: SerializedPlan)
-    extends CometLeafExec {
+    extends CometLeafExec
+    with DataSourceScanExec {
 
-  override def nodeName: String =
-    s"${super.nodeName}: ${tableIdentifier.map(_.toString).getOrElse("")}"
+  override lazy val metadata: Map[String, String] = originalPlan.metadata
 
-  override def outputPartitioning: Partitioning =
+  override val nodeName: String =
+    s"CometNativeScan $relation ${tableIdentifier.map(_.unquotedString).getOrElse("")}"
+
+  override lazy val outputPartitioning: Partitioning =
     UnknownPartitioning(originalPlan.inputRDD.getNumPartitions)
 
-  override def outputOrdering: Seq[SortOrder] = originalPlan.outputOrdering
+  override lazy val outputOrdering: Seq[SortOrder] = originalPlan.outputOrdering
+
+  override def doCanonicalize(): CometNativeScanExec = {
+    CometNativeScanExec(
+      nativeOp,
+      relation,
+      output.map(QueryPlan.normalizeExpressions(_, output)),
+      requiredSchema,
+      QueryPlan.normalizePredicates(
+        CometScanUtils.filterUnusedDynamicPruningExpressions(partitionFilters),
+        output),
+      optionalBucketSet,
+      optionalNumCoalescedBuckets,
+      QueryPlan.normalizePredicates(dataFilters, output),
+      None,
+      disableBucketedScan,
+      originalPlan.doCanonicalize(),
+      SerializedPlan(None))
+  }
 
   override def stringArgs: Iterator[Any] = Iterator(output)
 
   override def equals(obj: Any): Boolean = {
     obj match {
       case other: CometNativeScanExec =>
-        this.output == other.output &&
+        this.originalPlan == other.originalPlan &&
         this.serializedPlanOpt == other.serializedPlanOpt
       case _ =>
         false
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output)
+  override def hashCode(): Int = Objects.hashCode(originalPlan, serializedPlanOpt)
 
   override lazy val metrics: Map[String, SQLMetric] = {
     // We don't append CometMetricNode.baselineMetrics because
@@ -153,6 +176,11 @@ case class CometNativeScanExec(
           sparkContext,
           "Time spent reading and parsing metadata from the footer"))
   }
+
+  /**
+   * See [[org.apache.spark.sql.execution.DataSourceScanExec.inputRDDs]]. Only used for tests.
+   */
+  override def inputRDDs(): Seq[RDD[InternalRow]] = originalPlan.inputRDDs()
 }
 
 object CometNativeScanExec extends DataTypeSupport {

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

@@ -459,20 +459,13 @@ case class CometScanExec(
     }
   }
 
-  // Filters unused DynamicPruningExpression expressions - one which has been replaced
-  // with DynamicPruningExpression(Literal.TrueLiteral) during Physical Planning
-  private def filterUnusedDynamicPruningExpressions(
-      predicates: Seq[Expression]): Seq[Expression] = {
-    predicates.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral))
-  }
-
   override def doCanonicalize(): CometScanExec = {
     CometScanExec(
       relation,
       output.map(QueryPlan.normalizeExpressions(_, output)),
       requiredSchema,
       QueryPlan.normalizePredicates(
-        filterUnusedDynamicPruningExpressions(partitionFilters),
+        CometScanUtils.filterUnusedDynamicPruningExpressions(partitionFilters),
         output),
       optionalBucketSet,
       optionalNumCoalescedBuckets,

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

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.comet
+
+import org.apache.spark.sql.catalyst.expressions.{DynamicPruningExpression, Expression, Literal}
+
+object CometScanUtils {
+
+  /**
+   * Filters unused DynamicPruningExpression expressions - one which has been replaced with
+   * DynamicPruningExpression(Literal.TrueLiteral) during Physical Planning
+   */
+  def filterUnusedDynamicPruningExpressions(predicates: Seq[Expression]): Seq[Expression] = {
+    predicates.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral))
+  }
+}

spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala

@@ -813,11 +813,6 @@ class CometExecSuite extends CometTestBase {
   }
 
   test("explain native plan") {
-    // https://github.com/apache/datafusion-comet/issues/1441
-    assume(!CometConf.isExperimentalNativeScan)
-    // there are no assertions in this test to prove that the explain feature
-    // wrote the expected output to stdout, but we at least test that enabling
-    // the config does not cause any exceptions.
     withSQLConf(
       CometConf.COMET_EXPLAIN_NATIVE_ENABLED.key -> "true",
       SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {