[flink] Support the watermark function of auditlog table

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/SystemCatalogTable.java

@@ -19,17 +19,24 @@
 package org.apache.paimon.flink;
 
 import org.apache.paimon.table.Table;
+import org.apache.paimon.table.system.AuditLogTable;
 
 import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.WatermarkSpec;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.types.utils.TypeConversions;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
+import static org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK;
+import static org.apache.flink.table.descriptors.Schema.SCHEMA;
 import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
+import static org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.compoundKey;
+import static org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.deserializeWatermarkSpec;
 
 /** A {@link CatalogTable} to represent system table. */
 public class SystemCatalogTable implements CatalogTable {
@@ -46,10 +53,21 @@ public Table table() {
 
     @Override
     public Schema getUnresolvedSchema() {
-        return Schema.newBuilder()
-                .fromRowDataType(
-                        TypeConversions.fromLogicalToDataType(toLogicalType(table.rowType())))
-                .build();
+        Schema.Builder builder = Schema.newBuilder();
+        builder.fromRowDataType(
+                TypeConversions.fromLogicalToDataType(toLogicalType(table.rowType())));
+        if (table instanceof AuditLogTable) {
+            Map<String, String> newOptions = new HashMap<>(table.options());
+            if (newOptions.keySet().stream()
+                    .anyMatch(key -> key.startsWith(compoundKey(SCHEMA, WATERMARK)))) {
+                WatermarkSpec watermarkSpec = deserializeWatermarkSpec(newOptions);
+                return builder.watermark(
+                                watermarkSpec.getRowtimeAttribute(),
+                                watermarkSpec.getWatermarkExpr())
+                        .build();
+            }
+        }
+        return builder.build();
     }
 
     @Override

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/WatermarkITCase.java

@@ -41,6 +41,39 @@ public void testWatermark() throws Exception {
         innerTestWatermark();
     }
 
+    @Test
+    public void testAuditLogWatermark() throws Exception {
+        String[] options =
+                new String[] {
+                    "'scan.watermark.idle-timeout'='1s'",
+                    "'scan.watermark.alignment.group'='group'",
+                    "'scan.watermark.alignment.update-interval'='2s'",
+                    "'scan.watermark.alignment.max-drift'='1s'"
+                };
+        sql(
+                "CREATE TABLE T (f0 INT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts) WITH ("
+                        + String.join(",", options)
+                        + ")");
+
+        BlockingIterator<Row, Row> select =
+                BlockingIterator.of(
+                        streamSqlIter(
+                                "SELECT window_start, window_end, SUM(f0) FROM TABLE("
+                                        + "TUMBLE(TABLE T$audit_log, DESCRIPTOR(ts), INTERVAL '10' MINUTES))\n"
+                                        + "  GROUP BY window_start, window_end;"));
+
+        sql("INSERT INTO T VALUES (1, TIMESTAMP '2023-02-02 12:00:00')");
+        sql("INSERT INTO T VALUES (1, TIMESTAMP '2023-02-02 12:10:05')");
+
+        assertThat(select.collect(1))
+                .containsExactlyInAnyOrder(
+                        Row.of(
+                                LocalDateTime.parse("2023-02-02T12:00"),
+                                LocalDateTime.parse("2023-02-02T12:10"),
+                                1));
+        select.close();
+    }
+
     @Disabled // TODO unstable alignment may block watermark generation
     @Test
     public void testWatermarkAlignment() throws Exception {