This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 363023d27a0 [FLINK-28011][hive] HiveSource should get partitions in
batch instead of getting partition one by one
363023d27a0 is described below
commit 363023d27a0fe838a2fa20422ae111da43643df2
Author: luoyuxia <[email protected]>
AuthorDate: Sat Jun 18 16:27:38 2022 +0800
[FLINK-28011][hive] HiveSource should get partitions in batch instead of
getting partition one by one
This closes #20015
---
.../connectors/hive/util/HivePartitionUtils.java | 39 ++++++++++++++++++----
1 file changed, 32 insertions(+), 7 deletions(-)
diff --git
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
index 7bf8bbc83b7..3b3e28083ba 100644
---
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
+++
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
@@ -35,6 +35,7 @@ import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
@@ -165,13 +166,13 @@ public class HivePartitionUtils {
if (partitionColNames != null && partitionColNames.size() > 0) {
List<Partition> partitions = new ArrayList<>();
if (remainingPartitions != null) {
- for (Map<String, String> spec : remainingPartitions) {
- partitions.add(
- client.getPartition(
- dbName,
- tableName,
- partitionSpecToValues(spec,
partitionColNames)));
- }
+ List<String> partitionNames =
+ getPartitionNames(
+ remainingPartitions,
+ partitionColNames,
+
JobConfUtils.getDefaultPartitionName(jobConf));
+ partitions.addAll(
+ client.getPartitionsByNames(dbName, tableName,
partitionNames));
} else {
partitions.addAll(client.listPartitions(dbName, tableName,
(short) -1));
}
@@ -189,6 +190,30 @@ public class HivePartitionUtils {
return allHivePartitions;
}
+ /**
+ * Get the partitions' name by partitions' spec.
+ *
+ * @param partitionsSpec a list contains the spec of the partitions, one
of which is for one
+ * partition. The map for the spec of partition can be unordered.
+ * @param partitionColNames the partition column's name
+ * @param defaultStr the default value used to make partition name when
the key or value for the
+ * partition's spec partition column in the spec is null or empty
string.
+ * @return a list contains the partitions' name like "p1=v1/p2=v2", one of
which is for one
+ * partition.
+ */
+ public static List<String> getPartitionNames(
+ List<Map<String, String>> partitionsSpec,
+ List<String> partitionColNames,
+ String defaultStr) {
+ List<String> partitionNames = new ArrayList<>(partitionsSpec.size());
+ for (Map<String, String> partitionSpec : partitionsSpec) {
+ List<String> pVals = partitionSpecToValues(partitionSpec,
partitionColNames);
+ // Construct a pattern of the form:
partKey=partVal/partKey2=partVal2/...
+ partitionNames.add(FileUtils.makePartName(partitionColNames,
pVals, defaultStr));
+ }
+ return partitionNames;
+ }
+
public static List<String> partitionSpecToValues(
Map<String, String> spec, List<String> partitionColNames) {
checkArgument(
