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_r284966457
##########
File path:
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalogBase.java
##########
@@ -407,4 +443,201 @@ private Table getHiveTable(ObjectPath tablePath) throws
TableNotExistException {
String.format("Failed to get table %s from Hive
metastore", tablePath.getFullName()), e);
}
}
+
+ // ------ partitions ------
+
+ @Override
+ public boolean partitionExists(ObjectPath tablePath,
CatalogPartitionSpec partitionSpec)
+ throws CatalogException {
+ try {
+ Table hiveTable = getHiveTable(tablePath);
+ return client.getPartition(tablePath.getDatabaseName(),
tablePath.getObjectName(),
+ getFullPartitionValues(tablePath,
partitionSpec, getFieldNames(hiveTable.getPartitionKeys()))) != 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 createPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, CatalogPartition partition, boolean ignoreIfExists)
+ throws TableNotExistException, TableNotPartitionedException,
PartitionSpecInvalidException, PartitionAlreadyExistsException,
CatalogException {
+
+ validateCatalogPartition(partition);
+
+ 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 dropPartition(ObjectPath tablePath, CatalogPartitionSpec
partitionSpec, boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ try {
+ Table hiveTable = getHiveTable(tablePath);
+ ensurePartitionedTable(tablePath, hiveTable);
+ client.dropPartition(tablePath.getDatabaseName(),
tablePath.getObjectName(),
+ getFullPartitionValues(tablePath,
partitionSpec, getFieldNames(hiveTable.getPartitionKeys())), true);
+ } catch (NoSuchObjectException e) {
+ if (!ignoreIfNotExists) {
+ throw new
PartitionNotExistException(catalogName, tablePath, partitionSpec);
+ }
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to drop partition %s of
table %s", partitionSpec, tablePath));
+ } catch (TableNotPartitionedException | TableNotExistException
| PartitionSpecInvalidException e) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec, e);
+ }
+ }
+
+ @Override
+ public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath)
+ throws TableNotExistException, TableNotPartitionedException,
CatalogException {
+ Table hiveTable = getHiveTable(tablePath);
+
+ ensurePartitionedTable(tablePath, hiveTable);
+
+ try {
+ return
client.listPartitionNames(tablePath.getDatabaseName(),
tablePath.getObjectName(), (short) -1).stream()
+
.map(HiveCatalogBase::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 {
+ 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(HiveCatalogBase::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 {
+ try {
+ Table hiveTable = getHiveTable(tablePath);
+ ensurePartitionedTable(tablePath, hiveTable);
+ Partition hivePartition =
client.getPartition(tablePath.getDatabaseName(), tablePath.getObjectName(),
+ getFullPartitionValues(tablePath,
partitionSpec, getFieldNames(hiveTable.getPartitionKeys())));
+ return createCatalogPartition(hivePartition);
+ } catch (NoSuchObjectException e) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec);
+ } catch (TException e) {
+ throw new CatalogException(
+ String.format("Failed to get partition %s of
table %s", partitionSpec, tablePath), e);
+ } catch (TableNotPartitionedException | TableNotExistException
| PartitionSpecInvalidException e) {
+ throw new PartitionNotExistException(catalogName,
tablePath, partitionSpec, e);
+ }
+ }
+
+ @Override
+ 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);
+ ensurePartitionedTable(tablePath, hiveTable);
+ client.alter_partition(
Review comment:
Same as `dropPartition`. Will make the change
----------------------------------------------------------------
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
