Strengthen data source v2 operators' equality checks

Author: li-boxuan

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala

@@ -48,8 +48,8 @@ case class BatchScanExec(
   // TODO: unify the equal/hashCode implementation for all data source v2 query plans.
   override def equals(other: Any): Boolean = other match {
     case other: BatchScanExec =>
-      this.batch != null && this.batch == other.batch &&
-          this.runtimeFilters == other.runtimeFilters &&
+      this.getClass == other.getClass && this.batch != null &&
+        this.batch == other.batch && this.runtimeFilters == other.runtimeFilters &&
           this.spjParams == other.spjParams
     case _ =>
       false

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ContinuousScanExec.scala

@@ -38,7 +38,7 @@ case class ContinuousScanExec(
 
   // TODO: unify the equal/hashCode implementation for all data source v2 query plans.
   override def equals(other: Any): Boolean = other match {
-    case other: ContinuousScanExec => this.stream == other.stream
+    case other: ContinuousScanExec => this.getClass == other.getClass && this.stream == other.stream
     case _ => false
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MicroBatchScanExec.scala

@@ -41,7 +41,7 @@ case class MicroBatchScanExec(
 
   // TODO: unify the equal/hashCode implementation for all data source v2 query plans.
   override def equals(other: Any): Boolean = other match {
-    case other: MicroBatchScanExec => this.stream == other.stream
+    case other: MicroBatchScanExec => this.getClass == other.getClass && this.stream == other.stream
     case _ => false
   }