{"record":{"id":"51286bf9067575ae","repo":"apache/iceberg","slug":"creating-table-with-watermark-specs-is-not-support-51286b","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.2/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.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L621-L657","documentation":"FlinkCatalog.validateFlinkTable rejects table schemas that declare watermark specs, since Iceberg tables do not store Flink watermark definitions. If schema.getWatermarkSpecs() is non-empty, UnsupportedOperationException is thrown during createIcebergTable or alterTable.","triggerScenarios":"CREATE TABLE ... (`ts TIMESTAMP(3)`, WATERMARK FOR ts AS ts - INTERVAL '5' SECOND) with the iceberg connector; CREATE TABLE LIKE from a table with a WATERMARK; altering a table whose new schema contains watermark specs.","commonSituations":"Copying streaming source DDL (Kafka with event-time watermark) verbatim to an Iceberg sink; scaffolding generated by tools that include watermark clauses.","solutions":["Remove the WATERMARK FOR clause from the Iceberg DDL.","Declare the watermark in an intermediate Flink VIEW or in the source table, not on the Iceberg sink table.","Apply watermark logic in the job's Table API (via `.watermark()` on the source) rather than the catalog DDL."],"exampleFix":"// before\nCREATE TABLE t (ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND) WITH ('connector'='iceberg');\n\n// after\nCREATE TABLE t (ts TIMESTAMP(3)) WITH ('connector'='iceberg');","handlingStrategy":"validation","validationCode":"// reject watermark specs before CREATE TABLE\nif (!ddlSchema.getWatermarkSpecs().isEmpty()) {\n  // remove WATERMARK clauses; define them on the source/view instead\n}","typeGuard":null,"tryCatchPattern":"try {\n  catalog.createTable(tablePath, table, false);\n} catch (UnsupportedOperationException e) {\n  // strip watermark specs and retry\n}","preventionTips":["Never copy WATERMARK clauses from streaming-source DDL into Iceberg sink DDL","Define watermarks on source tables or in the job graph","Lint Iceberg DDL for the WATERMARK FOR keyword"],"tags":["flink","ddl","watermark"],"backgroundTag":"unsupported-operation","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"}