lirui-apache commented on a change in pull request #8449: [FLINK-12235][hive]
Support partition related operations in HiveCatalog
URL: https://github.com/apache/flink/pull/8449#discussion_r286297791
##########
File path:
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
##########
@@ -607,44 +609,231 @@ private static Table instantiateHiveTable(ObjectPath
tablePath, CatalogBaseTabl
// ------ partitions ------
@Override
- public void createPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, CatalogPartition partition, boolean ignoreIfExists)
- throws TableNotExistException,
TableNotPartitionedException, PartitionSpecInvalidException,
PartitionAlreadyExistsException, CatalogException {
- throw new UnsupportedOperationException();
+ public boolean partitionExists(ObjectPath tablePath,
CatalogPartitionSpec partitionSpec)
+ throws CatalogException {
+ try {
+ return getHivePartition(tablePath, partitionSpec) !=
null;
+ } catch (NoSuchObjectException | TableNotExistException |
PartitionSpecInvalidException e) {
+ return false;
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to get partition %s of
table %s", partitionSpec, tablePath), e);
+ }
}
@Override
- public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, boolean ignoreIfNotExists)
- throws PartitionNotExistException, CatalogException {
- throw new UnsupportedOperationException();
+ public void createPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, CatalogPartition partition, boolean ignoreIfExists)
+ throws TableNotExistException, TableNotPartitionedException,
PartitionSpecInvalidException, PartitionAlreadyExistsException,
CatalogException {
+
+ Table hiveTable = getHiveTable(tablePath);
+
+ ensurePartitionedTable(tablePath, hiveTable);
+
+ try {
+ client.add_partition(createHivePartition(hiveTable,
partitionSpec, partition));
+ } catch (AlreadyExistsException e) {
+ if (!ignoreIfExists) {
+ throw new
PartitionAlreadyExistsException(catalogName, tablePath, partitionSpec);
+ }
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to create partition %s of
table %s", partitionSpec, tablePath));
+ }
}
@Override
- public void alterPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, CatalogPartition newPartition, boolean ignoreIfNotExists)
- throws PartitionNotExistException, CatalogException {
- throw new UnsupportedOperationException();
+ public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ try {
+ Table hiveTable = getHiveTable(tablePath);
+ client.dropPartition(tablePath.getDatabaseName(),
tablePath.getObjectName(),
+ getOrderedFullPartitionValues(partitionSpec,
getFieldNames(hiveTable.getPartitionKeys()), tablePath), true);
+ } catch (NoSuchObjectException e) {
+ if (!ignoreIfNotExists) {
+ throw new
PartitionNotExistException(catalogName, tablePath, partitionSpec, e);
+ }
+ } catch (MetaException | TableNotExistException |
PartitionSpecInvalidException e) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec, e);
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to drop partition %s of
table %s", partitionSpec, tablePath));
+ }
}
@Override
public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath)
- throws TableNotExistException,
TableNotPartitionedException, CatalogException {
- throw new UnsupportedOperationException();
+ throws TableNotExistException, TableNotPartitionedException,
CatalogException {
+ Table hiveTable = getHiveTable(tablePath);
+
+ ensurePartitionedTable(tablePath, hiveTable);
+
+ try {
+ return
client.listPartitionNames(tablePath.getDatabaseName(),
tablePath.getObjectName(), (short) -1).stream()
+
.map(HiveCatalog::createPartitionSpec).collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to list partitions of
table %s", tablePath), e);
+ }
}
@Override
public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath,
CatalogPartitionSpec partitionSpec)
- throws TableNotExistException,
TableNotPartitionedException, CatalogException {
- throw new UnsupportedOperationException();
+ throws TableNotExistException, TableNotPartitionedException,
CatalogException {
+ Table hiveTable = getHiveTable(tablePath);
+
+ ensurePartitionedTable(tablePath, hiveTable);
+
+ try {
+ // partition spec can be partial
+ List<String> partialVals =
MetaStoreUtils.getPvals(hiveTable.getPartitionKeys(),
partitionSpec.getPartitionSpec());
+ return
client.listPartitionNames(tablePath.getDatabaseName(),
tablePath.getObjectName(), partialVals,
+ (short)
-1).stream().map(HiveCatalog::createPartitionSpec).collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to list partitions of
table %s", tablePath), e);
+ }
}
@Override
public CatalogPartition getPartition(ObjectPath tablePath,
CatalogPartitionSpec partitionSpec)
- throws PartitionNotExistException, CatalogException {
- throw new UnsupportedOperationException();
+ throws PartitionNotExistException, CatalogException {
+ try {
+ Partition hivePartition = getHivePartition(tablePath,
partitionSpec);
+ return createCatalogPartition(hivePartition);
+ } catch (NoSuchObjectException | MetaException |
TableNotExistException | PartitionSpecInvalidException e) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec, e);
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to get partition %s of
table %s", partitionSpec, tablePath), e);
+ }
}
@Override
- public boolean partitionExists(ObjectPath tablePath,
CatalogPartitionSpec partitionSpec) throws CatalogException {
- throw new UnsupportedOperationException();
+ public void alterPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, CatalogPartition newPartition, boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ // Explicitly check if the partition exists or not
+ // because alter_partition() doesn't throw
NoSuchObjectException like dropPartition() when the target doesn't exist
+ if (partitionExists(tablePath, partitionSpec)) {
+ try {
+ Table hiveTable = getHiveTable(tablePath);
+ Partition newHivePartition =
createHivePartition(hiveTable, partitionSpec, newPartition);
+ if (newHivePartition.getSd().getLocation() ==
null) {
+ Partition oldHivePartition =
getHivePartition(tablePath, partitionSpec);
+
newHivePartition.getSd().setLocation(oldHivePartition.getSd().getLocation());
+ }
+ client.alter_partition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ newHivePartition
+ );
+ } catch (InvalidOperationException | MetaException |
TableNotExistException | PartitionSpecInvalidException e) {
+ throw new
PartitionNotExistException(catalogName, tablePath, partitionSpec, e);
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to alter existing
partition with new partition %s of table %s",
+ partitionSpec, tablePath), e);
+ }
+ } else if (!ignoreIfNotExists) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec);
+ }
+ }
+
+ private Partition createHivePartition(Table hiveTable,
CatalogPartitionSpec partitionSpec, CatalogPartition catalogPartition)
Review comment:
OK
----------------------------------------------------------------
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
With regards,
Apache Git Services
