JingsongLi commented on a change in pull request #10381: [FLINK-14513][hive]
Implement listPartitionsByFilter to HiveCatalog
URL: https://github.com/apache/flink/pull/10381#discussion_r355107531
##########
File path:
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
##########
@@ -759,9 +763,33 @@ public void dropPartition(ObjectPath tablePath,
CatalogPartitionSpec partitionSp
}
@Override
- public List<CatalogPartitionSpec> listPartitionsByFilter(ObjectPath
tablePath, List<Expression> filters)
+ public List<CatalogPartitionSpec> listPartitionsByFilter(ObjectPath
tablePath, List<Expression> expressions)
throws TableNotExistException,
TableNotPartitionedException, CatalogException {
- throw new UnsupportedOperationException();
+ Table hiveTable = getHiveTable(tablePath);
+ ensurePartitionedTable(tablePath, hiveTable);
+ List<String> partColNames =
getFieldNames(hiveTable.getPartitionKeys());
+ Optional<String> filter = HiveTableUtil.makePartitionFilter(
+ getNonPartitionFields(hiveConf,
hiveTable).size(), partColNames, expressions);
+ if (!filter.isPresent()) {
+ throw new UnsupportedOperationException(
+ "HiveCatalog is unable to handle the
partition filter expressions: " + expressions);
+ }
+ try {
+ PartitionSpecProxy.PartitionIterator partitions =
client.listPartitionSpecsByFilter(
+ tablePath.getDatabaseName(),
tablePath.getObjectName(), filter.get(), (short) -1).getPartitionIterator();
+ List<CatalogPartitionSpec> res = new ArrayList<>();
+ while (partitions.hasNext()) {
+ Partition partition = partitions.next();
Review comment:
Unfortunately, we have got the information, but we need to go to get it
again.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
