diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index 813401773834..b03d30859e5f 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -42,9 +42,10 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference import org.apache.spark.sql.catalyst.plans.logical.Filter import org.apache.spark.sql.catalyst.util.DateTimeTestUtils import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC} -import org.apache.spark.sql.execution.{FormattedMode, SparkPlan} +import org.apache.spark.sql.execution.{FileSourceScanExec, FormattedMode, SparkPlan} import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition} import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileDataSourceV2, FileTable} +import org.apache.spark.sql.execution.planmerging.MergeSubplans import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.LegacyBehaviorPolicy import org.apache.spark.sql.internal.LegacyBehaviorPolicy._ @@ -3732,6 +3733,42 @@ class AvroV1Suite extends AvroSuite { .sparkConf .set(SQLConf.USE_V1_SOURCE_LIST, "avro") + test("SPARK-59107: positionalFieldMatching makes an avro read projection-sensitive") { + // Strictness pinned rather than inherited, so that positional matching is the only reason the + // read is projection-sensitive. AQE off because `AdaptiveSparkPlanExec` is a leaf node, so with + // it on the scans underneath it are not reachable from the executed plan. + withSQLConf( + SQLConf.IGNORE_CORRUPT_FILES.key -> "false", + SQLConf.IGNORE_MISSING_FILES.key -> "false", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b").write.format("avro").save(path) + withTempView("t") { + spark.read.option("positionalFieldMatching", "true").format("avro").load(path) + .createOrReplaceTempView("t") + val query = "SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)" + // Compared against the same query with merging excluded rather than against a literal + // row: positional matching resolves a column against its position in the read schema, so + // what `sum(b)` answers depends on its own subquery's projection. What this test pins is + // that merging changes neither value. + val unmerged = withSQLConf( + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> MergeSubplans.ruleName) { + sql(query).collect().toSeq + } + val df = sql(query) + checkAnswer(df, unmerged) + val scanColumns = df.queryExecution.executedPlan + .collectWithSubqueries { case s: FileSourceScanExec => s } + .map(_.requiredSchema.fieldNames.sorted.toSeq) + .sortBy(_.mkString(",")) + // One entry per column means the two subqueries kept their own scans. + assert(scanColumns === Seq(Seq("a"), Seq("b"))) + } + } + } + } + test("SPARK-36271: V1 insert should check schema field name too") { withView("v") { spark.range(1).createTempView("v") diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index 13154a752ba6..16be980896b5 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -272,7 +272,7 @@ SELECT They are merged into one aggregate that computes `min` and `max` together, so `store_sales` is read once. In `EXPLAIN` output a merged subplan shows up as a subquery whose single output column is named `mergedValue`, and the sites that share it as `ReusedSubquery`. -Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped. +Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. A V1 file relation whose rows depend on which columns the read asked for is merged only when both subplans read the same columns of it: `csv`, `json` and `xml`, whose parsers decide what counts as a malformed record from the required schema, `avro` read with `positionalFieldMatching`, which pairs a column with the Avro field at its position in that schema, and any file relation read with `spark.sql.files.ignoreCorruptFiles` enabled, as a read option or through the configuration, where a failure in a column only one side reads is swallowed together with the rest of that file's rows. `spark.sql.files.ignoreMissingFiles` counts too, not for that reason but because one predicate answers for both. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped. When only one of the two subplans has a filter, merging is always beneficial, because the unfiltered side reads all the data anyway. This case is on by default, unless the filter has to cross a `Join` to reach the aggregate, which needs the through-join configuration below. When both sides have a filter (the symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less selective than either original filter and can therefore read more data - for example when the filters prune partitions or Parquet row groups. That is why the symmetric case is disabled by default. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala index 1040bb031287..16593a268803 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala @@ -52,6 +52,11 @@ class FileSourceOptions( val ignoreMissingFiles: Boolean = parameters.get(IGNORE_MISSING_FILES).map(_.toBoolean) .getOrElse(SQLConf.get.ignoreMissingFiles) + /** + * Whether a read under these options fails rather than skipping missing or corrupt input files. + */ + def hasStrictFileReads: Boolean = !ignoreCorruptFiles && !ignoreMissingFiles + /** * Whether the data source may read tar archives (.tar/.tar.gz/.tgz) by streaming their entries. * Gated by [[SQLConf.ARCHIVE_FORMAT_READER_ENABLED]] and resolved at construction (on the driver, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala index 5a5466e7fc80..d5dcc42f4f64 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala @@ -380,10 +380,7 @@ case class CachedRDDBuilder( val cachedPlanConf = cachedPlan.conf.clone() def hasStrictReads(conf: SQLConf): Boolean = SQLConf.withExistingConf(conf) { - fileSourceOptions.forall { options => - val effectiveOptions = new FileSourceOptions(options) - !effectiveOptions.ignoreMissingFiles && !effectiveOptions.ignoreCorruptFiles - } + fileSourceOptions.forall(options => new FileSourceOptions(options).hasStrictFileReads) } val (inputRDD, strictPhysicalReads) = SQLConf.withExistingConf(materializationConf) { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala index fe7e8f3a89c0..66f87d93e9ba 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala @@ -29,12 +29,17 @@ import org.json4s.jackson.Serialization import org.apache.spark.{SparkException, SparkUpgradeException} import org.apache.spark.sql.{sources, SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY} +import org.apache.spark.sql.avro.{AvroFileFormat, AvroOptions} +import org.apache.spark.sql.catalyst.FileSourceOptions import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils} import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, Expression, ExpressionSet, PredicateHelper} import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, RebaseDateTime, TypeUtils} import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors} +import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat +import org.apache.spark.sql.execution.datasources.json.JsonFileFormat import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions +import org.apache.spark.sql.execution.datasources.xml.XmlFileFormat import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf} import org.apache.spark.sql.sources.BaseRelation import org.apache.spark.sql.types._ @@ -161,6 +166,47 @@ object DataSourceUtils extends PredicateHelper { case _ => false } + /** + * Returns whether the rows this relation returns, or the values it returns for a column, depend + * on which columns the read was asked for. For such a relation, reading a wider set of columns is + * not just more work: it can return different data for the columns that were already being read. + * + * Two things put a V1 file source here. Its parser may resolve or validate a column against the + * set of columns it was asked for, which lets a wider read drop or rewrite rows that the narrower + * one returned: CSV, JSON and XML build their parser from the required schema and take `mode` and + * the corrupt-record column from it, and Avro under `positionalFieldMatching` pairs a column with + * the Avro field at its position in that schema. SPARK-59108 proposes removing that at the root; + * the Avro arm of `hasProjectionSensitiveParser` can go once that fix is on this branch. Or the + * read is not strict: under `ignoreCorruptFiles` a failure in a column only the wider read + * touches is swallowed together with the rest of that file's rows, whatever the format. + * `ignoreMissingFiles` has no such mechanism, since a missing file is skipped whatever is + * projected; it is here to match `FileSourceOptions.hasStrictFileReads`, the same predicate the + * reader and the cache-repeatability check in `InMemoryRelation` use. + * + * Callers that widen a read need this. Subplan merging is one: top-level column pruning for a V1 + * file source happens in physical planning, from the attributes referenced above the relation, so + * reusing one relation for two subqueries that project different columns widens its read to the + * union of the two column sets. + */ + private[sql] def isProjectionSensitiveRead(relation: BaseRelation): Boolean = relation match { + case hs: HadoopFsRelation => + !new FileSourceOptions(hs.options).hasStrictFileReads || + hasProjectionSensitiveParser(hs.fileFormat, hs.options) + case _ => false + } + + private def hasProjectionSensitiveParser( + fileFormat: FileFormat, options: Map[String, String]): Boolean = fileFormat match { + case _: CSVFileFormat | _: JsonFileFormat | _: XmlFileFormat => true + // Read the option off the map rather than through `AvroOptions`, whose constructor resolves + // `avroSchemaUrl` and would do I/O here, and read it leniently so a malformed value still + // fails where Avro reports it rather than here. + case _: AvroFileFormat => + CaseInsensitiveMap(options).get(AvroOptions.POSITIONAL_FIELD_MATCHING) + .exists("true".equalsIgnoreCase) + case _ => false + } + private def getRebaseSpec( lookupFileMeta: String => String, modeByConfig: String, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala index ee6bf9c0abe9..d1339d493a56 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala @@ -91,7 +91,7 @@ class FileScanRDD( private val ignoreMissingFiles = options.ignoreMissingFiles /** Whether this reader fails instead of silently skipping missing or corrupt input files. */ - private[sql] def hasStrictFileReads: Boolean = !ignoreCorruptFiles && !ignoreMissingFiles + private[sql] val hasStrictFileReads: Boolean = options.hasStrictFileReads // Evaluated on the driver (sparkSession is @transient) and serialized to executors so the // `compute` iterator below can pass it through to ColumnVectorUtils.populate. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala index df74bbe0036d..30087563bca3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala @@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.plans.{Cross, Inner, JoinType, LeftAnti, Le import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join, LogicalPlan, Project} import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.connector.catalog.TableCapability +import org.apache.spark.sql.execution.datasources.{DataSourceUtils, HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, V2ScanRelationPushDown} import org.apache.spark.sql.internal.SQLConf @@ -52,8 +53,19 @@ case class MergeResult( * @param merged Whether this plan is the result of merging two or more plans (true), or * is an original unmerged plan (false). Merged plans typically require special * handling such as wrapping in CTEs. + * @param projectionSensitiveReads The columns the plans behind this entry read from each + * projection-sensitive relation, recorded when the entry was + * first cached. Every plan merged into the entry read those same + * columns, which is what merging one in requires, so the record + * stays true of the entry. It cannot be re-derived from `plan`, + * because a merge leaves projections that the `ColumnPruning` + * rerun after this rule narrows again. See + * `PlanMerger.collectProjectionSensitiveReads`. */ -case class MergedPlan(plan: LogicalPlan, merged: Boolean) +case class MergedPlan( + plan: LogicalPlan, + merged: Boolean, + projectionSensitiveReads: Map[LogicalPlan, Seq[Set[String]]]) object PlanMerger { // Marker tag placed on Filter nodes that were produced by filter propagation. Its presence @@ -158,6 +170,9 @@ class PlanMerger( * - An attribute mapping for rewriting expressions */ def merge(plan: LogicalPlan, subqueryPlan: Boolean): MergeResult = { + // Read once per call rather than once per cache entry, and empty for a plan that reads no + // projection-sensitive relation, which is the common case. + lazy val projectionSensitiveReads = collectProjectionSensitiveReads(plan) cache.zipWithIndex.collectFirst(Function.unlift { case (mp, i) => checkIdenticalPlans(plan, mp.plan).map { _ => @@ -165,24 +180,30 @@ class PlanMerger( // `ReusedSubqueryExec` rule can handle them without extracting the plans to CTEs. // But, when a non-subquery subplan is identical to a cached plan we need to mark the plan // `merged` and so extract it to a CTE later. - val newMergedPlan = MergedPlan(mp.plan, mp.merged || !subqueryPlan) + val newMergedPlan = mp.copy(merged = mp.merged || !subqueryPlan) cache(i) = newMergedPlan val outputMap = AttributeMap(plan.output.zipWithIndex) MergeResult(newMergedPlan, i, outputMap) }.orElse { - tryMergePlans(plan, mp.plan, MergeContext(filterPropagationSupported = false)).collect { - case TryMergeResult(mergedPlan, npMapping, None, None, None, _) => - val newMergedPlan = MergedPlan(mergedPlan, true) - cache(i) = newMergedPlan - val outputMap = AttributeMap(npMapping.iterator.map { case (origAttr, mergedAttr) => - origAttr -> mergedPlan.output.indexWhere(_.exprId == mergedAttr.exprId) - }.toSeq) - MergeResult(newMergedPlan, i, outputMap) + if (widensProjectionSensitiveRead(projectionSensitiveReads, mp)) { + // Reusing the shared relation would widen a read whose rows depend on the set of + // columns it is asked for, see `collectProjectionSensitiveReads`. + None + } else { + tryMergePlans(plan, mp.plan, MergeContext(filterPropagationSupported = false)).collect { + case TryMergeResult(mergedPlan, npMapping, None, None, None, _) => + val newMergedPlan = mp.copy(plan = mergedPlan, merged = true) + cache(i) = newMergedPlan + val outputMap = AttributeMap(npMapping.iterator.map { case (origAttr, mergedAttr) => + origAttr -> mergedPlan.output.indexWhere(_.exprId == mergedAttr.exprId) + }.toSeq) + MergeResult(newMergedPlan, i, outputMap) + } } } case _ => None }).getOrElse { - val newMergedPlan = MergedPlan(plan, false) + val newMergedPlan = MergedPlan(plan, false, projectionSensitiveReads) cache += newMergedPlan val outputMap = AttributeMap(plan.output.zipWithIndex) MergeResult(newMergedPlan, cache.length - 1, outputMap) @@ -208,6 +229,74 @@ class PlanMerger( } } + /** + * The columns each projection-sensitive relation in `plan` is read with, keyed by the + * canonicalized relation. One entry per occurrence, in the order the relations appear, since a + * plan can read the same relation more than once (a self join) with a different set of columns + * each time, and `tryMergePlans` pairs occurrences in that same order. + * + * Top-level column pruning for a V1 file source happens in physical planning, from the attributes + * referenced above the relation, so two `LogicalRelation`s over the same files canonicalize equal + * whatever each side projects, and reusing one of them widens its read to the union of the two + * column sets. For most relations that only changes how much is read, but not for the ones + * [[DataSourceUtils.isProjectionSensitiveRead]] names, where it can change the rows themselves. + * + * Keyed by column name rather than by attribute, because the two plans that get compared were + * analyzed separately and carry different expression ids for the same column. + * + * Only ever called on a plan as it arrives, never on a merged one: merging rebuilds projections + * from a side's whole output, which for a V1 relation is its full schema, and what narrows that + * again is the `ColumnPruning` that `SparkOptimizer`'s `Extract Python UDFs` batch reruns, after + * this rule. A merged cache entry therefore carries the record taken when it was first cached, + * see [[MergedPlan]]. + * + * This compares columns only. Symmetric filter propagation, which is off by default, can also + * widen the set of *files* a scan reads, by OR-ing the two sides' filters: a disjunct mixing a + * partition predicate with a data predicate prunes no partition at all, so the merged scan can + * read the whole table. Each side's own filter above the scan drops the rows that adds, so no + * answer changes, but a projection-sensitive read can still fail on a file neither side selected. + */ + private def collectProjectionSensitiveReads( + plan: LogicalPlan): Map[LogicalPlan, Seq[Set[String]]] = { + lazy val referenced = AttributeSet(plan.flatMap(_.references)) ++ AttributeSet(plan.output) + plan.collect { + case l: LogicalRelation if DataSourceUtils.isProjectionSensitiveRead(l.relation) => + l.canonicalized -> readColumnNames(referenced, l) + }.groupMap(_._1)(_._2) + } + + /** + * Whether merging a plan whose projection-sensitive reads are `reads` into `cachedPlan` could + * change what either of them reads. The two records have to match exactly: a relation read a + * different number of times, or with a different set of columns, or read by only one of the two, + * all count, and a plan that reads a strict subset counts too, because after the merge the entry + * would read more than that plan asked for. + * + * Only [[tryMergePlans]] needs this. Reuse of an identical plan cannot widen a read, because the + * two whole plans are canonically equal there, so everything above the relation references the + * same columns. The relation's own output says nothing about that: on the V1 path it is the full + * schema whatever each side projects, which is what makes this check necessary in the first + * place. + */ + private def widensProjectionSensitiveRead( + reads: Map[LogicalPlan, Seq[Set[String]]], + cachedPlan: MergedPlan): Boolean = reads != cachedPlan.projectionSensitiveReads + + /** + * Which of `relation`'s columns `referenced` covers, by name. `referenced` is every attribute the + * plan refers to plus its own output, since a merged plan is extracted to a CTE and everything + * the CTE outputs is read. Partition columns are left out, because their values come from the + * path rather than from the file, so referencing one does not widen what the reader parses. + */ + private def readColumnNames( + referenced: AttributeSet, relation: LogicalRelation): Set[String] = { + val partitionColumns = relation.relation match { + case hs: HadoopFsRelation => hs.partitionSchema.fieldNames.toSet + case _ => Set.empty[String] + } + referenced.filter(relation.outputSet.contains).map(_.name).toSet -- partitionColumns + } + /** * Result of a successful [[tryMergePlans]] call. * diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV1PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV1PlanMergingSuite.scala new file mode 100644 index 000000000000..16f74b7fde1d --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV1PlanMergingSuite.scala @@ -0,0 +1,330 @@ +/* + * 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.execution.planmerging + +import java.io.File +import java.nio.charset.StandardCharsets +import java.nio.file.Files + +import org.apache.spark.SparkConf +import org.apache.spark.sql.{DataFrame, QueryTest, Row} +import org.apache.spark.sql.execution.FileSourceScanExec +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Tests that subplan merging does not widen the set of columns a V1 file scan reads when the rows + * that scan returns depend on that set. + * + * Every test asserts the columns each scan in the plan reads, so that a decline is attributed to + * the merge being declined rather than inferred from the values, and finding a `FileSourceScanExec` + * at all is what pins the read to the V1 path. Most tests assert the rows as well; in the two that + * only flip `ignoreCorruptFiles` or `ignoreMissingFiles` the rows come back the same either way, so + * there the columns are the whole evidence. + */ +class FileSourceV1PlanMergingSuite extends QueryTest with SharedSparkSession { + + override protected def sparkConf: SparkConf = super.sparkConf + .set(SQLConf.USE_V1_SOURCE_LIST, "avro,csv,json,kafka,orc,parquet,text,xml") + .set(SQLConf.IGNORE_CORRUPT_FILES, false) + .set(SQLConf.IGNORE_MISSING_FILES, false) + .set(SQLConf.SUBQUERY_REUSE_ENABLED, true) + // Off because `AdaptiveSparkPlanExec` is a leaf node, so with it on the scans underneath it are + // not reachable from the executed plan. + .set(SQLConf.ADAPTIVE_EXECUTION_ENABLED, false) + + private val csvRows = "0,0\n1,10\n2,BAD\n3,30\n4,40" + + private val jsonRows = Seq( + """{"a":0,"b":0}""", + """{"a":1,"b":10}""", + """{"a":2,"b":"BAD"}""", + """{"a":3,"b":30}""", + """{"a":4,"b":40}""").mkString("\n") + + private val xmlRows = Seq( + "", + "00", + "110", + "2BAD", + "330", + "440", + "").mkString("\n") + + /** Writes one file of `content` into `dir` and returns the directory to read back. */ + private def writeFile(dir: File, name: String, content: String): String = { + Files.write(new File(dir, name).toPath, content.getBytes(StandardCharsets.UTF_8)) + dir.getCanonicalPath + } + + /** + * The columns each V1 file scan of `df`'s plan reads, one entry per scan, each sorted and the + * entries sorted too, so that an assertion does not depend on plan order. Two entries with a + * column each mean the two subqueries kept their own scans, and one entry holding both columns + * means they shared one, which reuse cannot produce because it only replaces a scan that reads + * the same columns. A replaced scan is absent from the list, since both `ReusedSubqueryExec` and + * `ReusedExchangeExec` are leaf nodes, which is why the self join below shows three scans of the + * four its plan contains. + */ + private def scanColumns(df: DataFrame): Seq[Seq[String]] = + df.queryExecution.executedPlan + .collectWithSubqueries { case s: FileSourceScanExec => s } + .map(_.requiredSchema.fieldNames.sorted.toSeq) + .sortBy(_.mkString(",")) + + // One test per format rather than `gridTest`, so that the format reads in the middle of the name. + Seq( + ("csv", "data.csv", csvRows, Map.empty[String, String]), + ("json", "data.json", jsonRows, Map.empty[String, String]), + ("xml", "data.xml", xmlRows, Map("rowTag" -> "row")) + ).foreach { case (format, fileName, content, extraOptions) => + test(s"SPARK-59107: $format DROPMALFORMED keeps a row malformed only in the other column") { + withTempDir { dir => + val path = writeFile(dir, fileName, content) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .options(extraOptions).format(format).load(path).createOrReplaceTempView("t") + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)") + // A scan of a alone parses no b, so the row malformed in b is not dropped for sum(a). + checkAnswer(df, Row(10L, 80L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("b"))) + } + } + } + } + + test("SPARK-59107: PERMISSIVE does not populate the corrupt-record column of a clean row") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long, _corrupt_record string") + .option("mode", "PERMISSIVE").option("columnNameOfCorruptRecord", "_corrupt_record") + .csv(path).createOrReplaceTempView("t") + val df = sql( + "SELECT (SELECT count(_corrupt_record) FROM t WHERE a >= 0), " + + "(SELECT sum(b) FROM t WHERE a >= 0)") + checkAnswer(df, Row(0L, 80L)) + assert(scanColumns(df) === Seq(Seq("_corrupt_record", "a"), Seq("a", "b"))) + } + } + } + + test("SPARK-59107: FAILFAST does not fail on a short row the narrower scan accepted") { + withTempDir { dir => + // One row carries fewer tokens than the schema has columns, which only a scan reading both + // columns trips. + val path = writeFile(dir, "data.csv", "0,0\n1,10\n2\n3,30\n4,40") + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "FAILFAST") + .csv(path).createOrReplaceTempView("t") + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)") + checkAnswer(df, Row(10L, 80L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("b"))) + } + } + } + + Seq(SQLConf.IGNORE_CORRUPT_FILES, SQLConf.IGNORE_MISSING_FILES).foreach { conf => + test(s"SPARK-59107: a read that is not strict does not share a scan (${conf.key})") { + withTempDir { dir => + val path = new File(dir, "data").getCanonicalPath + spark.range(0, 10).selectExpr("id AS a", "id * 2 AS b").write.parquet(path) + // The view is built outside the configuration scope: the gate is evaluated per merge, so a + // relation built before the configuration was set still answers for the read that runs. + withTempView("t") { + spark.read.parquet(path).createOrReplaceTempView("t") + withSQLConf(conf.key -> "true") { + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)") + // The rows are the same either way here; the columns are what says it declined. + checkAnswer(df, Row(45L, 90L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("b"))) + } + } + } + } + } + + test("SPARK-59107: ignoreCorruptFiles does not drop rows a narrower scan returned") { + withTempDir { dir => + val path = new File(dir, "data").getCanonicalPath + // b is written as a string and read as a long, so the reader fails only when it reads b. + spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path) + withTempView("t") { + spark.read.schema("a long, b long").parquet(path).createOrReplaceTempView("t") + withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") { + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t)") + // sum(a) touches only healthy data. A shared scan would fail on b, and that swallowed + // failure would drop the rest of the file's rows, leaving sum(a) null. + checkAnswer(df, Row(45L, 0L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("b"))) + } + } + } + } + + test("SPARK-59107: a self join reads the relation twice, and each read counts on its own") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", "0,0,0\n1,10,1\n2,BAD,2\n3,30,3\n4,40,4") + withTempView("t") { + spark.read.schema("a long, b long, k long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + // Both subqueries read the relation twice. The right side reads the same columns in both, + // so only the left side differs, and a check that kept one column set per relation rather + // than one per occurrence would compare the right sides and merge. + val df = sql( + "SELECT (SELECT count(t1.a) + sum(t2.b) FROM t t1 LEFT JOIN t t2 ON t1.k = t2.k), " + + "(SELECT count(t1.b) + sum(t2.b) FROM t t1 LEFT JOIN t t2 ON t1.k = t2.k)") + // Merging the left legs would read b for the first subquery too, dropping the row that is + // malformed in b and answering 84 for it. + checkAnswer(df, Row(85L, 84L)) + // Three of the four scans are visible: two subqueries reading (a, k) and (b, k), and one + // more (b, k), the fourth having been replaced by reuse of an identical scan. + assert(scanColumns(df) === Seq(Seq("a", "k"), Seq("b", "k"), Seq("b", "k"))) + } + } + } + + test("SPARK-59107: a merge that rebuilds a projection does not widen the read either") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + // Both subqueries read a alone, and filter propagation, which is on by default, merges + // them by rebuilding the projection above the relation. That rebuilt projection carries the + // relation's full output, but the `ColumnPruning` rerun after this rule narrows it again, + // so the shared scan reads only a and the answers do not change. + val df = sql( + "SELECT (SELECT sum(x) FROM (SELECT a * 2 AS x FROM t WHERE a > 1)), " + + "(SELECT sum(a) FROM t)") + checkAnswer(df, Row(18L, 10L)) + assert(scanColumns(df) === Seq(Seq("a"))) + } + } + } + + test("SPARK-59107: a third subquery is compared against what the merged pair read") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + // The first two subqueries merge on a, and the merged plan carries the relation's full + // output until `ColumnPruning` runs again. The third reads a and b, so it must be compared + // against what the first two read rather than against the merged plan, or it would join + // them and widen their read. + val df = sql( + "SELECT (SELECT sum(x) FROM (SELECT a * 2 AS x FROM t WHERE a > 1)), " + + "(SELECT sum(a) FROM t), (SELECT sum(a + b) FROM t)") + checkAnswer(df, Row(18L, 10L, 88L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("a", "b"))) + } + } + } + + test("SPARK-59107: a fourth subquery joins the entry the refused third one started") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + // The third subquery is refused by the entry the first two share, so it opens one of its + // own, and the fourth has to find that entry rather than stop at the refusal. + val df = sql( + "SELECT (SELECT sum(x) FROM (SELECT a * 2 AS x FROM t WHERE a > 1)), " + + "(SELECT sum(a) FROM t), (SELECT sum(a + b) FROM t), (SELECT count(a + b) FROM t)") + checkAnswer(df, Row(18L, 10L, 88L, 4L)) + assert(scanColumns(df) === Seq(Seq("a"), Seq("a", "b"))) + } + } + } + + test("SPARK-59107: a third subquery reading the same columns still shares the scan") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + val df = sql( + "SELECT (SELECT sum(a) FROM t), (SELECT count(a) FROM t), (SELECT max(a) FROM t)") + checkAnswer(df, Row(10L, 5L, 4L)) + assert(scanColumns(df) === Seq(Seq("a"))) + } + } + } + + test("SPARK-59107: parquet subqueries that project different columns still share a scan") { + withTempDir { dir => + val path = new File(dir, "data").getCanonicalPath + spark.range(0, 10).selectExpr("id AS a", "id * 2 AS b").write.parquet(path) + withTempView("t") { + spark.read.parquet(path).createOrReplaceTempView("t") + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)") + checkAnswer(df, Row(45L, 90L)) + assert(scanColumns(df) === Seq(Seq("a", "b")), + "a format that decodes a row from the columns asked for still merges") + } + } + } + + test("SPARK-59107: csv subqueries that read the same columns still share a scan") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", "0,0\n1,10\n2,20\n3,30\n4,40") + withTempView("t") { + spark.read.schema("a long, b long").csv(path).createOrReplaceTempView("t") + val df = sql( + "SELECT (SELECT sum(a) FROM t WHERE b > 0), (SELECT count(a) FROM t WHERE b > 0)") + // Both subqueries read a and b, so sharing one scan reads no more than either did. + checkAnswer(df, Row(10L, 4L)) + assert(scanColumns(df) === Seq(Seq("a", "b"))) + } + } + } + + test("SPARK-59107: identical subqueries over such a read still share a scan") { + withTempDir { dir => + val path = writeFile(dir, "data.csv", csvRows) + withTempView("t") { + spark.read.schema("a long, b long").option("mode", "DROPMALFORMED") + .csv(path).createOrReplaceTempView("t") + // End to end this shape runs one scan whatever the merger decides, because subquery reuse + // collapses two identical subqueries on its own. Here to pin that it stays that way. + val df = sql("SELECT (SELECT sum(b) FROM t), (SELECT sum(b) FROM t) + 1") + checkAnswer(df, Row(80L, 81L)) + assert(scanColumns(df) === Seq(Seq("b"))) + } + } + } + + test("SPARK-59107: a partition column reference does not count as a column read") { + withTempDir { dir => + val path = new File(dir, "data").getCanonicalPath + spark.range(0, 8).selectExpr("id AS a", "id % 2 AS p", "id % 4 AS q") + .write.partitionBy("p", "q").csv(path) + withTempView("t") { + spark.read.schema("a long, p long, q long").csv(path).createOrReplaceTempView("t") + // q comes from the path rather than from the file, so the two subqueries read the same + // column of it, a, and the merge is allowed even though this is a csv relation. + val df = sql( + "SELECT (SELECT sum(a) FROM t WHERE p = 1), (SELECT sum(a + q) FROM t WHERE p = 1)") + checkAnswer(df, Row(16L, 24L)) + assert(scanColumns(df) === Seq(Seq("a"))) + } + } + } +}