fsk119 commented on code in PR #21504:
URL: https://github.com/apache/flink/pull/21504#discussion_r1058050529
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/AlterSchemaConverter.java:
##########
@@ -417,6 +422,92 @@ void checkColumnExists(String columnName) {
}
}
+ private static class ModifySchemaConverter extends SchemaConverter {
+
+ ModifySchemaConverter(
+ Schema originalSchema,
+ FlinkTypeFactory typeFactory,
+ SqlValidator sqlValidator,
+ Consumer<SqlTableConstraint> constraintValidator,
+ Function<SqlNode, String> escapeExpressions,
+ SchemaResolver schemaResolver) {
+ super(
+ originalSchema,
+ typeFactory,
+ sqlValidator,
+ constraintValidator,
+ escapeExpressions,
+ schemaResolver);
+ }
+
+ @Override
+ void checkColumnExists(String columnName) {
+ if (!sortedColumnNames.contains(columnName)) {
+ throw new ValidationException(
+ String.format(
+ "%sTry to modify a column `%s` which does not
exist in the table.",
+ EX_MSG_PREFIX, columnName));
+ }
+ }
+
+ @Override
+ void checkPrimaryKeyExists() {
+ if (primaryKey == null) {
+ throw new ValidationException(
+ String.format(
+ "%sThe base table does not define any primary
key constraint. You might "
+ + "want to add a new one.",
+ EX_MSG_PREFIX));
+ }
+ }
+
+ @Override
+ void checkWatermarkExists() {
+ if (watermarkSpec == null) {
+ throw new ValidationException(
+ String.format(
+ "%sThe base table does not define any
watermark. You might "
+ + "want to add a new one.",
+ EX_MSG_PREFIX));
+ }
+ }
+
+ @Override
+ Optional<Integer> getColumnPosition(SqlTableColumnPosition
columnPosition) {
+ if (columnPosition.isFirstColumn() ||
columnPosition.isAfterReferencedColumn()) {
+
sortedColumnNames.remove(columnPosition.getColumn().getName().getSimple());
+ return super.getColumnPosition(columnPosition);
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ Schema.UnresolvedPhysicalColumn convertPhysicalColumn(
+ SqlTableColumn.SqlRegularColumn physicalColumn) {
+ Schema.UnresolvedPhysicalColumn newColumn =
super.convertPhysicalColumn(physicalColumn);
+ String columnName = newColumn.getName();
+ // preserves the primary key's nullability
+ if (primaryKey != null &&
primaryKey.getColumnNames().contains(columnName)) {
+ newColumn =
+ new Schema.UnresolvedPhysicalColumn(
+ columnName,
+ newColumn.getDataType().notNull(),
+ newColumn.getComment().orElse(null));
+ }
Review Comment:
I mean in the `SchemaConverter#convert`. BTW, do you think we need to modify
the type to not null here. As far as I know, in the
`SqlAlterTableSchema#validate` we have already mark used column not 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
