peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3881910283
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:
##########
@@ -111,7 +120,15 @@ abstract class FileTable(
override def properties: util.Map[String, String] =
options.asCaseSensitiveMap
- override def capabilities: java.util.Set[TableCapability] =
FileTable.CAPABILITIES
+ override def capabilities: java.util.Set[TableCapability] =
+ if (supportsScanMerging) FileTable.CAPABILITIES_WITH_SCAN_MERGING else
FileTable.CAPABILITIES
Review Comment:
**Finding 8.** The criterion holds only while the read is strict.
Your reply on @dongjoon-hyun's thread justifies the four overrides with "a
corrupt column chunk, or `datetimeRebaseModeInRead=EXCEPTION` meeting an
ancient value, makes any format throw on a column the narrower scan had
pruned". That is true only with `spark.sql.files.ignoreCorruptFiles=false`.
With it on, `FilePartitionReader.next` catches any `RuntimeException` or
`IOException` and returns `false`, dropping the rest of that file
(`FilePartitionReader.scala:74`, via
`DataSourceUtils.shouldIgnoreCorruptFileException`). So the merged scan returns
fewer rows than the narrow scan did. That is the "silent and unrecoverable
change to row membership" you gave as what separates CSV and JSON, reached by a
format that declares the capability.
Measured on this head, parquet, `spark.sql.files.ignoreCorruptFiles=true`.
`b` is written as a string and read as a long, so the reader throws only when
it actually reads `b`; a corrupt column chunk is the same code path, just
harder to build in a test.
```scala
spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS
b").write.parquet(path)
spark.read.schema("a long, b
long").parquet(path).createOrReplaceTempView("t")
sql("SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t)")
```
| path | default | `MergeSubplans` excluded |
|---|---|---|
| V2 | `[null, 0]` | `[45, 0]` |
| V1 | `[null, 0]` | `[45, 0]` |
`sum(a)` touches only healthy data and is correct at 45 today; after the
merge it is `null`. V1 is the same, so this is V1 parity rather than a V1/V2
split - but it is the parity you declined to copy for CSV and JSON, on the same
reasoning. ORC did not reproduce with this particular type mismatch; the
exposure is any per-column read failure, so it is not parquet-specific.
The fix has a precedent in the tree: `InMemoryRelation` refuses to treat a
file scan as repeatable under either flag (`InMemoryRelation.scala:378-398`,
and `FileScanRDD.hasStrictFileReads`).
```scala
override def capabilities: java.util.Set[TableCapability] =
if (supportsScanMerging && hasStrictFileReads) {
FileTable.CAPABILITIES_WITH_SCAN_MERGING
} else {
FileTable.CAPABILITIES
}
/**
* Whether a read of this table is strict. A best-effort read is not
reproducible: a failure on a
* column only the other scan projects is swallowed, so the merged scan
would not read a superset
* of either input's rows.
*/
private def hasStrictFileReads: Boolean = {
val fileSourceOptions = new
FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap)
!fileSourceOptions.ignoreCorruptFiles &&
!fileSourceOptions.ignoreMissingFiles
}
```
`FileSourceOptions` is already imported here and it resolves the per-read
option over the session conf, so both spellings are covered. Only
`ignoreCorruptFiles` has the failure above - a missing file drops the same rows
whatever is projected - so including `ignoreMissingFiles` is just matching the
existing predicate; drop it if you prefer the narrower gate.
If you would rather keep the capability unconditional, then the
migration-guide entry has to come back for this shape. Unlike the errors the
criterion accepts, this one is a silent result change.
##########
connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala:
##########
@@ -52,4 +52,9 @@ case class AvroTable(
override def supportsDataType(dataType: DataType): Boolean =
AvroUtils.supportsDataType(dataType)
override def formatName: String = "Avro"
+
+ // Every record is decoded against the full schema before the projection is
applied to the decoded
Review Comment:
**Finding 9.** The first sentence is not what `AvroPartitionReaderFactory`
does.
The deserializer is built from `readDataSchema`, i.e. the projection, not
the full schema (`AvroPartitionReaderFactory.scala:101-111`).
`positionalFieldMatching` is passed straight through, and
`AvroUtils.AvroSchemaHelper.getAvroField` then resolves a catalyst field by its
*position in that projected schema* (`AvroUtils.scala:463-469`). So the
projection is exactly what decides which Avro field feeds which column, and
widening it re-maps them.
Measured on this head. File fields `a`, `b`, `c` holding `id`, `100 * id`,
`10000 * id` for ids 0 to 4, read with `positionalFieldMatching=true`, query
`SELECT (SELECT sum(a) FROM t), (SELECT sum(c) FROM t)`:
| path | default | `MergeSubplans` excluded |
|---|---|---|
| V2 | `[10, 1000]` | `[10, 10]` |
| V1 | `[10, 1000]` | `[10, 10]` |
`sum(c)` is 100000. Both values are wrong, because Avro positional matching
against a pruned schema is already broken without this PR: the `c`-only scan
resolves position 0 and reads `a`. Merging cannot turn a correct answer into a
wrong one here either, since the union of two prefixes is the longer prefix. So
this is not a correctness regression, and I am not asking you to exclude Avro.
Two smaller asks. State the mechanism as it is - what makes merging safe for
Avro is that the format has no record-level parse verdict, not that decoding
ignores the projection - so a later reader does not build on a property the
reader does not have. And "No query result changes" in the description needs a
qualifier, or the pruning bug needs its own JIRA to point at.
@dongjoon-hyun named `positionalFieldMatching` as the Avro-specific risk on
the coverage thread above; the new `AvroV2Suite` test does not reach it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala:
##########
@@ -70,4 +70,8 @@ case class ParquetTable(
}
override def formatName: String = "Parquet"
+
+ // A row is decoded from the column chunks the scan asked for, so reading
more columns can only
+ // surface an error, never silently change which rows come back.
+ override def supportsScanMerging: Boolean = true
Review Comment:
**Finding 11.** This override widens the seam from `protected` to public.
`FileTable.supportsScanMerging` is `protected`, and an override with no
access modifier is public in Scala. So `ParquetTable`, `OrcTable`, `TextTable`
and `AvroTable` each expose it as public API while `CSVTable` and `JsonTable`
keep it protected. Either add the modifier to the four, or drop it from the
base if the seam is meant to be part of the format-author surface.
```suggestion
override protected def supportsScanMerging: Boolean = true
```
--
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]
