FANNG1 commented on code in PR #4097:
URL: https://github.com/apache/gravitino/pull/4097#discussion_r1673577860
##########
flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -462,6 +524,119 @@ private Column
toGravitinoColumn(org.apache.flink.table.catalog.Column column) {
null);
}
+ private List<TableChange> optionsChanges(
+ Map<String, String> currentOptions, Map<String, String> updatedOptions) {
+ List<TableChange> optionsChanges =
com.google.common.collect.Lists.newArrayList();
+ MapDifference<String, String> difference = Maps.difference(currentOptions,
updatedOptions);
+ difference
+ .entriesOnlyOnLeft()
+ .forEach((key, value) ->
optionsChanges.add(TableChange.removeProperty(key)));
+ difference
+ .entriesOnlyOnRight()
+ .forEach((key, value) ->
optionsChanges.add(TableChange.setProperty(key, value)));
+ difference
+ .entriesDiffering()
+ .forEach(
+ (key, value) -> {
+ optionsChanges.add(TableChange.setProperty(key,
value.rightValue()));
+ });
+ return optionsChanges;
+ }
+
+ private void removeProperty(
+ org.apache.flink.table.catalog.TableChange.ResetOption change,
List<TableChange> changes) {
+ changes.add(TableChange.removeProperty(change.getKey()));
+ }
+
+ private void setProperty(
+ org.apache.flink.table.catalog.TableChange.SetOption change,
List<TableChange> changes) {
+ changes.add(TableChange.setProperty(change.getKey(), change.getValue()));
+ }
+
+ private void dropColumn(
+ org.apache.flink.table.catalog.TableChange.DropColumn change,
List<TableChange> changes) {
+ changes.add(TableChange.deleteColumn(new String[]
{change.getColumnName()}, true));
+ }
+
+ private void addColumn(
+ org.apache.flink.table.catalog.TableChange.AddColumn change,
List<TableChange> changes) {
+ changes.add(
+ TableChange.addColumn(
+ new String[] {change.getColumn().getName()},
+
TypeUtils.toGravitinoType(change.getColumn().getDataType().getLogicalType()),
+ change.getColumn().getComment().orElse(null),
+ TableUtils.toGravitinoColumnPosition(change.getPosition())));
+ }
+
+ private void modifyColumn(
+ org.apache.flink.table.catalog.TableChange change, List<TableChange>
changes) {
+ if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnName) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnName
modifyColumnName =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnName) change;
+ changes.add(
+ TableChange.renameColumn(
+ new String[] {modifyColumnName.getOldColumnName()},
+ modifyColumnName.getNewColumnName()));
+ } else if (change
+ instanceof
org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType) {
+ org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType
modifyColumnType =
+
(org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType) change;
+ changes.add(
+ TableChange.updateColumnType(
+ new String[] {modifyColumnType.getOldColumn().getName()},
+
TypeUtils.toGravitinoType(modifyColumnType.getNewType().getLogicalType())));
+ } else if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnPosition) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnPosition
modifyColumnPosition =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnPosition)
change;
+ changes.add(
+ TableChange.updateColumnPosition(
+ new String[] {modifyColumnPosition.getOldColumn().getName()},
+
TableUtils.toGravitinoColumnPosition(modifyColumnPosition.getNewPosition())));
+ } else if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnComment) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnComment
modifyColumnComment =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnComment)
change;
+ changes.add(
+ TableChange.updateColumnComment(
+ new String[] {modifyColumnComment.getOldColumn().getName()},
+ modifyColumnComment.getNewComment()));
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Not support ModifyColumn : %s", change.getClass()));
+ }
+ }
+
+ @VisibleForTesting
+ TableChange[] getTableChanges(CatalogBaseTable existingTable,
CatalogBaseTable newTable) {
+ List<TableChange> changes = Lists.newArrayList();
+ changes.addAll(optionsChanges(existingTable.getOptions(),
newTable.getOptions()));
+ if (!Objects.equals(newTable.getComment(), existingTable.getComment())) {
Review Comment:
pre check newTable.getComment != null?
##########
flink-connector/src/test/java/com/datastrato/gravitino/flink/connector/integration/test/FlinkCommonIT.java:
##########
@@ -330,4 +331,214 @@ public void testGetSimpleTable() {
},
true);
}
+
+ @Test
+ public void testRenameColumn() {
+ String databaseName = "test_renam_column_db";
+ String tableName = "test_rename_column";
+ doWithSchema(
+ currentCatalog(),
+ databaseName,
+ catalog -> {
+ TableResult result =
+ sql(
+ "CREATE TABLE %s "
+ + "(user_id INT COMMENT 'USER_ID', "
+ + " order_amount DOUBLE COMMENT 'ORDER_AMOUNT')"
+ + " COMMENT 'test comment'",
+ tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+
+ result = sql("ALTER TABLE %s RENAME user_id TO user_id_new",
tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+
+ Column[] actual =
+ catalog
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(databaseName, tableName))
+ .columns();
+ Column[] expected =
+ new Column[] {
+ Column.of("user_id_new", Types.IntegerType.get(), "USER_ID"),
+ Column.of("order_amount", Types.DoubleType.get(),
"ORDER_AMOUNT")
+ };
+ assertColumns(expected, actual);
+ },
+ true);
+ }
+
+ @Test
+ public void testAlterTableAddColumn() {
+ String databaseName = "test_alter_table_add_column_db";
+ String tableName = "test_alter_table_add_column";
+ doWithSchema(
+ currentCatalog(),
+ databaseName,
+ catalog -> {
+ TableResult result =
+ sql(
+ "CREATE TABLE %s "
+ + "(user_id INT COMMENT 'USER_ID', "
+ + " order_amount INT COMMENT 'ORDER_AMOUNT')"
+ + " COMMENT 'test comment'",
+ tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+ result = sql("ALTER TABLE %s ADD new_column_2 INT AFTER
order_amount", tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+
+ Column[] actual =
+ catalog
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(databaseName, tableName))
+ .columns();
+ Column[] expected =
+ new Column[] {
+ Column.of("user_id", Types.IntegerType.get(), "USER_ID"),
+ Column.of("order_amount", Types.IntegerType.get(),
"ORDER_AMOUNT"),
+ Column.of("new_column_2", Types.IntegerType.get(), null),
+ };
+ assertColumns(expected, actual);
+ },
+ true);
+ }
+
+ @Test
+ public void testAlterTableDropColumn() {
+ String databaseName = "test_alter_table_drop_column_db";
+ String tableName = "test_alter_table_drop_column";
+ doWithSchema(
+ currentCatalog(),
+ databaseName,
+ catalog -> {
+ TableResult result =
+ sql(
+ "CREATE TABLE %s "
+ + "(user_id INT COMMENT 'USER_ID', "
+ + " order_amount INT COMMENT 'ORDER_AMOUNT')"
+ + " COMMENT 'test comment'",
+ tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+ result = sql("ALTER TABLE %s DROP user_id", tableName);
+ TestUtils.assertTableResult(result, ResultKind.SUCCESS);
+ Column[] actual =
+ catalog
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(databaseName, tableName))
+ .columns();
+ Column[] expected =
+ new Column[] {Column.of("order_amount", Types.IntegerType.get(),
"ORDER_AMOUNT")};
+ assertColumns(expected, actual);
+ },
+ true);
+ }
+
+ @Test
+ public void testAlterTableAlterColumn() {
Review Comment:
`AlterColumn` is too general, could you change another name to be more
specific?
##########
flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -462,6 +524,119 @@ private Column
toGravitinoColumn(org.apache.flink.table.catalog.Column column) {
null);
}
+ private List<TableChange> optionsChanges(
+ Map<String, String> currentOptions, Map<String, String> updatedOptions) {
+ List<TableChange> optionsChanges =
com.google.common.collect.Lists.newArrayList();
+ MapDifference<String, String> difference = Maps.difference(currentOptions,
updatedOptions);
+ difference
+ .entriesOnlyOnLeft()
+ .forEach((key, value) ->
optionsChanges.add(TableChange.removeProperty(key)));
+ difference
+ .entriesOnlyOnRight()
+ .forEach((key, value) ->
optionsChanges.add(TableChange.setProperty(key, value)));
+ difference
+ .entriesDiffering()
+ .forEach(
+ (key, value) -> {
+ optionsChanges.add(TableChange.setProperty(key,
value.rightValue()));
+ });
+ return optionsChanges;
+ }
+
+ private void removeProperty(
+ org.apache.flink.table.catalog.TableChange.ResetOption change,
List<TableChange> changes) {
+ changes.add(TableChange.removeProperty(change.getKey()));
+ }
+
+ private void setProperty(
+ org.apache.flink.table.catalog.TableChange.SetOption change,
List<TableChange> changes) {
+ changes.add(TableChange.setProperty(change.getKey(), change.getValue()));
+ }
+
+ private void dropColumn(
+ org.apache.flink.table.catalog.TableChange.DropColumn change,
List<TableChange> changes) {
+ changes.add(TableChange.deleteColumn(new String[]
{change.getColumnName()}, true));
+ }
+
+ private void addColumn(
+ org.apache.flink.table.catalog.TableChange.AddColumn change,
List<TableChange> changes) {
+ changes.add(
+ TableChange.addColumn(
+ new String[] {change.getColumn().getName()},
+
TypeUtils.toGravitinoType(change.getColumn().getDataType().getLogicalType()),
+ change.getColumn().getComment().orElse(null),
+ TableUtils.toGravitinoColumnPosition(change.getPosition())));
+ }
+
+ private void modifyColumn(
+ org.apache.flink.table.catalog.TableChange change, List<TableChange>
changes) {
+ if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnName) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnName
modifyColumnName =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnName) change;
+ changes.add(
+ TableChange.renameColumn(
+ new String[] {modifyColumnName.getOldColumnName()},
+ modifyColumnName.getNewColumnName()));
+ } else if (change
+ instanceof
org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType) {
+ org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType
modifyColumnType =
+
(org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType) change;
+ changes.add(
+ TableChange.updateColumnType(
+ new String[] {modifyColumnType.getOldColumn().getName()},
+
TypeUtils.toGravitinoType(modifyColumnType.getNewType().getLogicalType())));
+ } else if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnPosition) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnPosition
modifyColumnPosition =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnPosition)
change;
+ changes.add(
+ TableChange.updateColumnPosition(
+ new String[] {modifyColumnPosition.getOldColumn().getName()},
+
TableUtils.toGravitinoColumnPosition(modifyColumnPosition.getNewPosition())));
+ } else if (change instanceof
org.apache.flink.table.catalog.TableChange.ModifyColumnComment) {
+ org.apache.flink.table.catalog.TableChange.ModifyColumnComment
modifyColumnComment =
+ (org.apache.flink.table.catalog.TableChange.ModifyColumnComment)
change;
+ changes.add(
+ TableChange.updateColumnComment(
+ new String[] {modifyColumnComment.getOldColumn().getName()},
+ modifyColumnComment.getNewComment()));
+ } else {
+ throw new IllegalArgumentException(
+ String.format("Not support ModifyColumn : %s", change.getClass()));
+ }
+ }
+
+ @VisibleForTesting
+ TableChange[] getTableChanges(CatalogBaseTable existingTable,
CatalogBaseTable newTable) {
+ List<TableChange> changes = Lists.newArrayList();
+ changes.addAll(optionsChanges(existingTable.getOptions(),
newTable.getOptions()));
+ if (!Objects.equals(newTable.getComment(), existingTable.getComment())) {
+ changes.add(TableChange.updateComment(newTable.getComment()));
+ }
+ return changes.toArray(new TableChange[0]);
+ }
+
+ @VisibleForTesting
+ TableChange[]
getTableChanges(List<org.apache.flink.table.catalog.TableChange> tableChanges) {
Review Comment:
getGravitinoTableChanges?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
