LadyForest commented on a change in pull request #18236:
URL: https://github.com/apache/flink/pull/18236#discussion_r776606481
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java
##########
@@ -542,6 +551,60 @@ private Operation convertAlterTableReset(
return new AlterTableOptionsOperation(tableIdentifier,
oldTable.copy(newOptions));
}
+ private Operation convertAlterTableCompact(
+ ObjectIdentifier tableIdentifier,
+ ResolvedCatalogTable resolvedCatalogTable,
+ SqlAlterTableCompact alterTableCompact) {
+ Catalog catalog =
+ catalogManager
+ .getCatalog(tableIdentifier.getCatalogName())
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Catalog %s not
exists",
+
tableIdentifier.getCatalogName())));
+ if (ManagedTableListener.isManagedTable(catalog,
resolvedCatalogTable)) {
+ LinkedHashMap<String, String> partitionKVs =
alterTableCompact.getPartitionKVs();
+ CatalogPartitionSpec partitionSpec = null;
+ if (partitionKVs != null) {
+ List<String> orderedPartitionKeys =
resolvedCatalogTable.getPartitionKeys();
+ int index = 0;
+ String exMsg =
+ orderedPartitionKeys.isEmpty()
+ ? String.format(
+ "Table %s is not partitioned.",
+ tableIdentifier.asSummaryString())
+ : String.format(
+ "Available ordered partition columns:
[%s]",
+ orderedPartitionKeys.stream()
+
.collect(Collectors.joining("', '", "'", "'")));
+ for (Map.Entry<String, String> partitionKV :
partitionKVs.entrySet()) {
+ String partitionKey = partitionKV.getKey();
+ if (index >= orderedPartitionKeys.size()
+ ||
!orderedPartitionKeys.get(index).equals(partitionKey)) {
+ exMsg =
+ orderedPartitionKeys.contains(partitionKey)
+ ? String.format(
+ "Partition column '%s' not
appeared in order. %s",
+ partitionKey, exMsg)
+ : String.format(
+ "Partition column '%s' not
defined in the table schema. %s",
+ partitionKey, exMsg);
+ throw new ValidationException(exMsg);
+ }
+ index++;
+ }
+ partitionSpec = new CatalogPartitionSpec(partitionKVs);
+ }
+ return new AlterTableCompactOperation(tableIdentifier,
partitionSpec);
+ }
+ throw new UnsupportedOperationException(
Review comment:
> `ValidationException`?
I'm fine with that.
--
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]
