You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "linyanghao (via GitHub)" <gi...@apache.org> on 2023/05/17 06:38:59 UTC
[GitHub] [iceberg] linyanghao opened a new pull request, #7628: Flink: Support alter table column
linyanghao opened a new pull request, #7628:
URL: https://github.com/apache/iceberg/pull/7628
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264720082
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -605,6 +659,56 @@ private static void commitChanges(
transaction.commitTransaction();
}
+ private static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitChanges(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ FlinkAlterTableUtil.applyPropertyChanges(updateProperties, propertyChanges);
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ private static void commitChanges(Table table, String setSnapshotId, String pickSnapshotId) {
Review Comment:
Maybe `commitManageSnapshots`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] chenjunjiedada commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "chenjunjiedada (via GitHub)" <gi...@apache.org>.
chenjunjiedada commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1203364527
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +479,13 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
Review Comment:
That makes sense to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1735189816
@stevenzwu Hi, I rebased and addressed the style issues you mentioned. Could you check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1737050491
@stevenzwu Sure. Thanks for reviewing! Also thanks @chenjunjiedada and @hililiwei for reviewing!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1704961368
Hi @stevenzwu @hililiwei, do you have any more comments on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264720260
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
+ sql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ sql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ sql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ sql("ALTER TABLE tl RESET('oldK')");
+ properties.remove("oldK");
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
Review Comment:
Sure, will add more tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1262418547
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
+ sql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ sql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ sql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ sql("ALTER TABLE tl RESET('oldK')");
+ properties.remove("oldK");
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
Review Comment:
Also, add a test to remove a nonexistent field?
Similarly, we need more test cases to cover possible exceptions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1262403622
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
Review Comment:
nit: the previous exception message indicated the action to be taken, such as add, modify, but there is no. Should unify it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1334926298
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
Review Comment:
this (the extra whitespace in the end) needs to be addressed.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
Review Comment:
also the error msg doesn't follow the coding style. you can referring to `ArrowVectorAccessor` for examples. Please review all the error msgs in this class
E.g., it could be sth like `Unsupported schema change: adding watermark specs`
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,293 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail due to Iceberg's validation.
Review Comment:
oh. this is because `SchemaUpdate` by default doesn't allow incompatible change. can you add the note to the comment for easier understanding for future reader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1262417083
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
+ sql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ sql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ sql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ sql("ALTER TABLE tl RESET('oldK')");
+ properties.remove("oldK");
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testAlterTableAddColumn() {
Review Comment:
Do we need to add a case to test when the name of the field to be added already exists?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1556660051
For backward-compatibility, I'm currently remaining the old alterTable API as it is. Not sure if it's the right thing to do.
@hililiwei @stevenzwu @chenjunjiedada Could you take a look at this? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] chenjunjiedada commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "chenjunjiedada (via GitHub)" <gi...@apache.org>.
chenjunjiedada commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1200421450
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +479,13 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
Review Comment:
Should we deprecate this API?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1201604774
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +479,13 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
Review Comment:
We cannot deprecate this API from iceberg's side, since it's defined in flink repo. But we can log some warnings inside this method instead. Is that ok?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1736698932
thanks @linyanghao for the contribution and @chenjunjiedada for the review.
@linyanghao can you create a back port PR? you can follow the example here for back port: https://github.com/apache/iceberg/pull/8228
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1336932460
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,293 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail due to Iceberg's validation.
Review Comment:
Added to all assertions on validation failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu merged pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu merged PR #7628:
URL: https://github.com/apache/iceberg/pull/7628
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264718449
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -517,6 +532,61 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
}
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List<TableChange> tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ // Does not support altering partition yet.
+ validateTablePartition(toCatalogTable(icebergTable), (CatalogTable) newTable);
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ List<TableChange> propertyChanges = Lists.newArrayList();
+ List<TableChange> schemaChanges = Lists.newArrayList();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption set = (TableChange.SetOption) change;
+
+ if ("location".equalsIgnoreCase(set.getKey())) {
+ setLocation = set.getValue();
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ setSnapshotId = set.getValue();
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ pickSnapshotId = set.getValue();
Review Comment:
This was copied from the old alterTable method. It was originally `pickSnapshotId`. But `cherrypickSnapshotId` is of course better. Let me fix this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1715391451
@stevenzwu I addressed the comments. Regarding the warning message, I modified it so that users will only see it when they try altering schema using the old API. Could you take a look? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1737053175
@stevenzwu Wait, Flink does not support ALTER TABLE in 1.16 yet. I guess we don't need to backport?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264719352
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
Review Comment:
Thanks for catching this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1320859179
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ public static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ Map<String, String> setProperties) {
+ commitManageSnapshots(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!setProperties.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ setProperties.forEach(
+ (k, v) -> {
+ if (v == null) {
+ updateProperties.remove(k);
+ } else {
+ updateProperties.set(k, v);
+ }
+ });
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ public static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitManageSnapshots(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ FlinkAlterTableUtil.applyPropertyChanges(updateProperties, propertyChanges);
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ public static void commitManageSnapshots(
+ Table table, String setSnapshotId, String cherrypickSnapshotId) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is
+ // ambiguous and choosing
+ // one order leads to different results
Review Comment:
nit: fix the line break probably from google-java-format automation long time ago.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -514,7 +534,68 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
}
});
- commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ FlinkAlterTableUtil.commitChanges(
+ icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List<TableChange> tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ // Does not support altering partition yet.
+ validateTablePartition(toCatalogTable(icebergTable), (CatalogTable) newTable);
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String cherrypickSnapshotId = null;
+
+ List<TableChange> propertyChanges = Lists.newArrayList();
+ List<TableChange> schemaChanges = Lists.newArrayList();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption set = (TableChange.SetOption) change;
+
+ if ("location".equalsIgnoreCase(set.getKey())) {
+ setLocation = set.getValue();
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ setSnapshotId = set.getValue();
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ cherrypickSnapshotId = set.getValue();
+ } else {
+ propertyChanges.add(change);
+ }
+
+ } else if (change instanceof TableChange.ResetOption) {
+ propertyChanges.add(change);
+
Review Comment:
nit: empty line not needed. same for a couple other places in this method
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: add assertion on error msg
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING FIRST)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING AFTER id)");
+ Schema schemaAfterAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnComment() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING COMMENT 'some data')");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get(), "some data"))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableConstraint() {
+ sql("CREATE TABLE tl(id BIGINT NOT NULL, dt STRING NOT NULL, col1 STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+ Assert.assertEquals(ImmutableSet.of(), schemaBefore.identifierFieldNames());
+
+ sql("ALTER TABLE tl ADD (PRIMARY KEY (id) NOT ENFORCED)");
+ Schema schemaAfterAdd = table("tl").schema();
+ Assert.assertEquals(ImmutableSet.of("id"), schemaAfterAdd.identifierFieldNames());
+
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (dt) NOT ENFORCED)");
+ Schema schemaAfterModify = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterModify.asStruct());
+ Assert.assertEquals(ImmutableSet.of("dt"), schemaAfterModify.identifierFieldNames());
+
+ // Composite primary key
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, dt) NOT ENFORCED)");
+ Schema schemaAfterComposite = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterComposite.asStruct());
+ Assert.assertEquals(ImmutableSet.of("id", "dt"), schemaAfterComposite.identifierFieldNames());
+
+ // Setting an optional field as primary key should fail
+ Assertions.assertThatThrownBy(
+ () -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (col1) NOT ENFORCED)"))
+ .isInstanceOf(TableException.class);
+
+ // Setting a composite key containing an optional field should fail
+ Assertions.assertThatThrownBy(
+ () -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, col1) NOT ENFORCED)"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: assert error msg too
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: can we add assert on error msg too
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
Review Comment:
nit: nit: can we add assert on error msg too
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
Review Comment:
nit: can you comment which code perform the validation/check on adding a required field?
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
Review Comment:
nit: add assertion on error msg
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
Review Comment:
nit: add assertion on error msg
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
Review Comment:
can we add some failure tests? like non-exist column
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: add assertion on error msg
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING FIRST)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING AFTER id)");
+ Schema schemaAfterAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnComment() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING COMMENT 'some data')");
Review Comment:
nit: comment as `comment for dt field`?
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING FIRST)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING AFTER id)");
+ Schema schemaAfterAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnComment() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING COMMENT 'some data')");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get(), "some data"))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableConstraint() {
+ sql("CREATE TABLE tl(id BIGINT NOT NULL, dt STRING NOT NULL, col1 STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+ Assert.assertEquals(ImmutableSet.of(), schemaBefore.identifierFieldNames());
+
+ sql("ALTER TABLE tl ADD (PRIMARY KEY (id) NOT ENFORCED)");
+ Schema schemaAfterAdd = table("tl").schema();
+ Assert.assertEquals(ImmutableSet.of("id"), schemaAfterAdd.identifierFieldNames());
+
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (dt) NOT ENFORCED)");
+ Schema schemaAfterModify = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterModify.asStruct());
+ Assert.assertEquals(ImmutableSet.of("dt"), schemaAfterModify.identifierFieldNames());
+
+ // Composite primary key
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, dt) NOT ENFORCED)");
+ Schema schemaAfterComposite = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterComposite.asStruct());
+ Assert.assertEquals(ImmutableSet.of("id", "dt"), schemaAfterComposite.identifierFieldNames());
+
+ // Setting an optional field as primary key should fail
+ Assertions.assertThatThrownBy(
+ () -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (col1) NOT ENFORCED)"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: assert error msg too
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING FIRST)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(1, "id", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING AFTER id)");
+ Schema schemaAfterAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnComment() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl MODIFY (dt STRING COMMENT 'some data')");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get(), "some data"))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableConstraint() {
+ sql("CREATE TABLE tl(id BIGINT NOT NULL, dt STRING NOT NULL, col1 STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+ Assert.assertEquals(ImmutableSet.of(), schemaBefore.identifierFieldNames());
+
+ sql("ALTER TABLE tl ADD (PRIMARY KEY (id) NOT ENFORCED)");
+ Schema schemaAfterAdd = table("tl").schema();
+ Assert.assertEquals(ImmutableSet.of("id"), schemaAfterAdd.identifierFieldNames());
+
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (dt) NOT ENFORCED)");
+ Schema schemaAfterModify = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterModify.asStruct());
+ Assert.assertEquals(ImmutableSet.of("dt"), schemaAfterModify.identifierFieldNames());
+
+ // Composite primary key
+ sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, dt) NOT ENFORCED)");
+ Schema schemaAfterComposite = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()))
+ .asStruct(),
+ schemaAfterComposite.asStruct());
+ Assert.assertEquals(ImmutableSet.of("id", "dt"), schemaAfterComposite.identifierFieldNames());
+
+ // Setting an optional field as primary key should fail
+ Assertions.assertThatThrownBy(
+ () -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (col1) NOT ENFORCED)"))
+ .isInstanceOf(TableException.class);
+
+ // Setting a composite key containing an optional field should fail
+ Assertions.assertThatThrownBy(
+ () -> sql("ALTER TABLE tl MODIFY (PRIMARY KEY (id, col1) NOT ENFORCED)"))
+ .isInstanceOf(TableException.class);
+
+ // Dropping constraints is not supported yet
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP PRIMARY KEY"))
+ .isInstanceOf(TableException.class);
Review Comment:
nit: assert error msg too
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+
+ // To alter columns, use the other alterTable API and provide a list of TableChange's.
+ LOG.warn(
Review Comment:
but if this method is called for just changing table property, we will still see the warning msg. that can be confusing.
I am not sure why upstream Flink didn't mark this as deprecated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322714980
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
Review Comment:
I added comments on whether the failure is caused by Flink's validation or Iceberg's validation. Is that good?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1569752169
@chenjunjiedada Thanks for reviewing! I just fixed a bug on modifying the nullability of columns. You might want to check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322758470
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+
+ // To alter columns, use the other alterTable API and provide a list of TableChange's.
+ LOG.warn(
Review Comment:
Maybe the message should be included as part of the exception throw when detecting schema diff between the old and new CatalogTable object?
So that users will only see the message if they try to update schema using the old alterTable method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1737446047
if that's the case, yeah no need to back port
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1248843159
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+
+ // To alter columns, use the other alterTable API and provide a list of TableChange's.
+ LOG.warn(
Review Comment:
what is the purpose of this warning msg?
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.AddUniqueConstraint) {
+ TableChange.AddUniqueConstraint addPk = (TableChange.AddUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, addPk.getConstraint());
+
+ } else if (change instanceof TableChange.ModifyUniqueConstraint) {
+ TableChange.ModifyUniqueConstraint modifyPk = (TableChange.ModifyUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, modifyPk.getNewConstraint());
+
+ } else if (change instanceof TableChange.DropConstraint) {
+ throw new UnsupportedOperationException("Dropping constraints is not supported yet. ");
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateProperties} operation.
Review Comment:
nit: javadoc should be updated for property change
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
Review Comment:
nit: this empty line doesn't seem to following coding style. please check all places.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.AddUniqueConstraint) {
+ TableChange.AddUniqueConstraint addPk = (TableChange.AddUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, addPk.getConstraint());
+
+ } else if (change instanceof TableChange.ModifyUniqueConstraint) {
+ TableChange.ModifyUniqueConstraint modifyPk = (TableChange.ModifyUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, modifyPk.getNewConstraint());
+
+ } else if (change instanceof TableChange.DropConstraint) {
+ throw new UnsupportedOperationException("Dropping constraints is not supported yet. ");
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateProperties} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateProperty operation to configure
+ * @param propertyChanges a list of Flink table changes
+ */
+ public static void applyPropertyChanges(
+ UpdateProperties pendingUpdate, List<TableChange> propertyChanges) {
+ for (TableChange change : propertyChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption setOption = (TableChange.SetOption) change;
+ pendingUpdate.set(setOption.getKey(), setOption.getValue());
+
+ } else if (change instanceof TableChange.ResetOption) {
+ TableChange.ResetOption resetOption = (TableChange.ResetOption) change;
+ pendingUpdate.remove(resetOption.getKey());
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ private static void applyModifyColumn(
+ UpdateSchema pendingUpdate, TableChange.ModifyColumn modifyColumn) {
+ if (modifyColumn instanceof TableChange.ModifyColumnName) {
+ TableChange.ModifyColumnName modifyName = (TableChange.ModifyColumnName) modifyColumn;
+ pendingUpdate.renameColumn(modifyName.getOldColumnName(), modifyName.getNewColumnName());
+
+ } else if (modifyColumn instanceof TableChange.ModifyColumnPosition) {
+ TableChange.ModifyColumnPosition modifyPosition =
+ (TableChange.ModifyColumnPosition) modifyColumn;
+ applyModifyColumnPosition(pendingUpdate, modifyPosition);
+
+ } else if (modifyColumn instanceof TableChange.ModifyPhysicalColumnType) {
+ TableChange.ModifyPhysicalColumnType modifyType =
+ (TableChange.ModifyPhysicalColumnType) modifyColumn;
+ Type type = FlinkSchemaUtil.convert(modifyType.getNewType().getLogicalType());
+ String columnName = modifyType.getOldColumn().getName();
+ pendingUpdate.updateColumn(columnName, type.asPrimitiveType());
+ if (modifyType.getNewColumn().getDataType().getLogicalType().isNullable()) {
+ pendingUpdate.makeColumnOptional(columnName);
+ } else {
+ pendingUpdate.requireColumn(columnName);
+ }
+
+ } else if (modifyColumn instanceof TableChange.ModifyColumnComment) {
+ TableChange.ModifyColumnComment modifyComment =
+ (TableChange.ModifyColumnComment) modifyColumn;
+ pendingUpdate.updateColumnDoc(
+ modifyComment.getOldColumn().getName(), modifyComment.getNewComment());
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + modifyColumn);
+ }
+ }
+
+ private static void applyModifyColumnPosition(
+ UpdateSchema pendingUpdate, TableChange.ModifyColumnPosition modifyColumnPosition) {
+ TableChange.ColumnPosition newPosition = modifyColumnPosition.getNewPosition();
+ if (newPosition instanceof TableChange.First) {
+ pendingUpdate.moveFirst(modifyColumnPosition.getOldColumn().getName());
+
+ } else if (newPosition instanceof TableChange.After) {
+ TableChange.After after = (TableChange.After) newPosition;
+ pendingUpdate.moveAfter(modifyColumnPosition.getOldColumn().getName(), after.column());
+
+ } else {
+ throw new UnsupportedOperationException(
+ "Cannot apply unknown table change: " + modifyColumnPosition);
Review Comment:
nit: more specific error msg
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.AddUniqueConstraint) {
+ TableChange.AddUniqueConstraint addPk = (TableChange.AddUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, addPk.getConstraint());
+
+ } else if (change instanceof TableChange.ModifyUniqueConstraint) {
+ TableChange.ModifyUniqueConstraint modifyPk = (TableChange.ModifyUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, modifyPk.getNewConstraint());
+
+ } else if (change instanceof TableChange.DropConstraint) {
+ throw new UnsupportedOperationException("Dropping constraints is not supported yet. ");
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateProperties} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateProperty operation to configure
+ * @param propertyChanges a list of Flink table changes
+ */
+ public static void applyPropertyChanges(
+ UpdateProperties pendingUpdate, List<TableChange> propertyChanges) {
+ for (TableChange change : propertyChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption setOption = (TableChange.SetOption) change;
+ pendingUpdate.set(setOption.getKey(), setOption.getValue());
+
+ } else if (change instanceof TableChange.ResetOption) {
+ TableChange.ResetOption resetOption = (TableChange.ResetOption) change;
+ pendingUpdate.remove(resetOption.getKey());
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ private static void applyModifyColumn(
+ UpdateSchema pendingUpdate, TableChange.ModifyColumn modifyColumn) {
+ if (modifyColumn instanceof TableChange.ModifyColumnName) {
+ TableChange.ModifyColumnName modifyName = (TableChange.ModifyColumnName) modifyColumn;
+ pendingUpdate.renameColumn(modifyName.getOldColumnName(), modifyName.getNewColumnName());
+
+ } else if (modifyColumn instanceof TableChange.ModifyColumnPosition) {
+ TableChange.ModifyColumnPosition modifyPosition =
+ (TableChange.ModifyColumnPosition) modifyColumn;
+ applyModifyColumnPosition(pendingUpdate, modifyPosition);
+
+ } else if (modifyColumn instanceof TableChange.ModifyPhysicalColumnType) {
+ TableChange.ModifyPhysicalColumnType modifyType =
+ (TableChange.ModifyPhysicalColumnType) modifyColumn;
+ Type type = FlinkSchemaUtil.convert(modifyType.getNewType().getLogicalType());
+ String columnName = modifyType.getOldColumn().getName();
+ pendingUpdate.updateColumn(columnName, type.asPrimitiveType());
+ if (modifyType.getNewColumn().getDataType().getLogicalType().isNullable()) {
+ pendingUpdate.makeColumnOptional(columnName);
+ } else {
+ pendingUpdate.requireColumn(columnName);
+ }
+
+ } else if (modifyColumn instanceof TableChange.ModifyColumnComment) {
+ TableChange.ModifyColumnComment modifyComment =
+ (TableChange.ModifyColumnComment) modifyColumn;
+ pendingUpdate.updateColumnDoc(
+ modifyComment.getOldColumn().getName(), modifyComment.getNewComment());
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + modifyColumn);
Review Comment:
nit: more specific error msg
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -605,6 +659,56 @@ private static void commitChanges(
transaction.commitTransaction();
}
+ private static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitChanges(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ FlinkAlterTableUtil.applyPropertyChanges(updateProperties, propertyChanges);
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ private static void commitChanges(Table table, String setSnapshotId, String pickSnapshotId) {
Review Comment:
this method name is a little misleading and confusing. maybe `applyManageSnapshots` would be more intuitive.
also curious why this method doesn't live inside `FlinkAlterTableUtil`?
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
Review Comment:
nit: fix the line break from auto formatting. this can be combined with the other comment of moving to method Javadoc
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
Review Comment:
nit: empty space in the end is not necessary. please check all places
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -605,6 +659,56 @@ private static void commitChanges(
transaction.commitTransaction();
}
+ private static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitChanges(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ FlinkAlterTableUtil.applyPropertyChanges(updateProperties, propertyChanges);
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ private static void commitChanges(Table table, String setSnapshotId, String pickSnapshotId) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is
+ // ambiguous and choosing
Review Comment:
nit: fix line break
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateSchema} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param schemaChanges a list of Flink table changes
+ */
+ public static void applySchemaChanges(
+ 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),
+ "Adding computed columns is not supported yet: %s",
+ flinkColumn.getName());
+ Type icebergType = FlinkSchemaUtil.convert(flinkColumn.getDataType().getLogicalType());
+ pendingUpdate.addColumn(flinkColumn.getName(), icebergType);
+
+ } else if (change instanceof TableChange.ModifyColumn) {
+ TableChange.ModifyColumn modifyColumn = (TableChange.ModifyColumn) change;
+ applyModifyColumn(pendingUpdate, modifyColumn);
+
+ } else if (change instanceof TableChange.DropColumn) {
+ TableChange.DropColumn dropColumn = (TableChange.DropColumn) change;
+ pendingUpdate.deleteColumn(dropColumn.getColumnName());
+
+ } else if (change instanceof TableChange.AddWatermark) {
+ throw new UnsupportedOperationException("Adding watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.ModifyWatermark) {
+ throw new UnsupportedOperationException("Modifying watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.DropWatermark) {
+ throw new UnsupportedOperationException("Watermark specs is not supported yet. ");
+
+ } else if (change instanceof TableChange.AddUniqueConstraint) {
+ TableChange.AddUniqueConstraint addPk = (TableChange.AddUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, addPk.getConstraint());
+
+ } else if (change instanceof TableChange.ModifyUniqueConstraint) {
+ TableChange.ModifyUniqueConstraint modifyPk = (TableChange.ModifyUniqueConstraint) change;
+ applyUniqueConstraint(pendingUpdate, modifyPk.getNewConstraint());
+
+ } else if (change instanceof TableChange.DropConstraint) {
+ throw new UnsupportedOperationException("Dropping constraints is not supported yet. ");
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
+ }
+ }
+ }
+
+ /**
+ * Applies a list of Flink table changes to an {@link UpdateProperties} operation.
+ *
+ * @param pendingUpdate an uncommitted UpdateProperty operation to configure
+ * @param propertyChanges a list of Flink table changes
+ */
+ public static void applyPropertyChanges(
+ UpdateProperties pendingUpdate, List<TableChange> propertyChanges) {
+ for (TableChange change : propertyChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption setOption = (TableChange.SetOption) change;
+ pendingUpdate.set(setOption.getKey(), setOption.getValue());
+
+ } else if (change instanceof TableChange.ResetOption) {
+ TableChange.ResetOption resetOption = (TableChange.ResetOption) change;
+ pendingUpdate.remove(resetOption.getKey());
+
+ } else {
+ throw new UnsupportedOperationException("Cannot apply unknown table change: " + change);
Review Comment:
nit: error msg can be more specific to property changes.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
Review Comment:
also this comment probably should be moved as method Javadoc
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -605,6 +659,56 @@ private static void commitChanges(
transaction.commitTransaction();
}
+ private static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitChanges(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
Review Comment:
maybe `commit` action can be moved inside `FlinkAlterTableUtil`? then it would be more natural to move `applyManageSnapshots` into `FlinkAlterTableUtil` too as mentioned in a comment below.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -517,6 +532,61 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
}
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List<TableChange> tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ // Does not support altering partition yet.
+ validateTablePartition(toCatalogTable(icebergTable), (CatalogTable) newTable);
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ List<TableChange> propertyChanges = Lists.newArrayList();
+ List<TableChange> schemaChanges = Lists.newArrayList();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption set = (TableChange.SetOption) change;
+
+ if ("location".equalsIgnoreCase(set.getKey())) {
+ setLocation = set.getValue();
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ setSnapshotId = set.getValue();
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ pickSnapshotId = set.getValue();
Review Comment:
nit: `cherrypickSnapshotId` would be easier to understand
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1262414288
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
Review Comment:
This method is very similar to testAlterTable(), they're doing the same thing, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1732222090
@linyanghao I left a few comments. can you also rebase?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1262418547
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
+ sql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ sql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ sql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ sql("ALTER TABLE tl RESET('oldK')");
+ properties.remove("oldK");
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
Review Comment:
Also, add a test to remove a nonexistent field?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1266317835
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
Review Comment:
yes, I think so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264717731
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+
+ // To alter columns, use the other alterTable API and provide a list of TableChange's.
+ LOG.warn(
Review Comment:
The user calling this method will pass a CatalogBaseTable argument, which represents how the
user expect the table should be like after altering it. So I think it's better to warn the user, that the schema part of the CatalogBaseTable argument won't be effective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322712913
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -322,39 +323,265 @@ public void testAlterTable() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
// remove property
- CatalogTable catalogTable = catalogTable("tl");
+ sql("ALTER TABLE tl RESET('oldK')");
properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
Assert.assertEquals(properties, table("tl").properties());
}
@Test
- public void testAlterTableWithPrimaryKey() throws TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT, PRIMARY KEY(id) NOT ENFORCED) WITH ('oldK'='oldV')");
- Map<String, String> properties = Maps.newHashMap();
- properties.put("oldK", "oldV");
+ public void testAlterTableAddColumn() {
+ sql("CREATE TABLE tl(id BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaBefore.asStruct());
- // new
- sql("ALTER TABLE tl SET('newK'='newV')");
- properties.put("newK", "newV");
- Assert.assertEquals(properties, table("tl").properties());
+ sql("ALTER TABLE tl ADD (dt STRING)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
- // update old
- sql("ALTER TABLE tl SET('oldK'='oldV2')");
- properties.put("oldK", "oldV2");
- Assert.assertEquals(properties, table("tl").properties());
+ // Add multiple columns
+ sql("ALTER TABLE tl ADD (col1 STRING, col2 BIGINT)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter2.asStruct());
- // remove property
- CatalogTable catalogTable = catalogTable("tl");
- properties.remove("oldK");
- getTableEnv()
- .getCatalog(getTableEnv().getCurrentCatalog())
- .get()
- .alterTable(new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
- Assert.assertEquals(properties, table("tl").properties());
+ // Adding a required field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (pk STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Adding an existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl ADD (id STRING)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING, col1 STRING, col2 BIGINT)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl DROP (dt)");
+ Schema schemaAfter1 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "col1", Types.StringType.get()),
+ Types.NestedField.optional(4, "col2", Types.LongType.get()))
+ .asStruct(),
+ schemaAfter1.asStruct());
+
+ // Drop multiple columns
+ sql("ALTER TABLE tl DROP (col1, col2)");
+ Schema schemaAfter2 = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ schemaAfter2.asStruct());
+
+ // Dropping an non-existing field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (foo)"))
+ .isInstanceOf(ValidationException.class);
+
+ // Dropping an already-deleted field should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl DROP (dt)"))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnName() {
+ sql("CREATE TABLE tl(id BIGINT, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ sql("ALTER TABLE tl RENAME dt TO data");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnType() {
+ sql("CREATE TABLE tl(id INTEGER, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Promote type from Integer to Long
+ sql("ALTER TABLE tl MODIFY (id BIGINT)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+
+ // Type change that doesn't follow the type-promotion rule should fail
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt INTEGER)"))
+ .isInstanceOf(TableException.class);
+ }
+
+ @Test
+ public void testAlterTableModifyColumnNullability() {
+ sql("CREATE TABLE tl(id INTEGER NOT NULL, dt STRING)");
+ Schema schemaBefore = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaBefore.asStruct());
+
+ // Cannot change nullability from optional to required
+ Assertions.assertThatThrownBy(() -> sql("ALTER TABLE tl MODIFY (dt STRING NOT NULL)"))
+ .isInstanceOf(TableException.class);
+
+ // Set nullability from required to optional
+ sql("ALTER TABLE tl MODIFY (id INTEGER)");
+ Schema schemaAfter = table("tl").schema();
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get()))
+ .asStruct(),
+ schemaAfter.asStruct());
+ }
+
+ @Test
+ public void testAlterTableModifyColumnPosition() {
Review Comment:
Added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#issuecomment-1647313261
@stevenzwu @hililiwei Thanks for reviewing! I added a few patches. Could you take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264719063
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -605,6 +659,56 @@ private static void commitChanges(
transaction.commitTransaction();
}
+ private static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitChanges(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
Review Comment:
That's reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1264720526
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -357,6 +358,213 @@ public void testAlterTableWithPrimaryKey() throws TableNotExistException {
Assert.assertEquals(properties, table("tl").properties());
}
+ @Test
+ public void testAlterTableProperties() throws TableNotExistException {
Review Comment:
Yes, maybe I should remove the original testAlterTable()?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1201592463
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +479,13 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
Review Comment:
It seems that flink is not deprecating the old alterTable API yet. This is to support some legacy operations: `AlterTableSchemaOperation`, `AlterViewPropertiesOperation` and `AlterViewAsOperation`.
Among these operations:
- `AlterTableSchemaOperation` is now [only used to alter hive tables](https://github.com/apache/flink/blob/553c72a894b25c10e71e54e6567c187876234fd5/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java#L133).
- `AlterViewPropertiesOperation` and `AlterViewAsOperation` are for views.
So I guess it's safe to assume the old alterTable API won't be called by flink when altering iceberg tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322718278
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/util/FlinkAlterTableUtil.java:
##########
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.util;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+
+public class FlinkAlterTableUtil {
+ private FlinkAlterTableUtil() {}
+
+ public static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ Map<String, String> setProperties) {
+ commitManageSnapshots(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!setProperties.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ setProperties.forEach(
+ (k, v) -> {
+ if (v == null) {
+ updateProperties.remove(k);
+ } else {
+ updateProperties.set(k, v);
+ }
+ });
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ public static void commitChanges(
+ Table table,
+ String setLocation,
+ String setSnapshotId,
+ String pickSnapshotId,
+ List<TableChange> schemaChanges,
+ List<TableChange> propertyChanges) {
+ commitManageSnapshots(table, setSnapshotId, pickSnapshotId);
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation().setLocation(setLocation).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ UpdateSchema updateSchema = transaction.updateSchema();
+ FlinkAlterTableUtil.applySchemaChanges(updateSchema, schemaChanges);
+ updateSchema.commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ FlinkAlterTableUtil.applyPropertyChanges(updateProperties, propertyChanges);
+ updateProperties.commit();
+ }
+
+ transaction.commitTransaction();
+ }
+
+ public static void commitManageSnapshots(
+ Table table, String setSnapshotId, String cherrypickSnapshotId) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is
+ // ambiguous and choosing
+ // one order leads to different results
Review Comment:
I put "ambiguous and choosing" and "one order leads to different results" into one line. Did you mean this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322716448
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -514,7 +534,68 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
}
});
- commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ FlinkAlterTableUtil.commitChanges(
+ icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List<TableChange> tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ // Does not support altering partition yet.
+ validateTablePartition(toCatalogTable(icebergTable), (CatalogTable) newTable);
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String cherrypickSnapshotId = null;
+
+ List<TableChange> propertyChanges = Lists.newArrayList();
+ List<TableChange> schemaChanges = Lists.newArrayList();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption set = (TableChange.SetOption) change;
+
+ if ("location".equalsIgnoreCase(set.getKey())) {
+ setLocation = set.getValue();
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ setSnapshotId = set.getValue();
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ cherrypickSnapshotId = set.getValue();
+ } else {
+ propertyChanges.add(change);
+ }
+
+ } else if (change instanceof TableChange.ResetOption) {
+ propertyChanges.add(change);
+
Review Comment:
Fixed. Sorry for missing this out earlier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] linyanghao commented on a diff in pull request #7628: Flink 1.17: Support alter table column
Posted by "linyanghao (via GitHub)" <gi...@apache.org>.
linyanghao commented on code in PR #7628:
URL: https://github.com/apache/iceberg/pull/7628#discussion_r1322771918
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -472,11 +482,16 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
CatalogTable table = toCatalogTable(icebergTable);
- // Currently, Flink SQL only support altering table properties.
+ // This alterTable API only supports altering table properties.
- // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by
+ // Support for adding/removing/renaming columns cannot be done by
// comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+
+ // To alter columns, use the other alterTable API and provide a list of TableChange's.
+ LOG.warn(
Review Comment:
Sorry for missing out this comment earlier. I moved the message into the exception. You might want to check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org