peter-toth commented on code in PR #58411:
URL: https://github.com/apache/spark/pull/58411#discussion_r3887586264
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:
##########
@@ -161,6 +165,35 @@ 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 decide what counts
as a malformed record
+ * from the columns it was asked for, which lets a wider read drop or
rewrite rows that the
+ * narrower one returned: CSV, JSON and XML all build their parser from the
required schema and
+ * take `mode` and the corrupt-record column from it. Or the read is not
strict, in which case a
+ * failure in a column that only the wider read touches is swallowed
together with the rest of
+ * that file's rows, whatever the format.
+ *
+ * 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)
+ case _ => false
+ }
+
+ private def hasProjectionSensitiveParser(fileFormat: FileFormat): Boolean =
fileFormat match {
+ case _: CSVFileFormat | _: JsonFileFormat | _: XmlFileFormat => true
Review Comment:
**Finding 1.** `AvroFileFormat.buildReader` hands the pruned
`requiredSchema` to `AvroDeserializer`, and under `positionalFieldMatching`
catalyst field *i* takes Avro field *i* of the full file schema:
`AvroUtils.AvroSchemaHelper.getAvroField` returns
`avroFieldArray.lift(catalystPos)`
(`sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala:462`).
Widening the read therefore changes the values the columns already being read
come back with.
Measured on `00b48e24`, on the default configuration (`avro` is in the
`spark.sql.sources.useV1SourceList` default, so this is the V1 path):
```scala
spark.range(0, 5).selectExpr("id AS a", "id * 10 AS
b").write.format("avro").save(path)
spark.read.option("positionalFieldMatching",
"true").format("avro").load(path)
.createOrReplaceTempView("t")
sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)")
// merged [10, 100] one scan reading {a, b}
// MergeSubplans excluded [10, 10] two scans, {a} and {b}
```
Non-blocking because merging cannot turn a correct answer into a wrong one
here. Pruning keeps the fields' relative order, so a pruned schema is a
subsequence of the full one, and in the union of two subsequences every
retained field's index moves toward its true index without passing it. Both
`[10, 10]` and `[10, 100]` are wrong against the file, which is SPARK-59108.
What the merge does change is that one subquery's values depend on what its
sibling projects, and that is the invariant this PR is about.
#58340 withholds `SCAN_MERGING` from `AvroTable` under this option, so the
two read paths still disagree on it. `AvroFileFormat` lives in `sql/core` under
`org.apache.spark.sql.avro`, so the module argument in the description does not
cover it and this object can name it directly:
```scala
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
// Positional matching resolves each catalyst field against the Avro field
at the same
// position, so pruning shifts which Avro field a column reads. See
SPARK-59108.
case _: AvroFileFormat =>
CaseInsensitiveMap(options).get(AvroOptions.POSITIONAL_FIELD_MATCHING).exists(_.toBoolean)
case _ => false
}
```
Naming `AvroFileFormat` unconditionally would do too, at the cost of every
avro merge.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -212,6 +233,76 @@ 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]]] = {
+ plan.collect {
+ case l: LogicalRelation if
DataSourceUtils.isProjectionSensitiveRead(l.relation) =>
+ l.canonicalized -> readColumnNames(plan, 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.nonEmpty || cachedPlan.projectionSensitiveReads.nonEmpty) &&
Review Comment:
**Finding 5.** Two empty maps are already `==`, so this clause never changes
the result. When both are empty the comparison is false anyway, and when
exactly one is empty the comparison is true and so is the disjunction.
```scala
private def widensProjectionSensitiveRead(
reads: Map[LogicalPlan, Seq[Set[String]]],
cachedPlan: MergedPlan): Boolean = reads !=
cachedPlan.projectionSensitiveReads
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -162,31 +174,40 @@ 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 { _ =>
// Identical subquery expression plans are not marked as `merged` as
the
// `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)) {
Review Comment:
**Finding 2.** `docs/sql-performance-tuning.md:343` lists what two subplans
have to agree on to merge, ending with "and the leaves must read the same
input". This line adds a condition that is not there. It is a default-on
optimization that now silently does not fire for a whole class of relations, so
the tuning page is where someone will look, and #58340 updates the same file
for its half of this.
Something like this, appended to that paragraph:
> A V1 file relation whose rows depend on the columns the read asked for is
merged only when both subplans read the same columns of it. That covers `csv`,
`json` and `xml`, whose parsers decide what counts as a malformed record from
the required schema, and any file relation read with
`spark.sql.files.ignoreCorruptFiles` or `spark.sql.files.ignoreMissingFiles`
set, where a failure in a column only one side reads is swallowed together with
the rest of that file's rows.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:
##########
@@ -161,6 +165,35 @@ 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 decide what counts
as a malformed record
+ * from the columns it was asked for, which lets a wider read drop or
rewrite rows that the
+ * narrower one returned: CSV, JSON and XML all build their parser from the
required schema and
+ * take `mode` and the corrupt-record column from it. Or the read is not
strict, in which case a
+ * failure in a column that only the wider read touches is swallowed
together with the rest of
Review Comment:
**Finding 9.** This attributes the swallowing to both halves of
`hasStrictFileReads`, and only `ignoreCorruptFiles` has it. A missing file is
skipped whatever is projected, so it cannot make a read projection-sensitive.
Your own `FileTable.hasStrictFileReads` in #58340 says exactly that:
> `ignoreMissingFiles` drops the same rows whatever is projected, and is
included to match `FileScanRDD.hasStrictFileReads`, the same predicate on the
physical side.
Worth carrying that sentence over here, because this PR is what makes
`hasStrictFileReads` shared: `FileSourceOptions` now answers it for the
cache-repeatability question in `InMemoryRelation`, for the reader in
`FileScanRDD`, and for this one, and only here is one of the two flags carried
for consistency rather than for a mechanism. No behaviour change asked for.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:
##########
@@ -161,6 +165,35 @@ 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 decide what counts
as a malformed record
+ * from the columns it was asked for, which lets a wider read drop or
rewrite rows that the
+ * narrower one returned: CSV, JSON and XML all build their parser from the
required schema and
+ * take `mode` and the corrupt-record column from it. Or the read is not
strict, in which case a
+ * failure in a column that only the wider read touches is swallowed
together with the rest of
+ * that file's rows, whatever the format.
+ *
+ * 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)
+ case _ => false
+ }
+
+ private def hasProjectionSensitiveParser(fileFormat: FileFormat): Boolean =
fileFormat match {
Review Comment:
**Finding 4.** `supportNestedPredicatePushdown` at line 158 of this file
answers the same kind of question by short name. `hs.toString` is the format's
registered short name, so a list covers every format the same way, including
the ones this object cannot import:
```scala
private def hasProjectionSensitiveParser(hs: HadoopFsRelation): Boolean = {
val sensitive = Utils.stringToSeq(
SQLConf.get.getConf(SQLConf.PROJECTION_SENSITIVE_FILE_SOURCE_LIST).toLowerCase(Locale.ROOT))
sensitive.contains(hs.toString) // "csv", "json", "xml", "avro"
}
```
That drops the three imports, covers avro (finding 1), and lets someone add
a third-party format they know to be sensitive.
The counter-argument is real. `NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST`
gates an optimization, so shortening it costs performance, while shortening
this one returns wrong rows. A hard-coded `Set("csv", "json", "xml", "avro")`
of short names keeps the module independence without that, and is what I would
take if you do not want a new conf.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -212,6 +233,76 @@ 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]]] = {
+ plan.collect {
+ case l: LogicalRelation if
DataSourceUtils.isProjectionSensitiveRead(l.relation) =>
+ l.canonicalized -> readColumnNames(plan, 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.nonEmpty || cachedPlan.projectionSensitiveReads.nonEmpty) &&
+ reads != cachedPlan.projectionSensitiveReads
+ }
+
+ /**
+ * The names of `relation`'s columns that `plan` reads: every attribute
referenced in the plan,
+ * plus the plan's 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(plan: LogicalPlan, relation: LogicalRelation):
Set[String] = {
+ val referenced = AttributeSet(plan.flatMap(_.references)) ++
AttributeSet(plan.output)
Review Comment:
**Finding 6.** `referenced` does not depend on `relation`, but it is rebuilt
for each occurrence, so a plan with N projection-sensitive relations walks
itself N times and builds N `AttributeSet`s. Computing it once at the only
caller makes the independence visible too:
```scala
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)
}
```
##########
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] def hasStrictFileReads: Boolean = options.hasStrictFileReads
Review Comment:
**Finding 7.** Reading `options` from a method body turns the constructor
parameter into a field, so every `FileScanRDD` now carries a
`FileSourceOptions` into the task closure. `javap -p` on the two builds:
```
base private final boolean ...FileScanRDD$$ignoreCorruptFiles;
private final boolean ...FileScanRDD$$ignoreMissingFiles;
head private final org.apache.spark.sql.catalyst.FileSourceOptions options;
(plus both booleans)
```
Tiny in bytes, since `parameters` is `@transient`. Still, the two vals right
above already hold what this needs, so the dedup can keep the parameter
constructor-scoped:
```suggestion
private[sql] val hasStrictFileReads: Boolean = options.hasStrictFileReads
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV1PlanMergingSuite.scala:
##########
@@ -0,0 +1,327 @@
+/*
+ * 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; the two that
+ * only flip `ignoreCorruptFiles` or `ignoreMissingFiles` cannot, because
those rows come back the
+ * same either way, so for them 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(
+ "<rows>",
+ "<row><a>0</a><b>0</b></row>",
+ "<row><a>1</a><b>10</b></row>",
+ "<row><a>2</a><b>BAD</b></row>",
+ "<row><a>3</a><b>30</b></row>",
+ "<row><a>4</a><b>40</b></row>",
+ "</rows>").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.
Review Comment:
**Finding 8.** Missing blank line after `scanColumns`. Same at the two other
places where a `Seq`/`test` follows a closing brace:
`FileSourceV1PlanMergingSuite.scala:140` and `:199`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
