fhueske commented on code in PR #28763:
URL: https://github.com/apache/flink/pull/28763#discussion_r3604156843
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java:
##########
@@ -123,22 +124,27 @@ public void onMatch(RelOptRuleCall call) {
"Could not resolve the TABLE input of the SNAPSHOT scan on
the build side of "
+ "a LATERAL SNAPSHOT join. This is a bug, please
file an issue.");
}
- // The build-side input must declare exactly one watermark, otherwise
the operator cannot
- // determine when the LOAD phase is complete.
- final long rowtimeCount =
- rawTableInput.getRowType().getFieldList().stream()
- .filter(f ->
FlinkTypeFactory.isRowtimeIndicatorType(f.getType()))
- .count();
- if (rowtimeCount == 0) {
- throw new ValidationException(
- "LATERAL SNAPSHOT requires a watermark on the build-side
input.");
- }
- if (rowtimeCount > 1) {
- throw new ValidationException(
- String.format(
- "The build-side input of a LATERAL SNAPSHOT join
must not have more than one "
- + "row-time attribute, but found %d.",
- rowtimeCount));
+ // The build-side row-time attribute drives the streaming operator's
LOAD phase. In batch
+ // all input is bounded and the join degrades to a regular join (see
+ // BatchPhysicalLateralSnapshotJoinRule), so no watermark is required.
+ if (!ShortcutUtils.unwrapContext(join).isBatchMode()) {
+ // The build-side input must declare exactly one watermark,
otherwise the operator
+ // cannot determine when the LOAD phase is complete.
+ final long rowtimeCount =
+ rawTableInput.getRowType().getFieldList().stream()
+ .filter(f ->
FlinkTypeFactory.isRowtimeIndicatorType(f.getType()))
+ .count();
+ if (rowtimeCount == 0) {
+ throw new ValidationException(
+ "LATERAL SNAPSHOT requires a watermark on the
build-side input.");
+ }
+ if (rowtimeCount > 1) {
+ throw new ValidationException(
+ String.format(
+ "The build-side input of a LATERAL SNAPSHOT
join must not have more than one "
+ + "row-time attribute, but found %d.",
+ rowtimeCount));
Review Comment:
The `LateralSnapshotJoinOperator` orders build-side records based on their
event-time attribute. If there would be more than one, this behavior could be
ambiguous.
We could also ask for a timestamp column in the SNAPSHOT function, but that
would make it harder to use IMO.
--
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]
