Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-33482][SPARK-34756][SQL] Fix FileScan equality check #31848

Closed
wants to merge 10 commits into from
Closed

[SPARK-33482][SPARK-34756][SQL] Fix FileScan equality check #31848

Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e1a9722
[SPARK-34756][SQL] Fix FileScan equality check
peter-toth Mar 16, 2021
f65ebe3
normalize filters in `FileScan.equals()`
peter-toth Mar 16, 2021
776828e
add attribute name normalization
peter-toth Mar 17, 2021
6709185
add e3e test
peter-toth Mar 17, 2021
b631350
Merge commit '2e836cdb598255d6b43b386e98fcb79b70338e69' into SPARK-34…
peter-toth Mar 18, 2021
9711da0
add FileScanSuite and AvroScanSuite
peter-toth Mar 18, 2021
d782723
compact test code
peter-toth Mar 19, 2021
241e6e4
Merge commit '7953fcdb56abd2e574ea7fdfed925fe06dffdec0' into SPARK-34…
peter-toth Mar 22, 2021
30d2d8b
remove the case sensitivity handling
peter-toth Mar 22, 2021
0d38ac2
fix naming
peter-toth Mar 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 18 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,9 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config.IO_WARNING_LARGEFILETHRESHOLD
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionSet}
import org.apache.spark.sql.catalyst.expressions.{AttributeSet, Expression, ExpressionSet}
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.connector.read.{Batch, InputPartition, Scan, Statistics, SupportsReportStatistics}
import org.apache.spark.sql.execution.PartitionedFileUtil
import org.apache.spark.sql.execution.datasources._
Expand Down Expand Up @@ -84,11 +85,24 @@ trait FileScan extends Scan

protected def seqToString(seq: Seq[Any]): String = seq.mkString("[", ", ", "]")

private lazy val (normalizedPartitionFilters, normalizedDataFilters) = {
val output = readSchema().toAttributes
val partitionFilterAttributes = AttributeSet(partitionFilters).map(a => a.name -> a).toMap
val dataFiltersAttributes = AttributeSet(dataFilters).map(a => a.name -> a).toMap
val normalizedPartitionFilters = ExpressionSet(partitionFilters.map(
QueryPlan.normalizeExpressions(_, output.map(a =>
partitionFilterAttributes.getOrElse(a.name, a)))))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shall we consider case sensitivity here?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

or the attr name inside the data/partitionFilterAttributes have been normalized already before?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we shall, added in 776828e

val normalizedDataFilters = ExpressionSet(dataFilters.map(
QueryPlan.normalizeExpressions(_, output.map(a =>
dataFiltersAttributes.getOrElse(a.name, a)))))
(normalizedPartitionFilters, normalizedDataFilters)
}

override def equals(obj: Any): Boolean = obj match {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall I update hashCode() as well? Looks like we can easily come up with a better one...

Copy link
Member

@dongjoon-hyun dongjoon-hyun Mar 18, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we do that separately because it's irrelevant to the correctness issue?
In general, we expect a performance improvement with that, don't we?
Apache Spark doesn't allow to backport performance improvement.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please file a new JIRA and go for it, @peter-toth ! :)

case f: FileScan =>
fileIndex == f.fileIndex && readSchema == f.readSchema
ExpressionSet(partitionFilters) == ExpressionSet(f.partitionFilters) &&
ExpressionSet(dataFilters) == ExpressionSet(f.dataFilters)
fileIndex == f.fileIndex && readSchema == f.readSchema &&
normalizedPartitionFilters == f.normalizedPartitionFilters &&
normalizedDataFilters == f.normalizedDataFilters

case _ => false
}
Expand Down