yihua commented on code in PR #19914:
URL: https://github.com/apache/hudi/pull/19914#discussion_r4203390343
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala:
##########
@@ -349,6 +360,34 @@ private[parquet] object HoodieFileGroupReaderFunction {
iter.map(mapper.apply(_))
}
+ /**
+ * Evaluates `requiredFilters` on every row. Pushed-down filters only prune
row groups in the vectorized reader,
+ * and the plan-level filter for them is only added when the Hudi session
extension is registered, so the
+ * commit-time range of an incremental query is enforced here. The scan
returns rows rather than batches
+ * whenever required filters are present.
+ */
+ private def applyRequiredFilters(iter: Iterator[InternalRow],
Review Comment:
non-blocking: With the session extension registered,
`maybeApplyForNewFileFormat` already wraps the scan in a plan-level Filter on
these same required filters, so on that path every COW incremental row is now
evaluated twice, and the `requiredFilters.isEmpty` gate also gives up batch
output where the plan Filter would have handled it vectorized. Could the format
skip the predicate (and keep returning batches) when the plan-side filter is in
place, e.g. a flag set alongside `isProjected` in `applyNewFileFormatChanges`
and `createStreamingDataFrame`, rather than detecting the extension via conf,
which misses programmatic registration? Separately, if the predicate stays, the
V2 `In` over all included commit times should go through `InSet` (or an
equivalent set lookup), since `Predicate.create` on a raw `In` generates a
linear chain of string compares per row.
--
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]
