peter-toth commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4219270146
##########
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:
Thanks. I'd rather go the other way. In
[437bdf3594b](https://github.com/apache/spark/pull/59269/commits/437bdf3594b3824468dfe9b712ee21aa8db0fdde)
the gate is back to one rule: the column is added only where the scan still
returns columnar batches with it, as `FileSourceScanExec.supportsColumnar`
decides. So it never changes how the scan runs. I took the `CodegenFallback`
term from the last round out again.
Above the scan I'd not predict the fusion. Where a stay-up `Filter`, a
`CodegenFallback` conjunct or a runtime fallback keeps the `Filter` out of
codegen, the `Project` copies the surviving rows once. I measured that at about
10% with 90 long columns and half the rows surviving, and nothing measurable
with 8 columns or 5% surviving. That seems not worth a growing mirror of
`CollapseCodegenStages`, as you say in comment 2.
##########
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:
Agreed that the mirror would keep growing. I'd rather keep the gate small
than add a rule, see my reply on comment 1. The gate now asks only what the
scan itself decides, and no longer predicts `CollapseCodegenStages`.
##########
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:
The copy is gone with the term in
[437bdf3594b](https://github.com/apache/spark/pull/59269/commits/437bdf3594b3824468dfe9b712ee21aa8db0fdde),
see my reply on comment 1.
##########
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:
Done in
[437bdf3594b](https://github.com/apache/spark/pull/59269/commits/437bdf3594b3824468dfe9b712ee21aa8db0fdde),
with your wording and the pointer.
##########
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:
Done in
[437bdf3594b](https://github.com/apache/spark/pull/59269/commits/437bdf3594b3824468dfe9b712ee21aa8db0fdde).
I reflowed the conf doc paragraph too.
--
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]
