twalthr commented on code in PR #24886: URL: https://github.com/apache/flink/pull/24886#discussion_r1636474722
########## flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlDistribution.java: ########## @@ -70,17 +70,28 @@ public List<SqlNode> getOperandList() { @Override public void unparse(SqlWriter writer, int leftPrec, int rightPrec) { - writer.newlineAndIndent(); + unparse(writer, leftPrec, rightPrec, "DISTRIBUTED", true); + } + + public void unparseAlter(SqlWriter writer, int leftPrec, int rightPrec) { + unparse(writer, leftPrec, rightPrec, "DISTRIBUTION", false); + } + + public void unparse( Review Comment: can be private ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java: ########## @@ -2208,6 +2241,59 @@ public void testAlterTableModifyPk() throws Exception { .build()); } + @Test + public void testAlterTableAddDistribution() throws Exception { + prepareNonManagedTable("tb1", false); + + Operation operation = parse("alter table tb1 add distribution by hash(a) into 12 buckets"); + ObjectIdentifier tableIdentifier = ObjectIdentifier.of("cat1", "db1", "tb1"); + assertAlterTableDistribution( + operation, + tableIdentifier, + TableDistribution.ofHash(Collections.singletonList("a"), 12), + "ALTER TABLE cat1.db1.tb1\n" + " ADD DISTRIBUTED BY HASH(`a`) INTO 12 BUCKETS\n"); + } + + @Test + public void testFailedToAlterTableAddDistribution() throws Exception { + prepareNonManagedTableWithDistribution("tb1"); + + // modify watermark on a table without watermark Review Comment: incorrect comment ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java: ########## @@ -2208,6 +2241,59 @@ public void testAlterTableModifyPk() throws Exception { .build()); } + @Test + public void testAlterTableAddDistribution() throws Exception { + prepareNonManagedTable("tb1", false); + + Operation operation = parse("alter table tb1 add distribution by hash(a) into 12 buckets"); + ObjectIdentifier tableIdentifier = ObjectIdentifier.of("cat1", "db1", "tb1"); + assertAlterTableDistribution( + operation, + tableIdentifier, + TableDistribution.ofHash(Collections.singletonList("a"), 12), + "ALTER TABLE cat1.db1.tb1\n" + " ADD DISTRIBUTED BY HASH(`a`) INTO 12 BUCKETS\n"); + } + + @Test + public void testFailedToAlterTableAddDistribution() throws Exception { + prepareNonManagedTableWithDistribution("tb1"); + + // modify watermark on a table without watermark + assertThatThrownBy( + () -> parse("alter table tb1 add distribution by hash(a) into 12 buckets")) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("You can modify it or drop it before adding a new one."); + } + + @Test + public void testFailedToAlterTableModifyDistribution() throws Exception { + prepareNonManagedTable("tb2", false); + + // modify watermark on a table without watermark Review Comment: incorrect comment ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java: ########## @@ -1361,16 +1374,36 @@ public void testAlterTableDropConstraint() throws Exception { .getUnresolvedSchema() .getPrimaryKey()) .isNotPresent(); + } + + @Test + public void testAlterTableDropDistribution() throws Exception { + prepareNonManagedTableWithDistribution("tb1"); + String expectedSummaryString = "ALTER TABLE cat1.db1.tb1\n DROP DISTRIBUTION"; - operation = parse("alter table tb1 drop primary key"); + Operation operation = parse("alter table tb1 drop distribution"); assertThat(operation).isInstanceOf(AlterTableChangeOperation.class); assertThat(operation.asSummaryString()).isEqualTo(expectedSummaryString); - assertThat( - ((AlterTableChangeOperation) operation) - .getNewTable() - .getUnresolvedSchema() - .getPrimaryKey()) + assertThat(((AlterTableChangeOperation) operation).getNewTable().getDistribution()) .isNotPresent(); + + prepareNonManagedTableWithDistribution("tb3"); + // rename column used as distribution key Review Comment: comment seems to be wrong, also isn't this tested above already, at least `is used as a distribution key` occurs two times in this clas ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java: ########## @@ -2485,13 +2571,31 @@ private void prepareManagedTable(boolean hasPartition) throws Exception { prepareTable("tb1", true, hasPartition, false, 0); } + private void prepareNonManagedTableWithDistribution(String tableName) throws Exception { + TableDistribution distribution = + TableDistribution.of( + TableDistribution.Kind.HASH, 6, Collections.singletonList("c")); + prepareTable(tableName, false, false, false, 0, distribution); + } + private void prepareTable( String tableName, boolean managedTable, boolean hasPartition, boolean hasWatermark, int numOfPkFields) throws Exception { + prepareTable(tableName, managedTable, hasPartition, hasWatermark, numOfPkFields, null); + } + + private void prepareTable( + String tableName, + boolean managedTable, + boolean hasPartition, + boolean hasWatermark, + int numOfPkFields, + TableDistribution tableDistribution) Review Comment: add `@Nullable` if this arg can be null -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org