hudi-agent commented on code in PR #19948:
URL: https://github.com/apache/hudi/pull/19948#discussion_r4056085063
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala:
##########
@@ -85,6 +87,7 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
mergeType = mergeType,
fileSplits = fileSplits,
includedInstantTimeSet =
Option(includedCommits.map(_.requestedTime).toSet),
+ instantRangeOpt = queryContext.getInstantRange,
Review Comment:
🤖 This passes the range on the non-full-scan path too, and on this legacy
RDD path `requiredFilters` is `Seq.empty`, so before this change base rows and
pre-range log blocks were merged first and only then post-filtered by
`includedInstantTimeSet`. With EVENT_TIME_ORDERING (or a partial-update
payload), an in-range log record with a lower ordering value than the
out-of-range base row used to be suppressed by the merge but will now be
emitted on its own — is that the intended incremental semantics here, or should
the range be limited to the `fullTableScan` case on this path?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala:
##########
@@ -45,6 +46,7 @@ import scala.collection.immutable
trait MergeOnReadIncrementalRelation {
def listFileSplits(partitionFilters: Seq[Expression], dataFilters:
Seq[Expression]): Map[InternalRow, Seq[FileSlice]]
def getRequiredFilters: Seq[Filter]
+ def getInstantRange: HOption[InstantRange] = HOption.empty()
Review Comment:
🤖 nit: giving `getInstantRange` a default `HOption.empty()` in the trait
means any other implementation (e.g. V1) silently opts out of range filtering.
Have you considered leaving it abstract like `getRequiredFilters`, so each
relation makes the choice explicit?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java:
##########
@@ -429,6 +430,9 @@ protected List<PartitionPath>
listPartitionPaths(List<String> relativePartitionP
}
private boolean isBeforeTimelineStarts() {
+ if
(START_COMMIT_EARLIEST.equalsIgnoreCase(incrementalQueryStartTime.get())) {
Review Comment:
🤖 nit: a one-line comment on why the earliest marker short-circuits here
(it's not a real timestamp, so the completion/requested-time comparisons below
would be meaningless) would help — could also be pulled into a small
`isStartCommitEarliest(...)` helper next to `START_COMMIT_EARLIEST` in
`IncrementalQueryAnalyzer` if that check is repeated elsewhere.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
