{"record":{"id":"898f642ad324b97c","repo":"apache/iceberg","slug":"unsupported-table-change-addwatermark-898f64","errorCode":null,"errorMessage":"Unsupported table change: AddWatermark.","messagePattern":"Unsupported table change: AddWatermark\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java","lineNumber":135,"sourceCode":"  /**\n   * Applies a list of Flink table changes to an {@link UpdateSchema} operation.\n   *\n   * @param pendingUpdate an uncommitted UpdateSchema operation to configure\n   * @param schemaChanges a list of Flink table changes\n   */\n  public static void applySchemaChanges(\n      UpdateSchema pendingUpdate, List<TableChange> schemaChanges) {\n    for (TableChange change : schemaChanges) {\n      if (change instanceof TableChange.AddColumn) {\n        applyAddColumn(pendingUpdate, (TableChange.AddColumn) change);\n      } else if (change instanceof TableChange.ModifyColumn) {\n        TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;\n        applyModifyColumn(pendingUpdate, modifyColumn);\n      } else if (change instanceof TableChange.DropColumn) {\n        TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;\n        pendingUpdate.deleteColumn(dropColumn.getColumnName());\n      } else if (change instanceof TableChange.AddWatermark) {\n        throw new UnsupportedOperationException(\"Unsupported table change: AddWatermark.\");\n      } else if (change instanceof TableChange.ModifyWatermark) {\n        throw new UnsupportedOperationException(\"Unsupported table change: ModifyWatermark.\");\n      } else if (change instanceof TableChange.DropWatermark) {\n        throw new UnsupportedOperationException(\"Unsupported table change: DropWatermark.\");\n      } else if (change instanceof TableChange.AddUniqueConstraint) {\n        TableChange.AddUniqueConstraint addPk = (TableChange.AddUniqueConstraint) change;\n        applyUniqueConstraint(pendingUpdate, addPk.getConstraint());\n      } else if (change instanceof TableChange.ModifyUniqueConstraint) {\n        TableChange.ModifyUniqueConstraint modifyPk = (TableChange.ModifyUniqueConstraint) change;\n        applyUniqueConstraint(pendingUpdate, modifyPk.getNewConstraint());\n      } else if (change instanceof TableChange.DropConstraint) {\n        throw new UnsupportedOperationException(\"Unsupported table change: DropConstraint.\");\n      } else {\n        throw new UnsupportedOperationException(\"Cannot apply unknown table change: \" + change);\n      }\n    }\n  }\n","sourceCodeStart":117,"sourceCodeEnd":153,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java#L117-L153","documentation":"FlinkAlterTableUtil.applySchemaChanges maps Flink TableChange objects onto Iceberg UpdateSchema operations. Iceberg tables do not support watermarks, so an AddWatermark change is rejected with UnsupportedOperationException. The change is never applied and the ALTER fails.","triggerScenarios":"Executing a Flink SQL 'ALTER TABLE ... ADD WATERMARK' (or calling TableEnvironment/DynamicCatalog with a TableChange.AddWatermark) against a table backed by the Iceberg catalog.","commonSituations":"Porting SQL written for Flink's built-in connector tables (which support watermarks) to Iceberg tables.","solutions":["Remove the ADD WATERMARK statement; express event-time logic in the Flink job's SELECT instead (watermark on the source/derived table view)","Define the watermark on a Flink VIEW over the Iceberg table rather than on the Iceberg table itself","Patch via custom catalog if watermark semantics are truly needed (not supported upstream)"],"exampleFix":"-- before\nALTER TABLE iceberg_table ADD WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND;\n-- after\nCREATE VIEW v AS SELECT *, rowtime - INTERVAL '5' SECOND AS wm FROM iceberg_table; -- watermark on view/source","handlingStrategy":"try-catch","validationCode":"// before ALTER\nif (changes.stream().anyMatch(c -> c instanceof TableChange.AddWatermark)) {\n  throw new IllegalArgumentException(\"Watermark DDL is not supported on Iceberg tables\");\n}","typeGuard":"boolean isWatermarkChange(TableChange c) { return c instanceof TableChange.AddWatermark || c instanceof TableChange.ModifyWatermark || c instanceof TableChange.DropWatermark; }","tryCatchPattern":"try {\n  FlinkAlterTableUtil.applySchemaChanges(update, changes);\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage().contains(\"AddWatermark\")) { /* move watermark to a view or source */ }\n  throw e;\n}","preventionTips":["Never issue watermark DDL on Iceberg tables; put watermarks on Flink views or sources","Filter out watermark TableChange types before forwarding changes to the Iceberg catalog","Document watermark limitations in team DDL guidelines"],"tags":["flink","schema-change","watermark","unsupported"],"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"}