peter-toth commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4218817819
##########
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:
Thanks, you're right, and I overstated it in my reply to @sunchao. A file
with a nested `_tmp_metadata_row_index` fails the same way today when
`_metadata.row_index` is selected. I checked both.
So I narrowed the claim in
[a6cf0027a62](https://github.com/apache/spark/pull/59269/commits/a6cf0027a62f420d1252bda264714885b9813f51)
rather than changing the clipping.
`FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME` and the conf doc now say that a
file holding a column of that name outside the table's schema is not supported,
as for the row-index column. The reader javadoc says a primitive one is still
never decoded.
##########
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:
Thanks, done in
[a6cf0027a62](https://github.com/apache/spark/pull/59269/commits/a6cf0027a62f420d1252bda264714885b9813f51).
The gate also requires that no conjunct of the `Filter` falls back from
codegen, as `CollapseCodegenStages.supportCodegen` decides. The gate test has a
case with a `CodegenFallback` conjunct. A runtime fallback cannot be known at
planning, as you say.
I also measured it locally, with a literal bloom over a columnar scan. Only
one case showed a difference: 90 long columns, 50% of the rows surviving and a
`CodegenFallback` conjunct. There the column made the query about 10% slower.
With 8 long columns, with a 2 kB binary, or with 5% surviving, both were the
same within noise.
##########
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:
Done in
[a6cf0027a62](https://github.com/apache/spark/pull/59269/commits/a6cf0027a62f420d1252bda264714885b9813f51).
`create` rejects a key column of that name again, next to the same check for
the row-index column, and the create test covers it. The double blank line is
gone.
##########
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:
I'd leave this one. A file holding a column of that name outside the table's
schema is not supported (comment 1), and the reader only makes sure a primitive
one is not decoded. Leaving it out of `requestedColumns` would not cover a
declined file either, which the base reader reads.
##########
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:
Done in
[a6cf0027a62](https://github.com/apache/spark/pull/59269/commits/a6cf0027a62f420d1252bda264714885b9813f51),
with your wording.
##########
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:
Done in
[a6cf0027a62](https://github.com/apache/spark/pull/59269/commits/a6cf0027a62f420d1252bda264714885b9813f51).
--
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]
