fhueske commented on code in PR #29098:
URL: https://github.com/apache/flink/pull/29098#discussion_r3933044565
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java:
##########
@@ -376,15 +360,61 @@ private static RelNode replaceSnapshotScan(RelNode node) {
return null;
}
+ /**
+ * Resolves the build-side row-time column named by the {@code on_time}
argument to its field
+ * index in {@code buildInputNode}. Streaming requires the argument and
the referenced column to
+ * be a row-time attribute; batch does not use a row-time attribute and
returns {@code -1} when
+ * the argument is absent.
+ */
+ private static int resolveOnTimeIndex(
+ RexCall snapshotCall, RelNode buildInputNode, boolean isBatch) {
+ final List<RexNode> operands = snapshotCall.getOperands();
+ final int argIndex = LateralSnapshotTypeStrategy.ON_TIME_ARG_INDEX;
+ final RexNode timeColumnArg = argIndex < operands.size() ?
operands.get(argIndex) : null;
+ if (timeColumnArg == null || timeColumnArg.isA(SqlKind.DEFAULT)) {
+ if (!isBatch) {
+ throw new ValidationException(
+ "LATERAL SNAPSHOT requires the 'on_time' argument to
identify the "
+ + "build-side row-time attribute.");
+ }
+ return -1;
+ }
+ if (!(timeColumnArg instanceof RexCall)
+ || !(((RexCall) timeColumnArg).getOperator() instanceof
SqlDescriptorOperator)) {
+ throw new ValidationException("Argument 'on_time' of SNAPSHOT must
be a DESCRIPTOR.");
+ }
+ final List<RexNode> descriptorOperands = ((RexCall)
timeColumnArg).getOperands();
+ if (descriptorOperands.size() != 1 || !(descriptorOperands.get(0)
instanceof RexLiteral)) {
+ throw new ValidationException(
+ "Argument 'on_time' of SNAPSHOT must reference exactly one
column.");
+ }
+ final String timeColName = RexLiteral.stringValue((RexLiteral)
descriptorOperands.get(0));
+ final int timeColIdx =
buildInputNode.getRowType().getFieldNames().indexOf(timeColName);
+ if (timeColIdx < 0) {
+ throw new ValidationException(
+ String.format(
+ "Argument 'on_time' of SNAPSHOT references column
'%s' which is not "
+ + "present in the input table.",
Review Comment:
Yes, good point. We can remove some of the checks
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java:
##########
@@ -233,6 +250,50 @@ private static Optional<List<DataType>> validateInputs(
return Optional.of(callContext.getArgumentDataTypes());
}
+ /**
+ * Validates {@code on_time} when present: it must name exactly one
existing TIMESTAMP or
+ * TIMESTAMP_LTZ column (precision up to 3). Returns {@code null} when the
argument is absent or
+ * valid; otherwise returns the failure result of {@link CallContext#fail}.
+ */
+ private static Optional<List<DataType>> validateOnTime(
Review Comment:
good idea, thanks!
--
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]
