This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 4d5f5e7 Revert "[SPARK-33482][SPARK-34756][SQL] Fix FileScan equality
check"
4d5f5e7 is described below
commit 4d5f5e7532afb283bc093e77f082bd573f66cc59
Author: Dongjoon Hyun <dh...@apple.com>
AuthorDate: Tue Mar 23 10:57:18 2021 -0700
Revert "[SPARK-33482][SPARK-34756][SQL] Fix FileScan equality check"
This reverts commit 54295af99ea5819eb18cd6b7c3344de2190254a4.
---
.../org/apache/spark/sql/avro/AvroScanSuite.scala | 30 --
.../sql/execution/datasources/v2/FileScan.scala | 22 +-
.../scala/org/apache/spark/sql/FileScanSuite.scala | 374 ---------------------
.../scala/org/apache/spark/sql/SQLQuerySuite.scala | 24 --
4 files changed, 4 insertions(+), 446 deletions(-)
diff --git
a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroScanSuite.scala
b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroScanSuite.scala
deleted file mode 100644
index 98a7190..0000000
--- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroScanSuite.scala
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.avro
-
-import org.apache.spark.sql.FileScanSuiteBase
-import org.apache.spark.sql.v2.avro.AvroScan
-
-class AvroScanSuite extends FileScanSuiteBase {
- val scanBuilders = Seq[(String, ScanBuilder, Seq[String])](
- ("AvroScan",
- (s, fi, ds, rds, rps, f, o, pf, df) => AvroScan(s, fi, ds, rds, rps, o,
f, pf, df),
- Seq.empty))
-
- run(scanBuilders)
-}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
index 1066c76..6e05aa5 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala
@@ -24,9 +24,8 @@ 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.{AttributeSet, Expression,
ExpressionSet}
+import org.apache.spark.sql.catalyst.expressions.{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._
@@ -83,24 +82,11 @@ trait FileScan extends Scan with Batch with
SupportsReportStatistics with Loggin
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)))))
- 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 {
case f: FileScan =>
- fileIndex == f.fileIndex && readSchema == f.readSchema &&
- normalizedPartitionFilters == f.normalizedPartitionFilters &&
- normalizedDataFilters == f.normalizedDataFilters
+ fileIndex == f.fileIndex && readSchema == f.readSchema
+ ExpressionSet(partitionFilters) == ExpressionSet(f.partitionFilters) &&
+ ExpressionSet(dataFilters) == ExpressionSet(f.dataFilters)
case _ => false
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/FileScanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/FileScanSuite.scala
deleted file mode 100644
index c7087ac..0000000
--- a/sql/core/src/test/scala/org/apache/spark/sql/FileScanSuite.scala
+++ /dev/null
@@ -1,374 +0,0 @@
-/*
- * 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
-
-import scala.collection.mutable
-
-import com.google.common.collect.ImmutableMap
-import org.apache.hadoop.fs.{FileStatus, Path}
-
-import org.apache.spark.sql.catalyst.dsl.expressions._
-import org.apache.spark.sql.catalyst.expressions.{And, Expression, IsNull,
LessThan}
-import org.apache.spark.sql.execution.datasources.{PartitioningAwareFileIndex,
PartitionSpec}
-import org.apache.spark.sql.execution.datasources.v2.FileScan
-import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
-import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
-import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
-import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
-import org.apache.spark.sql.execution.datasources.v2.text.TextScan
-import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.util.CaseInsensitiveStringMap
-
-trait FileScanSuiteBase extends SharedSparkSession {
- private def newPartitioningAwareFileIndex() = {
- new PartitioningAwareFileIndex(spark, Map.empty, None) {
- override def partitionSpec(): PartitionSpec = {
- PartitionSpec.emptySpec
- }
-
- override protected def leafFiles: mutable.LinkedHashMap[Path,
FileStatus] = {
- mutable.LinkedHashMap.empty
- }
-
- override protected def leafDirToChildrenFiles: Map[Path,
Array[FileStatus]] = {
- Map.empty
- }
-
- override def rootPaths: Seq[Path] = {
- Seq.empty
- }
-
- override def refresh(): Unit = {}
- }
- }
-
- type ScanBuilder = (
- SparkSession,
- PartitioningAwareFileIndex,
- StructType,
- StructType,
- StructType,
- Array[Filter],
- CaseInsensitiveStringMap,
- Seq[Expression],
- Seq[Expression]) => FileScan
-
- def run(scanBuilders: Seq[(String, ScanBuilder, Seq[String])]): Unit = {
- val dataSchema = StructType.fromDDL("data INT, partition INT, other INT")
- val dataSchemaNotEqual = StructType.fromDDL("data INT, partition INT,
other INT, new INT")
- val readDataSchema = StructType.fromDDL("data INT")
- val readDataSchemaNotEqual = StructType.fromDDL("data INT, other INT")
- val readPartitionSchema = StructType.fromDDL("partition INT")
- val readPartitionSchemaNotEqual = StructType.fromDDL("partition INT, other
INT")
- val pushedFilters =
- Array[Filter](sources.And(sources.IsNull("data"),
sources.LessThan("data", 0)))
- val pushedFiltersNotEqual =
- Array[Filter](sources.And(sources.IsNull("data"),
sources.LessThan("data", 1)))
- val optionsMap = ImmutableMap.of("key", "value")
- val options = new CaseInsensitiveStringMap(ImmutableMap.copyOf(optionsMap))
- val optionsNotEqual =
- new CaseInsensitiveStringMap(ImmutableMap.copyOf(ImmutableMap.of("key2",
"value2")))
- val partitionFilters = Seq(And(IsNull('data.int), LessThan('data.int, 0)))
- val partitionFiltersNotEqual = Seq(And(IsNull('data.int),
LessThan('data.int, 1)))
- val dataFilters = Seq(And(IsNull('data.int), LessThan('data.int, 0)))
- val dataFiltersNotEqual = Seq(And(IsNull('data.int), LessThan('data.int,
1)))
-
- scanBuilders.foreach { case (name, scanBuilder, exclusions) =>
- test(s"SPARK-33482: Test $name equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanEquals = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema.copy(),
- readDataSchema.copy(),
- readPartitionSchema.copy(),
- pushedFilters.clone(),
- new CaseInsensitiveStringMap(ImmutableMap.copyOf(optionsMap)),
- Seq(partitionFilters: _*),
- Seq(dataFilters: _*))
-
- assert(scan === scanEquals)
- }
-
- test(s"SPARK-33482: Test $name fileIndex not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val partitioningAwareFileIndexNotEqual =
newPartitioningAwareFileIndex()
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndexNotEqual,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
-
- if (!exclusions.contains("dataSchema")) {
- test(s"SPARK-33482: Test $name dataSchema not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchemaNotEqual,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
- }
-
- test(s"SPARK-33482: Test $name readDataSchema not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchemaNotEqual,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
-
- test(s"SPARK-33482: Test $name readPartitionSchema not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchemaNotEqual,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
-
- if (!exclusions.contains("pushedFilters")) {
- test(s"SPARK-33482: Test $name pushedFilters not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFiltersNotEqual,
- options,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
- }
-
- test(s"SPARK-33482: Test $name options not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- optionsNotEqual,
- partitionFilters,
- dataFilters)
-
- assert(scan !== scanNotEqual)
- }
-
- test(s"SPARK-33482: Test $name partitionFilters not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFiltersNotEqual,
- dataFilters)
- assert(scan !== scanNotEqual)
- }
-
- test(s"SPARK-33482: Test $name dataFilters not equals") {
- val partitioningAwareFileIndex = newPartitioningAwareFileIndex()
-
- val scan = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFilters)
-
- val scanNotEqual = scanBuilder(
- spark,
- partitioningAwareFileIndex,
- dataSchema,
- readDataSchema,
- readPartitionSchema,
- pushedFilters,
- options,
- partitionFilters,
- dataFiltersNotEqual)
- assert(scan !== scanNotEqual)
- }
- }
- }
-}
-
-class FileScanSuite extends FileScanSuiteBase {
- val scanBuilders = Seq[(String, ScanBuilder, Seq[String])](
- ("ParquetScan",
- (s, fi, ds, rds, rps, f, o, pf, df) =>
- ParquetScan(s, s.sessionState.newHadoopConf(), fi, ds, rds, rps, f, o,
pf, df),
- Seq.empty),
- ("OrcScan",
- (s, fi, ds, rds, rps, f, o, pf, df) =>
- OrcScan(s, s.sessionState.newHadoopConf(), fi, ds, rds, rps, o, f, pf,
df),
- Seq.empty),
- ("CSVScan",
- (s, fi, ds, rds, rps, f, o, pf, df) => CSVScan(s, fi, ds, rds, rps, o,
f, pf, df),
- Seq.empty),
- ("JsonScan",
- (s, fi, ds, rds, rps, f, o, pf, df) => JsonScan(s, fi, ds, rds, rps, o,
pf, df),
- Seq("pushedFilters")),
- ("TextScan",
- (s, fi, _, rds, rps, _, o, pf, df) => TextScan(s, fi, rds, rps, o, pf,
df),
- Seq("dataSchema", "pushedFilters")))
-
- run(scanBuilders)
-}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index abb0650..88524d7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -37,7 +37,6 @@ import
org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupporte
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
-import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
CartesianProductExec, SortMergeJoinExec}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
@@ -3691,29 +3690,6 @@ class SQLQuerySuite extends QueryTest with
SharedSparkSession with AdaptiveSpark
}
}
}
-
- test("SPARK-33482: Fix FileScan canonicalization") {
- withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
- withTempPath { path =>
- spark.range(5).toDF().write.mode("overwrite").parquet(path.toString)
- withTempView("t") {
- spark.read.parquet(path.toString).createOrReplaceTempView("t")
- val df = sql(
- """
- |SELECT *
- |FROM t AS t1
- |JOIN t AS t2 ON t2.id = t1.id
- |JOIN t AS t3 ON t3.id = t2.id
- |""".stripMargin)
- df.collect()
- val reusedExchanges = collect(df.queryExecution.executedPlan) {
- case r: ReusedExchangeExec => r
- }
- assert(reusedExchanges.size == 1)
- }
- }
- }
- }
}
case class Foo(bar: Option[String])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
