peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4084294794
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -421,11 +552,313 @@ public boolean nextBatch() throws IOException {
return true;
}
+ /**
+ * Splicing emit path. Closes the previous emit's dequeued key vectors
(releasing survivor memory
+ * incrementally), advances to the next row group if needed via {@link
#checkEndOfRowGroup()},
+ * dequeues one survivor key vector per key column, drives non-key column
readers for {@code num}
+ * rows, and assembles a fresh {@link ColumnarBatch} interleaving key
(dequeued) and non-key
+ * (persistent value-vector) slots in the original projection order.
+ * The per-emit batch is a transient view over vectors owned elsewhere; see
{@link #close()} and
+ * {@link #closeSplicingState()}.
+ */
+ private boolean nextBatchSplicing() throws IOException {
Review Comment:
Fair question, and it is now answered in the description under **Design
decisions** as well as in the reader's own note.
Re-reading is where this started, and it was measured slower: on TPCDS q37,
3949 ms splicing against 4511 ms re-reading, with the feature off at 3199 ms.
The 562 ms between the two is one more read of `cs_item_sk`, and nothing
absorbs it on object storage -- parquet caches footers rather than pages, and
S3A's default input stream caches nothing, so it is a new GET (the caching
stream types are opt-in).
Two corrections to the cost model. The re-read does not remove the `LIMIT`
stall, since phase 1 still walks the whole row group to build `finalRanges`.
And it would not remove the buffering either, for the same reason -- what it
removes is the splicing machinery, which is real but is the part with tests
around it.
Your comment did change the planner, though. q37 is an all-keys projection,
and on that shape neither design can win: the reader has to read the key column
to evaluate the filter on it, so the scan reads the same columns for the same
rows as a plain one. That is what 3199 against 3949 shows. Extraction is now
dropped when every projected data column is a key column of the filter.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1885,6 +1885,21 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED =
+ buildConf("spark.sql.parquet.storageFilterPushdown.enabled")
+ .doc("If true, allows the vectorized Parquet reader to evaluate runtime
storage filters " +
+ "(e.g. bloom filters from join runtime filtering) at the scan level
using late " +
+ "materialization: read key columns first, evaluate the filter per row,
then read data " +
+ "columns restricted to surviving rows. This is a planning-time
decision only: when " +
+ "false, no storage filter is attached to a scan in the first place and
the filter is " +
+ "applied as an ordinary post-scan filter instead. Note that the
surviving key values of " +
+ "a whole row group are buffered before the first batch of that row
group is produced, " +
+ "so a task holds up to one extra copy of the key columns for one row
group.")
+ .version("5.0.0")
Review Comment:
Master-only on purpose, so `5.0.0` is right. The reader needs the parquet
1.18 APIs -- `RowRanges` public in `filter2.columnindex`,
`RowRanges.builder()`, `addSelectedRow(long)` and a public
`ParquetFileReader.getRowRanges(int)` -- and that upgrade (#58120) went to
master alone, with `branch-4.x` still on 1.17.0.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -151,6 +152,63 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
+ /**
+ * Splits `afterScanFilters` into bloom-filter conjuncts that can be pushed
to the storage layer
+ * for late materialization (returned as the first element) and the
remaining filters that stay as
+ * a post-scan FilterExec (returned as the second element).
+ *
+ * Eligibility (statically checked here so the runtime never silently loses
the filter):
+ * - The storage-filter pushdown SQL conf is on.
+ * - The file format is exactly [[ParquetFileFormat]]. Subclasses are
excluded on purpose: they
+ * may customize reading by overriding `buildReaderWithPartitionValues`,
and attaching storage
+ * filters would route the scan through `ParquetFileFormat`'s own reader
instead, silently
+ * dropping whatever the subclass does.
+ * - The vectorized reader is feasible for the schema the reader will
actually see, i.e.
+ * `partitionSchema ++ outputDataSchema` -- the same schema
`ParquetFileFormat.buildReader`
+ * derives `enableVectorizedReader` from.
+ * - The conjunct is a top-level [[BloomFilterMightContain]] (not nested
under OR/NOT).
+ * - The conjunct is deterministic. `ParquetStorageFilter.test` evaluates
the predicate without
+ * calling `BasePredicate.initialize(partitionIndex)`, which
`GeneratePredicate` emits for a
+ * `Nondeterministic` expression, so a non-deterministic conjunct would
fail at task time. No
+ * such bloom exists today, because the only producer is
`InjectRuntimeFilter` and a join key
+ * is deterministic, but this gate should not depend on a distant rule.
+ * - The bloom's value-side references are projected data columns whose
type the reader's value
+ * copier supports (see [[ParquetStorageFilter.isSupportedKeyType]]).
+ *
+ * If any condition fails, ALL bloom filters stay in the second element to
preserve the existing
+ * fallback behavior.
+ */
+ private def extractStorageFilters(
+ afterScanFilters: ExpressionSet,
+ fsRelation: HadoopFsRelation,
+ readDataColumns: Seq[Attribute],
+ outputDataSchema: StructType): (Seq[Expression], ExpressionSet) = {
+ val sparkSession = fsRelation.sparkSession
+ val sqlConf = sparkSession.sessionState.conf
+ if (!sqlConf.parquetStorageFilterPushdownEnabled) return (Nil,
afterScanFilters)
+ if (fsRelation.fileFormat.getClass != classOf[ParquetFileFormat]) {
Review Comment:
Done, as `FileFormat.supportsStorageFilter(expr)` defaulting to false.
`FileSourceStrategy` no longer imports the parquet package, and the key-type
list lives next to the copier that defines it. The subclass exclusion moved
into `ParquetFileFormat`'s override, where it reads as a fact about that
reader, exactly as you put it.
One deviation: the signature takes only the expression, not the read schema.
Everything the format needs is reachable from the expression's references, and
schema-level feasibility is `supportBatch`, which the planner asks a few lines
above. It can grow the parameter when a second format needs one.
--
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]
