remove the customer equality

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

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.execution.datasources.v2
 
-import java.util.Objects
-
 import org.apache.commons.lang3.StringUtils
 
 import org.apache.spark.sql.catalyst.expressions.Attribute
@@ -63,22 +61,6 @@ trait DataSourceV2QueryPlan {
     case _ => Set.empty
   }
 
-  /**
-   * The metadata of this data source query plan that can be used for equality check.
-   */
-  private def metadata: Seq[Any] = Seq(source.getClass, output, options, filters)
-
-  def canEqual(other: Any): Boolean
-
-  override def equals(other: Any): Boolean = other match {
-    case other: DataSourceV2QueryPlan => canEqual(other) && metadata == other.metadata
-    case _ => false
-  }
-
-  override def hashCode(): Int = {
-    metadata.map(Objects.hashCode).foldLeft(0)((a, b) => 31 * a + b)
-  }
-
   private def sourceName: String = source match {
     case registered: DataSourceRegister => registered.shortName()
     case _ => source.getClass.getSimpleName.stripSuffix("$")

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

@@ -40,8 +40,6 @@ case class DataSourceV2Relation(
 
   import DataSourceV2Relation._
 
-  override def canEqual(other: Any): Boolean = other.isInstanceOf[DataSourceV2Relation]
-
   override def simpleString: String = "V2Relation " + metadataString
 
   override lazy val schema: StructType = reader.readSchema()
@@ -121,8 +119,6 @@ case class StreamingDataSourceV2Relation(
 
   override def isStreaming: Boolean = true
 
-  override def canEqual(other: Any): Boolean = other.isInstanceOf[StreamingDataSourceV2Relation]
-
   override def simpleString: String = "Streaming V2Relation " + metadataString
 
   override def newInstance(): LogicalPlan = copy(output = output.map(_.newInstance()))

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

@@ -42,8 +42,6 @@ case class DataSourceV2ScanExec(
     @transient reader: DataSourceReader)
   extends LeafExecNode with DataSourceV2QueryPlan with ColumnarBatchScan {
 
-  override def canEqual(other: Any): Boolean = other.isInstanceOf[DataSourceV2ScanExec]
-
   override def simpleString: String = "V2Scan " + metadataString
 
   override def outputPartitioning: physical.Partitioning = reader match {