JingsongLi commented on a change in pull request #18236:
URL: https://github.com/apache/flink/pull/18236#discussion_r776575853
##########
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(
Review comment:
orElse(null).
`isManagedTable` accepts nullable catalog.
We can throws exception to a unified mode in below.
##########
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`?
##########
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()
Review comment:
I thought about it, maybe it doesn't have to be ordered, just hit the
specified partition is OK.
We can just do something like `DynamicSinkUtils.validatePartitioning`
--
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]
