You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by hu...@apache.org on 2026/05/21 00:25:13 UTC
(iceberg) branch main updated: Flink: Backport add column fix to flink v1.20, v2.0 (#16447)
This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 90c2c4c539 Flink: Backport add column fix to flink v1.20, v2.0 (#16447)
90c2c4c539 is described below
commit 90c2c4c539095b066c683483792a750cef37c592
Author: Stepan Stepanishchev <11...@users.noreply.github.com>
AuthorDate: Thu May 21 07:25:07 2026 +0700
Flink: Backport add column fix to flink v1.20, v2.0 (#16447)
---
.../iceberg/flink/util/FlinkAlterTableUtil.java | 40 ++++++++++++++--------
.../iceberg/flink/TestFlinkCatalogTable.java | 25 ++++++++++++++
.../iceberg/flink/util/FlinkAlterTableUtil.java | 40 ++++++++++++++--------
.../iceberg/flink/TestFlinkCatalogTable.java | 25 ++++++++++++++
4 files changed, 102 insertions(+), 28 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
index 2bbc9cf208..35ada6106f 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
@@ -124,20 +124,7 @@ public class FlinkAlterTableUtil {
UpdateSchema pendingUpdate, List<TableChange> schemaChanges) {
for (TableChange change : schemaChanges) {
if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- Column flinkColumn = addColumn.getColumn();
- Preconditions.checkArgument(
- FlinkCompatibilityUtil.isPhysicalColumn(flinkColumn),
- "Unsupported table change: Adding computed column %s.",
- flinkColumn.getName());
- Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
- if (flinkColumn.getDataType().getLogicalType().isNullable()) {
- pendingUpdate.addColumn(
- flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
- } else {
- pendingUpdate.addRequiredColumn(
- flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
- }
+ applyAddColumn(pendingUpdate, (TableChange.AddColumn) change);
} else if (change instanceof TableChange.ModifyColumn) {
TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
applyModifyColumn(pendingUpdate, modifyColumn);
@@ -164,6 +151,31 @@ public class FlinkAlterTableUtil {
}
}
+ private static void applyAddColumn(UpdateSchema pendingUpdate, TableChange.AddColumn addColumn) {
+ Column flinkColumn = addColumn.getColumn();
+ Preconditions.checkArgument(
+ FlinkCompatibilityUtil.isPhysicalColumn(flinkColumn),
+ "Unsupported table change: Adding computed column %s.",
+ flinkColumn.getName());
+
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+
+ if (flinkColumn.getDataType().getLogicalType().isNullable()) {
+ pendingUpdate.addColumn(
+ flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
+ } else {
+ pendingUpdate.addRequiredColumn(
+ flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
+ }
+
+ if (addColumn.getPosition() != null) {
+ TableChange.ColumnPosition position = addColumn.getPosition();
+ TableChange.ModifyColumnPosition modifyColumnPosition =
+ new TableChange.ModifyColumnPosition(addColumn.getColumn(), position);
+ applyModifyColumnPosition(pendingUpdate, modifyColumnPosition);
+ }
+ }
+
/**
* Applies a list of Flink table property changes to an {@link UpdateProperties} operation.
*
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index 2cfee87e46..757f22857b 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -397,6 +397,31 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
.hasMessageContaining("Try to add a column `id` which already exists in the table.");
}
+ @TestTemplate
+ public void testAlterTableAddColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, name STRING)");
+ Schema schemaBefore = table("tl").schema();
+ assertThat(schemaBefore.asStruct())
+ .isEqualTo(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()))
+ .asStruct());
+
+ sql("ALTER TABLE tl ADD (col1 STRING FIRST)");
+ sql("ALTER TABLE tl ADD (col2 INT AFTER id)");
+
+ Schema schemaAfter = table("tl").schema();
+ assertThat(schemaAfter.asStruct())
+ .isEqualTo(
+ new Schema(
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(4, "col2", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()))
+ .asStruct());
+ }
+
@TestTemplate
public void testAlterTableDropColumn() {
sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
index 2bbc9cf208..35ada6106f 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java
@@ -124,20 +124,7 @@ public class FlinkAlterTableUtil {
UpdateSchema pendingUpdate, List<TableChange> schemaChanges) {
for (TableChange change : schemaChanges) {
if (change instanceof TableChange.AddColumn) {
- TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
- Column flinkColumn = addColumn.getColumn();
- Preconditions.checkArgument(
- FlinkCompatibilityUtil.isPhysicalColumn(flinkColumn),
- "Unsupported table change: Adding computed column %s.",
- flinkColumn.getName());
- Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
- if (flinkColumn.getDataType().getLogicalType().isNullable()) {
- pendingUpdate.addColumn(
- flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
- } else {
- pendingUpdate.addRequiredColumn(
- flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
- }
+ applyAddColumn(pendingUpdate, (TableChange.AddColumn) change);
} else if (change instanceof TableChange.ModifyColumn) {
TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
applyModifyColumn(pendingUpdate, modifyColumn);
@@ -164,6 +151,31 @@ public class FlinkAlterTableUtil {
}
}
+ private static void applyAddColumn(UpdateSchema pendingUpdate, TableChange.AddColumn addColumn) {
+ Column flinkColumn = addColumn.getColumn();
+ Preconditions.checkArgument(
+ FlinkCompatibilityUtil.isPhysicalColumn(flinkColumn),
+ "Unsupported table change: Adding computed column %s.",
+ flinkColumn.getName());
+
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+
+ if (flinkColumn.getDataType().getLogicalType().isNullable()) {
+ pendingUpdate.addColumn(
+ flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
+ } else {
+ pendingUpdate.addRequiredColumn(
+ flinkColumn.getName(), icebergType, flinkColumn.getComment().orElse(null));
+ }
+
+ if (addColumn.getPosition() != null) {
+ TableChange.ColumnPosition position = addColumn.getPosition();
+ TableChange.ModifyColumnPosition modifyColumnPosition =
+ new TableChange.ModifyColumnPosition(addColumn.getColumn(), position);
+ applyModifyColumnPosition(pendingUpdate, modifyColumnPosition);
+ }
+ }
+
/**
* Applies a list of Flink table property changes to an {@link UpdateProperties} operation.
*
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index f7848a5d22..091a7b67b4 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -404,6 +404,31 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
.hasMessageContaining("Try to add a column `id` which already exists in the table.");
}
+ @TestTemplate
+ public void testAlterTableAddColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, name STRING)");
+ Schema schemaBefore = table("tl").schema();
+ assertThat(schemaBefore.asStruct())
+ .isEqualTo(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()))
+ .asStruct());
+
+ sql("ALTER TABLE tl ADD (col1 STRING FIRST)");
+ sql("ALTER TABLE tl ADD (col2 INT AFTER id)");
+
+ Schema schemaAfter = table("tl").schema();
+ assertThat(schemaAfter.asStruct())
+ .isEqualTo(
+ new Schema(
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(4, "col2", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()))
+ .asStruct());
+ }
+
@TestTemplate
public void testAlterTableDropColumn() {
sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");