ForVic commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4406404238
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -142,17 +144,26 @@ object ParquetStorageFilter {
maxSplicedRowGroupBytes: Long): ParquetStorageFilter = {
require(boundExpressions.nonEmpty,
"storage filters must be non-empty; callers with nothing to push must
not call create")
- val expr = boundExpressions.reduce(And)
+ // The conjuncts over the blooms' key columns go under an `If` on the
blooms, so they run only
+ // on the rows the blooms keep. In one `And`, subexpression elimination
would compute what
+ // they share on every row.
+ val (blooms, onKeys) =
boundExpressions.partition(_.isInstanceOf[BloomFilterMightContain])
Review Comment:
`create` now puts `IsNotNull` first among the key conditions, as
`FilterExec` does. I added `an IsNotNull over the key columns is evaluated
before the other conjuncts over them`, which uses your probe
(`assert_not_null(k) > 0 AND isnotnull(k)` with odd keys null). It asserts the
same rows as a plain read, 999 excluded, and no row group given up. It fails
without the reordering.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -345,20 +359,23 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
// Offered conjuncts become `storageFilters` on the scan and stay in the
post-scan Filter too.
// 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 with it, as `FileSourceScanExec.supportsColumnar`
decides, so the column
// never changes how the scan runs. Above a row-based scan, the Project
that drops the
// column would copy every surviving row.
- val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ val checkedColumn =
AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
- }.filter { checked =>
+ val marksChecked = {
Review Comment:
Fixed. After the rebase onto SPARK-60465, `storageFiltersFor` returns the
offered conjuncts together with the checked column from
`StorageFilterPlanning.checkedColumn`. That builds nothing when no bloom is
offered, so a scan without storage filters no longer creates the attribute or
walks its fields. The post-scan Filter passes
`storageFilters.filterNot(_.isInstanceOf[IsNotNull])` to `skipCheckedRows`, so
the shared helper is unchanged.
--
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]
