peter-toth commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4211298887
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -418,11 +441,19 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
Alias(KnownNotNull(CreateStruct(structColumns.toImmutableArraySeq)),
FileFormat.METADATA_NAME)(exprId = metadataStruct.exprId)
execution.ProjectExec(
- readDataColumns ++ partitionColumns :+ metadataAlias, scan)
+ readDataColumns ++ storageFilterCheckedColumn ++ partitionColumns :+
metadataAlias, scan)
}.getOrElse(scan)
+ // The offered conjuncts are skipped on the rows the reader marks as
checked. They are
+ // matched by `ExpressionSet`, since `storageFilters` carry the
relation's column names.
+ val postScanFilters = storageFilterCheckedColumn match {
+ case Some(checked) =>
+ val offered = ExpressionSet(storageFilters)
+ afterScanFilters.toSeq.map(f => if (offered.contains(f)) Or(checked,
f) else f)
Review Comment:
Thanks, fixed in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
The rewrite is `IF(checked, true, conjunct)` now, as @dongjoon-hyun suggested
on the same line. A null mark then leaves the conjunct, and the `AND` around
it, as they were.
Since the fix for his comment on L373, the column goes only to a scan that
returns columnar batches into whole-stage codegen, which parquet-mr does not
read. So the new test checks the expression itself, generated and interpreted:
a null mark, a bloom that rejects the row, and a conjunct after it that throws.
It also checks that the `OR` form throws.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1982,13 +1982,16 @@ 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. The post-scan
filter skips it on " +
+ "the rows the reader applied it to, except where one more column would
push the scan " +
+ "past spark.sql.codegen.maxFields. A row group where it stops reads " +
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
The conf doc and the user-facing section now say that a table with a column
named `_tmp_storage_filter_checked` is not offered storage filters.
A file that holds such a column outside the relation's schema is handled now
too. The late-materialization reader never reads it and writes the marks itself
(@dongjoon-hyun's comment on L203). parquet-mr does not get the column, since
it goes only to a scan that returns columnar batches.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -194,6 +200,10 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
outputDataSchema: StructType): Seq[Expression] = {
val sparkSession = fsRelation.sparkSession
if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return Nil
+ if (fsRelation.schema.fieldNames.exists(
Review Comment:
Thanks, done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
`suppliesOwnVector` answers true for the checked slot, and its column is left
out of `nonKey`, compared case-insensitively. So the file's column is never
decoded.
A new test reads a file with an upper-case INT64 column of that name, which
failed before. It also reads a file whose only other column is a boolean one of
that name, which must still be declined.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -595,6 +634,8 @@ protected boolean loadNextRowGroup() throws IOException {
m.rowsExcludedWithinRowGroup().add(baselineRows - finalRowCount);
}
+ // The rows phase 2 read are exactly the survivors unless the filter was
given up.
+ rowsChecked = !filterGivenUp;
Review Comment:
Thanks for the detailed analysis. I'd keep marking these row groups checked,
and keep the caveat documented.
- Splicing is given up mostly because the survivors pass the budget. So
these are the row groups with the most survivors, and re-checking them would
cost the most.
- A wrong offset index can already pair a row's key with another row's
values, which the `Filter` cannot catch either.
- `parquet.filter.columnindex.enabled=false` turns both off, and the conf
doc points to it.
I fixed the two places you quoted. `FileFormat`'s contract now says
"evaluated on the row", without "the row's own values". `FileSourceStrategy` no
longer calls the check exact.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +360,36 @@ 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 storageFilters = storageFiltersFor(normalizedFilters, fsRelation,
readDataColumns,
+ (readDataColumns ++ generatedMetadataColumns).toStructType)
+ // The column the reader marks the rows it checked in, which the
post-scan Filter reads. See
+ // `FileFormat.buildReaderWithStorageFilters`. It is left out where it
alone would push the
+ // scan past the whole-stage codegen field limit, which would cost the
scan its columnar
+ // output. The post-scan Filter then evaluates the offered conjuncts on
every row.
+ val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
+ }.filterNot { checked =>
Review Comment:
Thanks, done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4),
with your condition. The column goes only to a scan that returns columnar
batches into whole-stage codegen with it, where the `Filter` and the `Project`
are fused. Everywhere else the scan still gets the filter but no column, and
the `Filter` evaluates the bloom on every row, as before this PR.
The test now has a partition column, and covers the limit with and without
the column, and whole-stage codegen off. It checks `supportsColumnar` at each.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +360,36 @@ 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 storageFilters = storageFiltersFor(normalizedFilters, fsRelation,
readDataColumns,
+ (readDataColumns ++ generatedMetadataColumns).toStructType)
+ // The column the reader marks the rows it checked in, which the
post-scan Filter reads. See
+ // `FileFormat.buildReaderWithStorageFilters`. It is left out where it
alone would push the
+ // scan past the whole-stage codegen field limit, which would cost the
scan its columnar
+ // output. The post-scan Filter then evaluates the offered conjuncts on
every row.
+ val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
Review Comment:
Thanks. I'd keep the column there. Whether a file has an offset index is
known only when the reader opens it, and one scan can mix both kinds of files.
The cost is a load and a branch per row, next to the bloom's hash and probes.
The `columnindex` conf is known at planning, but it is an escape hatch, and
I'd rather not add a planner rule for it. The conf doc now says that in both
cases the post-scan filter evaluates the filter on each row of a row group with
a survivor.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -418,11 +441,19 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
Alias(KnownNotNull(CreateStruct(structColumns.toImmutableArraySeq)),
FileFormat.METADATA_NAME)(exprId = metadataStruct.exprId)
execution.ProjectExec(
- readDataColumns ++ partitionColumns :+ metadataAlias, scan)
+ readDataColumns ++ storageFilterCheckedColumn ++ partitionColumns :+
metadataAlias, scan)
}.getOrElse(scan)
+ // The offered conjuncts are skipped on the rows the reader marks as
checked. They are
+ // matched by `ExpressionSet`, since `storageFilters` carry the
relation's column names.
+ val postScanFilters = storageFilterCheckedColumn match {
+ case Some(checked) =>
+ val offered = ExpressionSet(storageFilters)
+ afterScanFilters.toSeq.map(f => if (offered.contains(f)) Or(checked,
f) else f)
Review Comment:
Thanks, done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4),
together with @sunchao's comment on this line. It is `IF(checked, true,
conjunct)` now.
--
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]
