twalthr commented on code in PR #29098:
URL: https://github.com/apache/flink/pull/29098#discussion_r3932058210
##########
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:
Feel free to make some of the SystemTypeInference methods public and
deduplicate code. Esp. for data type precision of a time attribute etc.
##########
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:
isn't this already covered by SystemTypeInference and can be removed? Looks
like duplicate code to me.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -240,6 +241,78 @@ void testBuildRowtimeIsNotForwarded() {
+ " GROUP BY TUMBLE(pts, INTERVAL '1'
MINUTE)");
}
+ @Test
+ void testBuildSideWatermarkOnHiddenMetadataColumn() {
Review Comment:
Move this to ColumnExpansionTest
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java:
##########
@@ -412,6 +443,22 @@ private static SourceTestStep appendBuild(List<Row> data) {
.build();
}
+ /**
+ * Insert-only build whose watermark is declared on a virtual metadata
column {@code rt}. The
+ * row-time value is supplied as a trailing metadata field of each
produced row.
+ */
+ private static SourceTestStep hiddenMetadataBuild(List<Row> data) {
+ return SourceTestStep.newBuilder("b")
+ .addSchema(
+ "bk STRING",
+ "bv INT",
+ "rt TIMESTAMP(3) METADATA VIRTUAL",
+ "WATERMARK FOR rt AS rt")
+ .addOption("readable-metadata", "rt:TIMESTAMP(3)")
+ .producedValues(data.toArray(new Row[0]))
+ .build();
+ }
Review Comment:
I would leave out the virtual metadata thing from these tests. They can be
declared in the ColumnExpansionTest. Orthogonal problem unrelated to the
LateralSnapshotJoin.
##########
docs/content.zh/docs/sql/reference/queries/joins.md:
##########
@@ -379,13 +380,14 @@ The `SNAPSHOT` function accepts the following arguments:
| Argument | Type | Required | Description
|
| --- | --- | ---
|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `input` | TABLE | yes | The build-side table. It may use any changelog mode
(inserts, updates, and deletes). In streaming mode it must declare a
[watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes"
>}}#event-time).
|
+| `input` | TABLE | yes | The build-side table. It may use any changelog mode
(inserts, updates, and deletes).
|
+| `on_time` | DESCRIPTOR | no | Declares a build-side row-time column that
defines the order in which the build-side changes are applied. The referenced
column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up
to precision 3) that is declared as a [watermarked row-time attribute]({{< ref
"docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The
argument is **required for streaming queries**. |
Review Comment:
```suggestion
| `on_time` | DESCRIPTOR | no | Declares a build-side rowtime column that
defines the order in which the build-side changes are applied. The referenced
column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up
to precision 3) that is declared as a [watermarked rowtime attribute]({{< ref
"docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The
argument is **required for streaming queries**. |
```
--
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]
