dongjoon-hyun commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4212446466
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -604,12 +655,14 @@ protected boolean loadNextRowGroup() throws IOException {
}
/**
- * A spliced row group's key slots take their values from the survivor queue
at emit, so phase 2
- * reads no pages for them. Key columns are primitive, which is what the
base class asks of a slot
- * answered true here.
+ * The checked slot always holds this reader's own vector, so a file column
of that name, which a
+ * case-insensitive or user-given schema can match, is never decoded into
the batch. A spliced row
+ * group's key slots take their values from the survivor queue at emit, so
phase 2 reads no pages
+ * for them. Both are primitive, which is what the base class asks of a slot
answered true here.
*/
@Override
protected boolean suppliesOwnVector(int slot) {
+ if (slot == checkedSlot) return true;
Review Comment:
**1. A nested file column with the reserved name still fails the read**
The reply says a file holding this column outside the relation's schema is
handled now. From the code, that holds only when the column is primitive. A
group column of that name fails in `initialize`, before `suppliesOwnVector` is
asked:
- `ParquetReadSupport.clipParquetTypeDefault` returns the file's group
unchanged for the Spark `BooleanType` (`case _ => parquetType`).
- `ParquetToSparkSchemaConverter.convertGroupField` then runs
`convertInternal(groupColumn, sparkReadType.map(_.asInstanceOf[StructType]))`
on `BooleanType`, which throws a `ClassCastException`.
Two files would hit this when the table's schema comes from the catalog or
the user and leaves the column out:
- one holding `_tmp_storage_filter_checked struct<a:int>`, or the upper-case
name under case-insensitive resolution;
- one holding both `_tmp_storage_filter_checked` and
`_TMP_STORAGE_FILTER_CHECKED`, which fails case-insensitive clipping with the
duplicate-field error.
A query over such a file worked before this PR, since nothing asked for the
column. The new test covers an INT64 column and a boolean one, but no nested
one.
Could the checked field be left out of the schema handed to clipping and
conversion, with the reader adding the slot itself? That would cover every
type, as well as the wasted IO in comment 4. Otherwise, please add a nested
case to the test and narrow the claim in the javadoc above.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -436,9 +481,13 @@ private void initializeLateMaterialization() throws
IOException {
// a file missing every projected non-key leaf, which also covers the
row-index column. A struct
// the file has, but none of whose requested fields it has, counts only
under the legacy
// `returnNullStructIfAllFieldsMissing`, since otherwise the clipped
schema reads one of its
- // other fields to tell a null struct from one whose requested fields are
all null.
+ // other fields to tell a null struct from one whose requested fields are
all null. The checked
+ // column is left out too, since a file column of its name is never read.
+ String checkedName =
+ checkedSlot >= 0 ? sparkRequestedSchema.fields()[checkedSlot].name() :
null;
List<ColumnDescriptor> nonKey = requestedColumns.stream()
.filter(column -> !keyTopLevelNames.contains(column.getPath()[0]))
+ .filter(column -> !column.getPath()[0].equalsIgnoreCase(checkedName))
Review Comment:
**4. The file's column of that name is still read from storage**
This leaves the file's column of the checked name out of `nonKeyColumns`,
but `requestedColumns` and `requestedPaths` still hold it. So:
- phase 2 reads its chunk whenever splicing or the filter is given up,
through `setRequestedSchema(requestedColumns)`;
- `super.loadNextRowGroup()` reads it for a declined file;
- `hasOffsetIndexes(..., requestedPaths)` and the byte metrics count it.
Nothing ever decodes it, so a wide hidden column (a large binary, say) costs
IO for nothing. Leaving it out of `requestedColumns` as well, or out of the
clipped schema as in comment 1, would avoid that.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +357,34 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
- val outputDataSchema = (readDataColumns ++
generatedMetadataColumns).toStructType
-
// Offered conjuncts become `storageFilters` on the scan and stay in the
post-scan Filter too.
- // This runs here because eligibility depends on `outputDataSchema`.
- val storageFilters =
- storageFiltersFor(normalizedFilters, fsRelation, readDataColumns,
outputDataSchema)
+ // This runs here because eligibility depends on the data columns the
reader reads.
+ val readerDataColumns = readDataColumns ++ generatedMetadataColumns
+ val storageFilters = storageFiltersFor(
+ normalizedFilters, fsRelation, readDataColumns,
readerDataColumns.toStructType)
+ // The column the reader marks the rows it checked in, see
+ // `FileFormat.buildReaderWithStorageFilters`. It is added only where
the scan still returns
+ // columnar batches into whole-stage codegen with it, which fuses the
Filter and the Project
+ // that drops the column. Elsewhere that Project would copy every
surviving row.
+ val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
+ }.filter { checked =>
+ val conf = fsRelation.sparkSession.sessionState.conf
+ val output = readerDataColumns ++ partitionColumns ++
constantMetadataColumns :+ checked
+ conf.wholeStageEnabled && !WholeStageCodegenExec.isTooManyFields(conf,
output.toStructType)
Review Comment:
**2. The gate does not see a `Filter` that whole-stage codegen refuses**
The comment above says the column is added only where the `Filter` and the
`Project` that drops it are fused into whole-stage codegen.
`CollapseCodegenStages.supportCodegen` also refuses an operator whose
expressions contain a `CodegenFallback` (`willFallback`), and this condition
does not check for that.
For example, `SELECT * FROM fact JOIN dim ON ... WHERE exists(fact.tags, t
-> t = 'x')` with an injected runtime bloom. Higher-order functions and
`reflect` are `CodegenFallback`, so:
- the `FilterExec` runs interpreted, outside the stage;
- the scan still gets the checked column;
- the `ProjectExec` that exists only to drop it gets a codegen stage of its
own over an `InputAdapter`, and copies every surviving row through an
`UnsafeProjection`.
That is the cost the gate is meant to avoid. A runtime whole-stage fallback
(`hugeMethodLimit`, a compile failure) ends the same way, though that one
cannot be known at planning time.
Could the condition also require that no conjunct of `afterScanFilters`
contains a `CodegenFallback`? A case in the gate test would cover it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -203,6 +204,7 @@ object ParquetStorageFilter {
s"${ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME}, because the
reader writes row " +
"indexes over that column, so what it reads back depends on how its
row group was read")
+
Review Comment:
**3. The guard that kept a key off the checked slot was dropped**
The removed `require(!originalOrdinals.contains(checkedColumnIndex))` is not
re-established anywhere, while the reader's class javadoc still says that what
`ParquetStorageFilter.create` guarantees is asserted there.
If a planner bug, or a later producer of storage filters, ever binds a key
to that slot, nothing fails fast any more:
- `suppliesOwnVector(checkedSlot)` answers true before the key check, so a
row group that is not spliced never decodes that key;
- the emit loop sets `spliceBatchColumns[slot]` to the persistent or
survivor vector, which replaces `checkedVector` in the batch, while
`putBooleans` keeps writing to the detached one.
The rows would then carry stale key values and no marks, where they used to
get an internal error. Could the check move into the reader, for example in
`initializeLateMaterialization` against `keyIndices`?
Nit: removing the block also left two blank lines here.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -436,9 +481,13 @@ private void initializeLateMaterialization() throws
IOException {
// a file missing every projected non-key leaf, which also covers the
row-index column. A struct
// the file has, but none of whose requested fields it has, counts only
under the legacy
// `returnNullStructIfAllFieldsMissing`, since otherwise the clipped
schema reads one of its
- // other fields to tell a null struct from one whose requested fields are
all null.
+ // other fields to tell a null struct from one whose requested fields are
all null. The checked
+ // column is left out too, since a file column of its name is never read.
+ String checkedName =
+ checkedSlot >= 0 ? sparkRequestedSchema.fields()[checkedSlot].name() :
null;
Review Comment:
**6. Nit: `checkedName` is always the constant**
`checkedColumnIndex` finds the slot by an exact `indexOf` of
`FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME`, so reading the name back from
`sparkRequestedSchema` always gives that constant. Using the constant, guarded
by `checkedSlot >= 0`, would make it clear the name cannot vary.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1982,16 +1982,22 @@ object SQLConf {
"only for the rows that survived. This is a planning-time decision. " +
"A filter that is attached also stays in the post-scan filter, the way
a pushed data " +
"filter does, so the reader is free to stop applying it wherever doing
so would cost " +
- "more than it saves, and the answer does not change. A row group where
it stops reads " +
+ "more than it saves, and the answer does not change. In a scan that
returns columnar " +
+ "batches into whole-stage codegen, the post-scan filter skips it on
the rows the reader " +
+ "applied it to, which the reader marks in a column named
_tmp_storage_filter_checked. " +
+ "A table with a column of that name is not offered storage filters. A
row group where " +
+ "it stops reads " +
Review Comment:
**5. "it" lost its antecedent**
"A row group where it stops reads its key columns twice" used to follow "the
reader is free to stop applying it". With the new sentence in between, "it" now
reads as the table, or as its column. How about:
```suggestion
"A table with a column of that name is not offered storage filters.
A row group where " +
"the reader stops applying the filter reads " +
```
--
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]
