[GitHub] [flink] wuchong commented on a change in pull request #15755: [FLINK-22318][table] Support RENAME column name for ALTER TABLE state…
wuchong commented on a change in pull request #15755:
URL: https://github.com/apache/flink/pull/15755#discussion_r620947942
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
##
@@ -144,6 +149,116 @@ public static Operation convertChangeColumn(
// TODO: handle watermark and constraints
}
+public static Operation convertRenameColumn(
+ObjectIdentifier tableIdentifier,
+String originColumnName,
+String newColumnName,
+CatalogTable catalogTable) {
+
+Schema modifiedTableSchema = catalogTable.getUnresolvedSchema();
+validateColumnName(originColumnName, newColumnName,
modifiedTableSchema);
+
+Schema.Builder builder = Schema.newBuilder();
+// build column
+modifiedTableSchema.getColumns().stream()
+.forEach(
+column -> {
+if (StringUtils.equals(column.getName(),
originColumnName)) {
+buildNewColumnFromOriginColumn(builder,
column, newColumnName);
+} else {
+buildNewColumnFromOriginColumn(builder,
column, column.getName());
+}
+});
+// build primary key column
+List originPrimaryKeyNames =
+modifiedTableSchema
+.getPrimaryKey()
+.map(Schema.UnresolvedPrimaryKey::getColumnNames)
+.orElseGet(Lists::newArrayList);
+
+List newPrimaryKeyNames =
+originPrimaryKeyNames.stream()
+.map(
+pkName ->
+StringUtils.equals(pkName,
originColumnName)
+? newColumnName
+: pkName)
+.collect(Collectors.toList());
+
+if (newPrimaryKeyNames.size() > 0) {
+builder.primaryKey(newPrimaryKeyNames);
+}
+// build watermark
+modifiedTableSchema.getWatermarkSpecs().stream()
+.forEach(
+watermarkSpec -> {
+String watermarkRefColumnName =
watermarkSpec.getColumnName();
+Expression watermarkExpression =
watermarkSpec.getWatermarkExpression();
+if (StringUtils.equals(watermarkRefColumnName,
originColumnName)) {
+String newWatermarkExpression =
+((SqlCallExpression)
watermarkExpression)
+.getSqlExpression()
+
.replace(watermarkRefColumnName, newColumnName);
Review comment:
Thanks @twalthr . I suggest resolve the schema first and traverse the
node tree to check whether it contains the renamed column, and throw a helpful
exception if contains.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15755: [FLINK-22318][table] Support RENAME column name for ALTER TABLE state…
wuchong commented on a change in pull request #15755:
URL: https://github.com/apache/flink/pull/15755#discussion_r620886208
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
##
@@ -144,6 +149,116 @@ public static Operation convertChangeColumn(
// TODO: handle watermark and constraints
}
+public static Operation convertRenameColumn(
+ObjectIdentifier tableIdentifier,
+String originColumnName,
+String newColumnName,
+CatalogTable catalogTable) {
+
+Schema modifiedTableSchema = catalogTable.getUnresolvedSchema();
+validateColumnName(originColumnName, newColumnName,
modifiedTableSchema);
+
+Schema.Builder builder = Schema.newBuilder();
+// build column
+modifiedTableSchema.getColumns().stream()
+.forEach(
+column -> {
+if (StringUtils.equals(column.getName(),
originColumnName)) {
+buildNewColumnFromOriginColumn(builder,
column, newColumnName);
+} else {
+buildNewColumnFromOriginColumn(builder,
column, column.getName());
+}
+});
+// build primary key column
+List originPrimaryKeyNames =
+modifiedTableSchema
+.getPrimaryKey()
+.map(Schema.UnresolvedPrimaryKey::getColumnNames)
+.orElseGet(Lists::newArrayList);
+
+List newPrimaryKeyNames =
+originPrimaryKeyNames.stream()
+.map(
+pkName ->
+StringUtils.equals(pkName,
originColumnName)
+? newColumnName
+: pkName)
+.collect(Collectors.toList());
+
+if (newPrimaryKeyNames.size() > 0) {
+builder.primaryKey(newPrimaryKeyNames);
+}
+// build watermark
+modifiedTableSchema.getWatermarkSpecs().stream()
+.forEach(
+watermarkSpec -> {
+String watermarkRefColumnName =
watermarkSpec.getColumnName();
+Expression watermarkExpression =
watermarkSpec.getWatermarkExpression();
+if (StringUtils.equals(watermarkRefColumnName,
originColumnName)) {
+String newWatermarkExpression =
+((SqlCallExpression)
watermarkExpression)
+.getSqlExpression()
+
.replace(watermarkRefColumnName, newColumnName);
Review comment:
Would be better to investigate solutions first, I'm not sure whether
there are any other ways to do this.
cc @twalthr , do you have any ideas?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #15755: [FLINK-22318][table] Support RENAME column name for ALTER TABLE state…
wuchong commented on a change in pull request #15755:
URL: https://github.com/apache/flink/pull/15755#discussion_r620825538
##
File path:
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
##
@@ -478,6 +481,17 @@ SqlAlterTable SqlAlterTable() :
tableIdentifier,
newTableIdentifier);
}
+|
+
+originColumnName = SimpleIdentifier()
+
+newColumnName = SimpleIdentifier()
+{
+return new SqlAlterTableRenameColumn(
+startPos.plus(getPos()),
+tableIdentifier,
+originColumnName,newColumnName);
Review comment:
nit: Please make them in separate lines.
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
##
@@ -144,6 +149,116 @@ public static Operation convertChangeColumn(
// TODO: handle watermark and constraints
}
+public static Operation convertRenameColumn(
+ObjectIdentifier tableIdentifier,
+String originColumnName,
+String newColumnName,
+CatalogTable catalogTable) {
+
+Schema modifiedTableSchema = catalogTable.getUnresolvedSchema();
Review comment:
`modifiedTableSchema` sounds like this schema has been modified, would
be better to call `originSchema`.
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
##
@@ -144,6 +149,116 @@ public static Operation convertChangeColumn(
// TODO: handle watermark and constraints
}
+public static Operation convertRenameColumn(
+ObjectIdentifier tableIdentifier,
+String originColumnName,
+String newColumnName,
+CatalogTable catalogTable) {
+
+Schema modifiedTableSchema = catalogTable.getUnresolvedSchema();
+validateColumnName(originColumnName, newColumnName,
modifiedTableSchema);
+
+Schema.Builder builder = Schema.newBuilder();
+// build column
+modifiedTableSchema.getColumns().stream()
+.forEach(
+column -> {
+if (StringUtils.equals(column.getName(),
originColumnName)) {
+buildNewColumnFromOriginColumn(builder,
column, newColumnName);
+} else {
+buildNewColumnFromOriginColumn(builder,
column, column.getName());
+}
+});
+// build primary key column
+List originPrimaryKeyNames =
+modifiedTableSchema
+.getPrimaryKey()
+.map(Schema.UnresolvedPrimaryKey::getColumnNames)
+.orElseGet(Lists::newArrayList);
+
+List newPrimaryKeyNames =
+originPrimaryKeyNames.stream()
+.map(
+pkName ->
+StringUtils.equals(pkName,
originColumnName)
+? newColumnName
+: pkName)
+.collect(Collectors.toList());
+
+if (newPrimaryKeyNames.size() > 0) {
+builder.primaryKey(newPrimaryKeyNames);
+}
+// build watermark
+modifiedTableSchema.getWatermarkSpecs().stream()
+.forEach(
+watermarkSpec -> {
+String watermarkRefColumnName =
watermarkSpec.getColumnName();
+Expression watermarkExpression =
watermarkSpec.getWatermarkExpression();
+if (StringUtils.equals(watermarkRefColumnName,
originColumnName)) {
+String newWatermarkExpression =
+((SqlCallExpression)
watermarkExpression)
+.getSqlExpression()
+
.replace(watermarkRefColumnName, newColumnName);
Review comment:
1. We can't guarantee this is always `SqlCallExpression`.
2. We can't use String `replace` for renaming, this is very error-prone,
e.g. the original expressions is `f123 - f1` and `f1` is rename to `f2`, then
the replaced result would be `f223 - f2` which is wrong.
If we don't have a good way to replace column names in expression, we can
disallow rename for columns referenced in expressions in the first version.
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
##
@@ -144,6 +149,116 @@ public static Operation convertChangeColumn(
//
