dongjoon-hyun commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4219028686
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +358,41 @@ 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
whole-stage codegen
+ // fuses the Filter and the Project that drops the column, over a scan
that still returns
+ // columnar batches with it. Elsewhere that Project would copy every
surviving row. A
+ // conjunct that falls back from codegen keeps the Filter out, as
`CollapseCodegenStages`
+ // decides.
+ 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
+ val filterFallsBack = afterScanFilters.exists(_.exists {
Review Comment:
**1. The gate does not look at the stay-up `Filter`s**
`filterFallsBack` checks `afterScanFilters` only. `finalFilters` (L466) also
stacks one `FilterExec` per entry of `stayUpFilters` between this `Filter` and
the `Project` that drops the checked column, and nothing checks whether those
are fused.
`ScanOperation` puts a filter there when it follows a nondeterministic one,
e.g. `reflect(...)`, or a higher-order function after a `rand()` filter. When
such a `FilterExec` holds a `CodegenFallback`:
- `CollapseCodegenStages` refuses it, so the stage ends at the lower
`Filter`;
- the `ProjectExec` that drops the column gets a stage of its own over an
`InputAdapter`, and copies every surviving row;
- when `projects == withFilter.output`, that `Project` would not exist
without the column.
An `InjectRuntimeFilter` bloom rarely ends up in this shape today, since it
lands above the user's nondeterministic filters and is then not offered at all.
But the comment above says the column is added only where the `Filter` and the
`Project` are fused, and the gate test has no stay-up case. Could the check
also cover `stayUpFilters`, with a case in the test?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +358,41 @@ 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
whole-stage codegen
+ // fuses the Filter and the Project that drops the column, over a scan
that still returns
+ // columnar batches with it. Elsewhere that Project would copy every
surviving row. A
+ // conjunct that falls back from codegen keeps the Filter out, as
`CollapseCodegenStages`
+ // decides.
+ 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
+ val filterFallsBack = afterScanFilters.exists(_.exists {
+ case _: LeafExpression => false
+ case e => e.isInstanceOf[CodegenFallback]
+ })
+ conf.wholeStageEnabled && !filterFallsBack &&
Review Comment:
**2. The gate keeps re-deriving `CollapseCodegenStages` one term at a time**
Each round has added one more term of `CollapseCodegenStages.supportCodegen`
to this strategy-time prediction: `wholeStageEnabled`, `isTooManyFields`, and
now `CodegenFallback`. It still misses the operators between the scan and the
`Project` (comment 1), and any condition added to `supportCodegen` later will
be missed in the same way.
The decision is made here before the `Filter` and `Project` exist, so it can
only predict what `CollapseCodegenStages` will do with them. Could a physical
preparation rule placed right before `CollapseCodegenStages` decide instead? It
would ask `supportCodegen` about the actual `Filter`/`Project` chain above each
scan that has the checked column. Where the chain cannot be fused, it would
drop the column from the scan's output and `requiredSchema` and unwrap
`IF(checked, true, f)` back to `f`. That would be one rule that always agrees
with the fusion, instead of a mirror that grows each round. If that is too much
for this PR, a TODO or a JIRA for it would also help.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +358,41 @@ 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
whole-stage codegen
+ // fuses the Filter and the Project that drops the column, over a scan
that still returns
+ // columnar batches with it. Elsewhere that Project would copy every
surviving row. A
+ // conjunct that falls back from codegen keeps the Filter out, as
`CollapseCodegenStages`
+ // decides.
+ 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
+ val filterFallsBack = afterScanFilters.exists(_.exists {
+ case _: LeafExpression => false
Review Comment:
**3. This copies `CollapseCodegenStages.supportCodegen(e: Expression)`**
`case _: LeafExpression => false; case e => e.isInstanceOf[CodegenFallback]`
is the inverse of the private rule at `WholeStageCodegenExec.scala` L918-923.
If that rule changes which expressions block fusion, this copy drifts without
any failure. That happens if it lets a leaf `CodegenFallback` through, or if it
adds another marker.
How about moving the expression rule into `object WholeStageCodegenExec`,
next to `isTooManyFields`, which this gate already calls, and using it in both
places?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -177,6 +177,20 @@ trait FileFormat {
* leaves every one of them in the post-scan `Filter` as well, so ignoring
one is a missed
* optimization rather than a wrong answer.
*
+ * The `Filter` can skip them on the rows the reader says it has checked.
The planner asks for
+ * that only where whole-stage codegen fuses the `Filter` with a scan that
returns columnar
+ * batches. `requiredSchema`
+ * then ends with a nullable boolean column named
+ * [[FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME]], which is not a column
of the relation, and
+ * the `Filter` evaluates each storage filter as `IF(checked, true, filter)`.
+ * - True says every storage filter was evaluated on the row and kept it.
+ * - False or null says the row was not checked, and the `Filter` decides
it as if the column
+ * were not there. So a reader that does not know the column, and reads
it as missing from
+ * the file, is still right.
+ * A reader that marks rows writes the column itself and never reads it from
the file. The column
Review Comment:
**4. The contract still says the column is never read from the file**
"never reads it from the file" no longer matches the code or the narrowed
wording elsewhere:
- the reader javadoc now says only that a primitive one "is still never
decoded";
- the file's chunk of that name is still requested and fetched
(`requestedColumns`/`requestedPaths`, and the base read of a declined file), as
you said in the reply on comment 4 of the previous round, which I am not
raising again;
- a nested one fails `initialize`, which is why
`STORAGE_FILTER_CHECKED_COLUMN_NAME` now says such a file is not supported.
How about "writes the column itself rather than decoding it from the file",
with a pointer to the not-supported note on
`STORAGE_FILTER_CHECKED_COLUMN_NAME`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -177,6 +177,20 @@ trait FileFormat {
* leaves every one of them in the post-scan `Filter` as well, so ignoring
one is a missed
* optimization rather than a wrong answer.
*
+ * The `Filter` can skip them on the rows the reader says it has checked.
The planner asks for
+ * that only where whole-stage codegen fuses the `Filter` with a scan that
returns columnar
+ * batches. `requiredSchema`
+ * then ends with a nullable boolean column named
Review Comment:
**5. Nit: ragged reflow**
The edit left a short line here. How about:
```suggestion
* batches. `requiredSchema` then ends with a nullable boolean column named
```
The conf doc has the same breaks, at `SQLConf.scala` L1988-1991 (`"named
_tmp_storage_filter_checked. " +`, `"one outside the table's schema is not
supported. A row group where " +`, `"the reader stops applying the filter reads
" +`). Reflowing that paragraph would keep later diffs on it smaller.
--
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]
