[FLINK-34172] Add support for altering a distribution via ALTER TABLE #24886

jnh5y · 2024-06-04T21:48:26Z

What is the purpose of the change

This PR implements the SQL parser changes for ALTER TABLE to support ADD, MODIFY, and DROP DISTRIBUTION statements.

Brief change log

The SQL Parser has been updated.
The AlterSchemaConverter has been updated to pass the changes in a DISTRIBUTION through to the Operation.

Verifying this change

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2024-06-04T21:53:08Z

CI report:

b7f8e0d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

twalthr

Some initial feedback.

twalthr · 2024-06-05T13:59:07Z

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/SqlUnparseUtils.java

@@ -40,7 +41,8 @@ public static void unparseTableSchema(
            int rightPrec,
            SqlNodeList columnList,
            List<SqlTableConstraint> constraints,
-            @Nullable SqlWatermark watermark) {
+            @Nullable SqlWatermark watermark,
+            @Nullable SqlDistribution distribution) {


This doesn't seem right. A distribution is not part of the table schema. A table schema is enclosed in ( ) like in CREATE TABLE <name> (<table schema>). The distribution comes after the schema.

...table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java

...le-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java

twalthr · 2024-06-05T14:35:20Z

...le-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java

+                        .partitionKeys(oldTable.getPartitionKeys())
+                        .options(oldTable.getOptions());
+
+        if (alterTable instanceof SqlAlterTableSchema) {


why can't we simply call oldTable.getDistribution?

Ah, I ended up adding oldTable towards the end of my work. I can try to use it here.

After looking at this again, I figured out that we need to create the new table. Since that table needs the added/modified distribution, we have to take it from the SqlAlterTableSchema rather than the oldTable.

...le-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java

twalthr · 2024-06-05T14:46:47Z

...le-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java

@@ -110,6 +113,7 @@ public Operation convertAlterSchema(
        SchemaConverter converter = createSchemaConverter(alterTableSchema, oldTable);
        converter.updateColumn(alterTableSchema.getColumnPositions().getList());
        alterTableSchema.getWatermark().ifPresent(converter::updateWatermark);
+        alterTableSchema.getDistribution().ifPresent(converter::updateDistribution);


I thought the main change in this class is only to update the distribution in case of a column name change similar to buildUpdatedPrimaryKey.

I think we have to pass the distribution around in order to create the new table objects. I may be missing some alternative here.

twalthr

Thanks for the update. I only had some minor comments. The PR should be good to go in the next iteration.

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlDistribution.java

twalthr · 2024-06-12T13:44:16Z

.../src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

                .isNotPresent();
+
+        prepareNonManagedTableWithDistribution("tb3");
+        // rename column used as distribution key


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

Removing this case of dropping the distribution. The other two cases are for a drop and modify, so I think both of those make sense.

.../src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

…ropped. Moving logic out of AlterSchemaConverter.

twalthr

Thanks for the update @jnh5y. LGTM.

flinkbot added the component=<none> label Jun 4, 2024

jnh5y force-pushed the flink-34172 branch from 4081f6d to ad1b489 Compare June 5, 2024 00:29

twalthr reviewed Jun 5, 2024

View reviewed changes

jnh5y force-pushed the flink-34172 branch from 3b272fb to 1b03e39 Compare June 5, 2024 20:54

twalthr reviewed Jun 12, 2024

View reviewed changes

jnh5y marked this pull request as ready for review June 12, 2024 14:32

jnh5y added 6 commits June 12, 2024 10:40

[FLINK-34172] Add support for altering a distribution via ALTER TABLE

c155354

Flink SQL Parser changes.

006359e

Adding tests to check when a distribution key column is modified or d…

6a7bfad

…ropped. Moving logic out of AlterSchemaConverter.

Test clean up.

34c8e31

Minor refactoring.

4a8f2be

Addressing feedback.

b7f8e0d

jnh5y force-pushed the flink-34172 branch from 9f19163 to b7f8e0d Compare June 12, 2024 14:40

twalthr approved these changes Jun 13, 2024

View reviewed changes

twalthr merged commit d0f9bb4 into apache:master Jun 13, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-34172] Add support for altering a distribution via ALTER TABLE #24886

[FLINK-34172] Add support for altering a distribution via ALTER TABLE #24886

jnh5y commented Jun 4, 2024 •

edited

Loading

flinkbot commented Jun 4, 2024 •

edited

Loading

twalthr left a comment

twalthr Jun 5, 2024

jnh5y Jun 6, 2024

twalthr Jun 5, 2024

jnh5y Jun 5, 2024

jnh5y Jun 6, 2024

twalthr Jun 5, 2024

jnh5y Jun 6, 2024

twalthr left a comment

twalthr Jun 12, 2024

jnh5y Jun 12, 2024

twalthr left a comment

[FLINK-34172] Add support for altering a distribution via ALTER TABLE #24886

[FLINK-34172] Add support for altering a distribution via ALTER TABLE #24886

Conversation

jnh5y commented Jun 4, 2024 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jun 4, 2024 • edited Loading

CI report:

twalthr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr left a comment

Choose a reason for hiding this comment

jnh5y commented Jun 4, 2024 •

edited

Loading

flinkbot commented Jun 4, 2024 •

edited

Loading