{"record":{"id":"69c7489c24e382af","repo":"apache/iceberg","slug":"creating-table-with-watermark-specs-is-not-support-69c748","errorCode":null,"errorMessage":"Creating table with watermark specs is not supported yet.","messagePattern":"Creating table with watermark specs is not supported yet\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java","lineNumber":639,"sourceCode":"  }\n\n  private static void validateFlinkTable(CatalogBaseTable table) {\n    Preconditions.checkArgument(\n        table instanceof CatalogTable, \"The Table should be a CatalogTable.\");\n\n    org.apache.flink.table.api.Schema schema = table.getUnresolvedSchema();\n    schema\n        .getColumns()\n        .forEach(\n            column -> {\n              if (!FlinkCompatibilityUtil.isPhysicalColumn(column)) {\n                throw new UnsupportedOperationException(\n                    \"Creating table with computed columns is not supported yet.\");\n              }\n            });\n\n    if (!schema.getWatermarkSpecs().isEmpty()) {\n      throw new UnsupportedOperationException(\n          \"Creating table with watermark specs is not supported yet.\");\n    }\n  }\n\n  private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {\n    PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);\n    partitionKeys.forEach(builder::identity);\n    return builder.build();\n  }\n\n  private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {\n    ImmutableList.Builder<String> partitionKeysBuilder = ImmutableList.builder();\n    for (PartitionField field : spec.fields()) {\n      if (field.transform().isIdentity()) {\n        partitionKeysBuilder.add(icebergSchema.findColumnName(field.sourceId()));\n      } else {\n        // Not created by Flink SQL.\n        // For compatibility with iceberg tables, return empty.","sourceCodeStart":621,"sourceCodeEnd":657,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L621-L657","documentation":"Guard in FlinkCatalog.validateFlinkTable (reached from create/alter paths): the Flink table schema includes watermark specs, which Iceberg tables cannot store or enforce. Validation happens before any catalog mutation, so no partial table state is created.","triggerScenarios":"CREATE TABLE ... WATERMARK FOR rowtime_column AS <expression> processed through createIcebergTable or alterTable.","commonSituations":"Defining a source-table schema with event-time watermarks and pointing it at the Iceberg catalog; copying Kafka connector DDL (where WATERMARK is common) into Iceberg table creation; template DDL generators that always emit watermark clauses.","solutions":["Remove the WATERMARK clause from the CREATE TABLE statement; declare watermarks in the Flink job's TableEnvironment/ view instead.","Create the rowtime/watermark attributes in an intermediate Flink view over the Iceberg table used for streaming reads.","Keep the Iceberg DDL limited to physical columns."],"exampleFix":"// before\nCREATE TABLE t (id BIGINT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND);\n// after\nCREATE TABLE t (id BIGINT, ts TIMESTAMP(3));\n-- apply WATERMARK in a view: CREATE VIEW v AS SELECT * FROM t;\n-- then declare watermark on the view schema in the job","handlingStrategy":"validation","validationCode":"if (!table.getUnresolvedSchema().getWatermarkSpecs().isEmpty()) { /* strip watermarks before create */ }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never emit WATERMARK clauses in Iceberg CREATE TABLE","Declare watermarks in Flink views or the streaming job instead","Sanitize connector-style DDL templates before applying to Iceberg catalogs"],"tags":["flink","ddl","watermark","unsupported"],"backgroundTag":"operation-not-supported","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}