{"record":{"id":"e4fd136a492ef745","repo":"apache/iceberg","slug":"creating-table-with-watermark-specs-is-not-support","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":"java.lang.UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L621-L657","documentation":"FlinkCatalog.validateFlinkTable rejects Flink schemas that declare watermark specifications, because watermark strategies are stream-processing concepts with no representation in Iceberg table metadata. Any non-empty getWatermarkSpecs() triggers UnsupportedOperationException.","triggerScenarios":"CREATE TABLE against an Iceberg catalog whose DDL includes `WATERMARK FOR rowtime_column AS watermark_strategy`; also hit via alterTable paths that call validateFlinkTable.","commonSituations":"Copying stream-table DDL (Kafka connector style with watermark clauses) to Iceberg; templates that append watermark declarations to time-attribute columns; developers conflating event-time handling with table schema.","solutions":["Remove the WATERMARK clause from the Iceberg table DDL.","Declare watermarks in the Flink streaming job or a VIEW over the table, not in the Iceberg catalog table definition.","Keep the watermark on the source-side table (e.g. Kafka connector table) and treat the Iceberg table as a plain sink."],"exampleFix":"// before\nCREATE TABLE t (ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND) WITH (...);\n// after\nCREATE TABLE t (ts TIMESTAMP(3));\n// declare watermark in a view or the source table","handlingStrategy":"validation","validationCode":"if (!table.getUnresolvedSchema().getWatermarkSpecs().isEmpty()) {\n  throw new IllegalArgumentException(\"Watermarks are not allowed in Iceberg table DDL\");\n}","typeGuard":"boolean hasNoWatermarks(org.apache.flink.table.api.Schema s) {\n  return s.getWatermarkSpecs().isEmpty();\n}","tryCatchPattern":"try { catalog.createTable(path, table); }\ncatch (UnsupportedOperationException e) { /* strip WATERMARK clause and retry */ }","preventionTips":["Never include WATERMARK clauses in Iceberg CREATE TABLE DDL","Declare watermarks in streaming jobs or views over the table","Keep watermark config on the source connector table, not the Iceberg sink"],"tags":["flink","watermark","schema-validation","unsupported-operation"],"backgroundTag":"schema-validation-failed","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"}