{"record":{"id":"a399e6c593525560","repo":"apache/iceberg","slug":"cannot-apply-unknown-table-change","errorCode":null,"errorMessage":"Cannot apply unknown table change: ","messagePattern":"Cannot apply unknown table change: ","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java","lineNumber":149,"sourceCode":"      } 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\n  private static void applyAddColumn(UpdateSchema pendingUpdate, TableChange.AddColumn addColumn) {\n    Column flinkColumn = addColumn.getColumn();\n    Preconditions.checkArgument(\n        FlinkCompatibilityUtil.isPhysicalColumn(flinkColumn),\n        \"Unsupported table change: Adding computed column %s.\",\n        flinkColumn.getName());\n\n    Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());\n\n    if (flinkColumn.getDataType().getLogicalType().isNullable()) {\n      pendingUpdate.addColumn(\n          flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));\n    } else {\n      pendingUpdate.addRequiredColumn(","sourceCodeStart":131,"sourceCodeEnd":167,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java#L131-L167","documentation":"The final else branch of applySchemaChanges rejects any TableChange type the util does not explicitly handle, preventing silent loss of schema updates. The thrown message includes the change's toString.","triggerScenarios":"Passing a TableChange subclass not covered by the if/else chain (e.g. a new Flink TableChange type added in a newer Flink version) into applySchemaChanges.","commonSituations":"Upgrading Flink and using newly introduced TableChange types with an older Iceberg connector; generic DDL-application frameworks forwarding all changes blindly.","solutions":["Upgrade iceberg-flink to a version handling the new change type","Filter out unsupported TableChange instances before calling applySchemaChanges","Apply the unhandled change through a different mechanism (raw Iceberg API, Spark)","Log and ignore the change if it is semantically a no-op for Iceberg"],"exampleFix":"// before\nFlinkAlterTableUtil.applySchemaChanges(table, changes);\n// after\nList<TableChange> supported = changes.stream().filter(c ->\n    c instanceof TableChange.AddColumn || c instanceof TableChange.DropColumn\n    || c instanceof TableChange.SetOption /* ... */).collect(Collectors.toList());\nFlinkAlterTableUtil.applySchemaChanges(table, supported);","handlingStrategy":"validation","validationCode":"Set<Class<?>> supported = Set.of(TableChange.AddColumn.class, TableChange.DropColumn.class,\n    TableChange.ModifyColumn.class, TableChange.AddUniqueConstraint.class,\n    TableChange.SetOption.class);\nboolean allSupported = changes.stream().allMatch(c ->\n    supported.stream().anyMatch(s -> s.isInstance(c)));","typeGuard":null,"tryCatchPattern":"try {\n  FlinkAlterTableUtil.applySchemaChanges(table, changes);\n} catch (UnsupportedOperationException e) {\n  // identify the offending change from the message and route it elsewhere\n}","preventionTips":["Whitelist known TableChange types before applying","Pin connector and Flink versions so available change types match","Handle unknown changes explicitly rather than forwarding blindly"],"tags":["flink","ddl","forward-compatibility"],"backgroundTag":"unsupported-operation","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}