raminqaf commented on code in PR #29040:
URL: https://github.com/apache/flink/pull/29040#discussion_r3881284559
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java:
##########
@@ -285,7 +299,28 @@ public static RelNode convertMaterializedTableAsToRel(
null,
isOverwrite,
sink,
- null);
+ conflictStrategy);
+ }
+
+ /** Whether every table scanned by {@code rel} declares a watermark. */
+ private static boolean allSourcesHaveWatermarks(RelNode rel) {
+ if (rel instanceof TableScan) {
+ final TableSourceTable table =
+ ((TableScan)
rel).getTable().unwrap(TableSourceTable.class);
+ // An unresolvable scan can't be proven to lack a watermark, so it
isn't treated as a
+ // reason to fall back - matching the same convention used for the
physical-tree check.
+ return table == null
+ || !table.contextResolvedTable()
+ .getResolvedSchema()
+ .getWatermarkSpecs()
+ .isEmpty();
+ }
+ for (RelNode input : rel.getInputs()) {
+ if (!allSourcesHaveWatermarks(input)) {
+ return false;
+ }
+ }
+ return true;
}
Review Comment:
Also noticed that `collectSourcesWithoutWatermarks` does the same check
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java:
##########
@@ -285,7 +299,28 @@ public static RelNode convertMaterializedTableAsToRel(
null,
isOverwrite,
sink,
- null);
+ conflictStrategy);
+ }
+
+ /** Whether every table scanned by {@code rel} declares a watermark. */
+ private static boolean allSourcesHaveWatermarks(RelNode rel) {
+ if (rel instanceof TableScan) {
+ final TableSourceTable table =
+ ((TableScan)
rel).getTable().unwrap(TableSourceTable.class);
+ // An unresolvable scan can't be proven to lack a watermark, so it
isn't treated as a
+ // reason to fall back - matching the same convention used for the
physical-tree check.
+ return table == null
+ || !table.contextResolvedTable()
+ .getResolvedSchema()
+ .getWatermarkSpecs()
+ .isEmpty();
+ }
+ for (RelNode input : rel.getInputs()) {
+ if (!allSourcesHaveWatermarks(input)) {
+ return false;
+ }
+ }
+ return true;
}
Review Comment:
We can simplify this
```java
private static boolean allSourcesHaveWatermarks(RelNode rel) {
if (rel instanceof TableScan) {
final TableSourceTable table =
((TableScan) rel).getTable().unwrap(TableSourceTable.class);
return table == null
||
!table.contextResolvedTable().getResolvedSchema().getWatermarkSpecs().isEmpty();
}
return
rel.getInputs().stream().allMatch(DynamicSinkUtils::allSourcesHaveWatermarks);
}
```
--
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]
