LadyForest commented on code in PR #21504:
URL: https://github.com/apache/flink/pull/21504#discussion_r1058054716
##########
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:
> in the SqlAlterTableSchema#validate we have already mark used column not
null.
Actually, `SqlAlterTableSchema#validate` is not suitable under this
condition. Consider the previous case I mentioned,
`SqlAlterTableSchema#validate` will do nothing because this change does not
reveal any primary key info.
--
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]
