Skip to content

[improvement](hive) Push partition filters to HMS #67725

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhaorongsheng wants to merge 1 commit into
base: master
Choose a base branch
from
+699 −78
Open

[improvement](hive) Push partition filters to HMS #67725

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 195 additions & 39 deletions ...e-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.doris.connector.hms.HmsCreateDatabaseRequest;
import org.apache.doris.connector.hms.HmsCreateTableRequest;
import org.apache.doris.connector.hms.HmsPartitionBatchResult;
import org.apache.doris.connector.hms.HmsPartitionBatchStats;
import org.apache.doris.connector.hms.HmsPartitionInfo;
import org.apache.doris.connector.hms.HmsTableInfo;
import org.apache.doris.connector.hms.HmsTypeMapping;
Expand Down Expand Up @@ -472,18 +473,24 @@ public Optional<ConnectorTableHandle> getTableHandle(

// Build partition key column names
List<String> partKeyNames = Collections.emptyList();
Map<String, String> partKeyTypes = Collections.emptyMap();
List<ConnectorColumn> partKeys = tableInfo.getPartitionKeys();
if (partKeys != null && !partKeys.isEmpty()) {
partKeyNames = partKeys.stream()
.map(ConnectorColumn::getName)
.collect(Collectors.toList());
partKeyTypes = new HashMap<>();
for (ConnectorColumn partKey : partKeys) {
partKeyTypes.put(partKey.getName(), partKey.getType().getTypeName());
}
}

HiveTableHandle handle = new HiveTableHandle.Builder(dbName, tableName, tableType)
.inputFormat(tableInfo.getInputFormat())
.serializationLib(tableInfo.getSerializationLib())
.location(tableInfo.getLocation())
.partitionKeyNames(partKeyNames)
.partitionKeyTypes(partKeyTypes)
.sdParameters(tableInfo.getSdParameters())
.tableParameters(tableInfo.getParameters())
.firstColumnIsString(firstColumnIsString(tableInfo))
Expand Down Expand Up @@ -572,6 +579,9 @@ public ConnectorTableSchema getTableSchema(
// would also admit hudi-on-HMS, which legacy excluded). This branch is reached only for a HiveTableHandle;
// an iceberg-on-HMS table is served by the delegation branch above (which reflects the iceberg sibling's
// own auto-analyze capability), and a hudi-on-HMS table's connector declares neither.
if (!partitionKeys.isEmpty()) {
perTableCapabilities.add(ConnectorCapability.SUPPORTS_CONNECTOR_PARTITION_PRUNING);
}
if (supportsHiveColumnAutoAnalyze(tableInfo)) {
perTableCapabilities.add(ConnectorCapability.SUPPORTS_COLUMN_AUTO_ANALYZE);
}
Expand Down Expand Up @@ -1168,46 +1178,14 @@ public Optional<FilterApplicationResult<ConnectorTableHandle>> applyFilter(
return Optional.empty();
}

// Extract equality predicates on partition columns from the expression
Map<String, List<String>> partitionPredicates = extractPartitionPredicates(
constraint.getExpression(), partKeyNames);
if (partitionPredicates.isEmpty()) {
PartitionPruningResult pruningResult = prunePartitions(session, hiveHandle, constraint.getExpression());
if (pruningResult == null) {
return Optional.empty();
}

// Build partition name filter patterns for HMS
List<String> allPartNames = hmsClient.listPartitionNames(
hiveHandle.getDbName(), hiveHandle.getTableName(), 100000);
List<String> matchedPartNames = prunePartitionNames(
allPartNames, partKeyNames, partitionPredicates);

if (matchedPartNames.size() == allPartNames.size()) {
// No pruning effect
return Optional.empty();
}

HmsPartitionBatchResult pruningResult;
try {
pruningResult = matchedPartNames.isEmpty()
? null : hmsClient.getExistingPartitionsWithStats(
hiveHandle.getDbName(), hiveHandle.getTableName(), matchedPartNames);
} catch (HmsClientException e) {
if (e.getPartitionBatchStats() != null) {
HiveScanPlanProvider.recordPruningFailure(
session, hiveHandle.getDbName(), hiveHandle.getTableName(), e.getPartitionBatchStats());
}
throw e;
}
List<HmsPartitionInfo> prunedPartitions = pruningResult == null
? Collections.emptyList() : pruningResult.getPartitions();

LOG.info("Partition pruning: {}.{} all={} pruned={}",
hiveHandle.getDbName(), hiveHandle.getTableName(),
allPartNames.size(), prunedPartitions.size());

HiveTableHandle newHandle = hiveHandle.toBuilder()
.prunedPartitions(prunedPartitions)
.pruningBatchStats(pruningResult == null ? null : pruningResult.getStats())
.prunedPartitions(pruningResult.partitions)
.pruningBatchStats(pruningResult.batchStats)
.build();
return Optional.of(new FilterApplicationResult<>(
newHandle, constraint.getExpression(), false));
Expand All @@ -1234,9 +1212,9 @@ public List<String> listPartitionNames(ConnectorSession session, ConnectorTableH
}

/**
* Lists all partitions with metadata. The {@code filter} is intentionally ignored: legacy hive
* materialized its full partition view and pruned FE-side (mirrors {@code PaimonConnectorMetadata} /
* {@code MaxComputeConnectorMetadata}).
* Lists all partitions with metadata. A filter that contains supported Hive partition equality or IN
* predicates is resolved through HMS before the generic FE partition map is built; unsupported filters keep
* the existing full-list-and-local-pruning fallback.
*
* <p>{@code lastModifiedMillis} is deliberately left {@link ConnectorPartitionInfo#UNKNOWN} (-1):
* reading each partition's {@code transient_lastDdlTime} requires a {@code get_partitions_by_names}
Expand All @@ -1261,6 +1239,15 @@ public List<ConnectorPartitionInfo> listPartitions(ConnectorSession session,
return siblingMetadata(session, handle).listPartitions(session, handle, filter);
}
HiveTableHandle hiveHandle = (HiveTableHandle) handle;
if (hiveHandle.getPrunedPartitions() != null) {
return toConnectorPartitionInfos(hiveHandle.getPrunedPartitions(), hiveHandle.getPartitionKeyNames());
}
if (filter.isPresent()) {
PartitionPruningResult pruningResult = prunePartitions(session, hiveHandle, filter.get());
if (pruningResult != null) {
return toConnectorPartitionInfos(pruningResult.partitions, hiveHandle.getPartitionKeyNames());
}
}
if (partitionViewCache == null || filter.isPresent()) {
return listPartitionsUncached(hiveHandle);
}
Expand Down Expand Up @@ -1288,6 +1275,78 @@ private List<ConnectorPartitionInfo> listPartitionsUncached(HiveTableHandle hive
return result;
}

private PartitionPruningResult prunePartitions(ConnectorSession session, HiveTableHandle hiveHandle,
ConnectorExpression expression) {
List<String> partKeyNames = hiveHandle.getPartitionKeyNames();
Map<String, List<String>> partitionPredicates = extractPartitionPredicates(expression, partKeyNames);
if (partitionPredicates.isEmpty()) {
return null;
}

String hmsFilter = buildHmsPartitionFilter(partKeyNames, hiveHandle.getPartitionKeyTypes(),
partitionPredicates);
if (hmsFilter != null) {
try {
List<HmsPartitionInfo> prunedPartitions = hmsClient.listPartitionsByFilter(
hiveHandle.getDbName(), hiveHandle.getTableName(), hmsFilter);
LOG.info("Partition pruning through HMS filter: {}.{} filter={} pruned={}",
hiveHandle.getDbName(), hiveHandle.getTableName(), hmsFilter, prunedPartitions.size());
return new PartitionPruningResult(prunedPartitions, null);
} catch (HmsClientException | UnsupportedOperationException e) {
LOG.warn("Failed to prune Hive partitions through HMS filter for {}.{} with filter '{}', "
+ "falling back to local partition pruning",
hiveHandle.getDbName(), hiveHandle.getTableName(), hmsFilter, e);
}
}

List<String> allPartNames = hmsClient.listPartitionNames(
hiveHandle.getDbName(), hiveHandle.getTableName(), -1);
List<String> matchedPartNames = prunePartitionNames(
allPartNames, partKeyNames, partitionPredicates);
if (matchedPartNames.size() == allPartNames.size()) {
return null;
}
HmsPartitionBatchResult batchResult;
try {
batchResult = matchedPartNames.isEmpty() ? null : hmsClient.getExistingPartitionsWithStats(
hiveHandle.getDbName(), hiveHandle.getTableName(), matchedPartNames);
} catch (HmsClientException e) {
if (e.getPartitionBatchStats() != null) {
HiveScanPlanProvider.recordPruningFailure(
session, hiveHandle.getDbName(), hiveHandle.getTableName(), e.getPartitionBatchStats());
}
throw e;
}
List<HmsPartitionInfo> prunedPartitions = batchResult == null
? Collections.emptyList() : batchResult.getPartitions();
LOG.info("Partition pruning through local partition names: {}.{} all={} pruned={}",
hiveHandle.getDbName(), hiveHandle.getTableName(), allPartNames.size(), prunedPartitions.size());
return new PartitionPruningResult(prunedPartitions, batchResult == null ? null : batchResult.getStats());
}

/** Connector-filtered partitions and optional HMS batch telemetry for the fallback path. */
private static final class PartitionPruningResult {
private final List<HmsPartitionInfo> partitions;
private final HmsPartitionBatchStats batchStats;

private PartitionPruningResult(List<HmsPartitionInfo> partitions, HmsPartitionBatchStats batchStats) {
this.partitions = partitions;
this.batchStats = batchStats;
}
}

private static List<ConnectorPartitionInfo> toConnectorPartitionInfos(List<HmsPartitionInfo> partitions,
List<String> partKeyNames) {
List<ConnectorPartitionInfo> result = new ArrayList<>(partitions.size());
for (HmsPartitionInfo partition : partitions) {
List<String> values = partition.getValues();
result.add(new ConnectorPartitionInfo(HiveWriteUtils.makePartName(partKeyNames, values),
toPartitionValueMap(values, partKeyNames), Collections.emptyMap(), values,
toPartitionValueNullFlags(values)));
}
return result;
}

/**
* Per-value SQL-NULL flags for the ordered partition values (as produced by
* {@link HiveWriteUtils#toPartitionValues}), positionally aligned so flag {@code i} zips to value {@code i}
Expand Down Expand Up @@ -1339,6 +1398,10 @@ private List<String> collectPartitionNames(HiveTableHandle handle, boolean bypas
*/
private static Map<String, String> toPartitionValueMap(String partitionName, List<String> partKeyNames) {
List<String> values = HiveWriteUtils.toPartitionValues(partitionName);
return toPartitionValueMap(values, partKeyNames);
}

private static Map<String, String> toPartitionValueMap(List<String> values, List<String> partKeyNames) {
if (partKeyNames == null || values.size() != partKeyNames.size()) {
return Collections.emptyMap();
}
Expand Down Expand Up @@ -2492,6 +2555,99 @@ private List<String> prunePartitionNames(List<String> allPartNames,
return matched;
}

private static String buildHmsPartitionFilter(List<String> partKeyNames, Map<String, String> partKeyTypes,
Map<String, List<String>> partitionPredicates) {
List<String> filters = new ArrayList<>();
for (String partKeyName : partKeyNames) {
List<String> values = partitionPredicates.get(partKeyName);
if (values == null || values.isEmpty()) {
continue;
}
if (!isHmsFilterIdentifier(partKeyName)) {
return null;
}
List<String> valueFilters = new ArrayList<>();
for (String value : values) {
String literal = toHmsFilterLiteral(value, partKeyTypes.get(partKeyName));
if (literal == null) {
return null;
}
valueFilters.add(partKeyName + " = " + literal);
}
filters.add(valueFilters.size() == 1 ? valueFilters.get(0)
: "(" + String.join(" OR ", valueFilters) + ")");
}
return filters.isEmpty() ? null : "(" + String.join(" AND ", filters) + ")";
}

private static boolean isHmsFilterIdentifier(String value) {
if (value.isEmpty() || !isHmsFilterLetterOrDigit(value.charAt(0))) {
return false;
}
for (int index = 1; index < value.length(); index++) {
char character = value.charAt(index);
if (!isHmsFilterLetterOrDigit(character) && character != '_') {
return false;
}
}
return true;
}

private static boolean isHmsFilterLetterOrDigit(char value) {
return value >= 'a' && value <= 'z'
|| value >= 'A' && value <= 'Z'
|| value >= '0' && value <= '9';
}

private static String toHmsFilterLiteral(String value, String typeName) {
if (isHmsIntegralType(typeName)) {
return isIntegralLiteral(value) ? value : null;
}
if (typeName != null && !isHmsStringType(typeName)) {
return null;
}
if (value.indexOf('\\') >= 0 || value.indexOf('\'') >= 0) {
return null;
}
return "'" + value + "'";
Comment thread
zhaorongsheng marked this conversation as resolved.
}

private static boolean isHmsIntegralType(String typeName) {
if (typeName == null) {
return false;
}
switch (typeName.toUpperCase(Locale.ROOT)) {
case "TINYINT":
case "SMALLINT":
case "INT":
case "INTEGER":
case "BIGINT":
return true;
default:
return false;
}
}

private static boolean isHmsStringType(String typeName) {
String upperTypeName = typeName.toUpperCase(Locale.ROOT);
return "STRING".equals(upperTypeName) || "VARCHAR".equals(upperTypeName)
|| "CHAR".equals(upperTypeName);
}

private static boolean isIntegralLiteral(String value) {
int start = value.startsWith("-") ? 1 : 0;
if (start == value.length()) {
return false;
}
for (int index = start; index < value.length(); index++) {
char character = value.charAt(index);
if (character < '0' || character > '9') {
return false;
}
}
return true;
}

static Map<String, String> parsePartitionName(String partName,
List<String> partKeyNames) {
Map<String, String> values = new HashMap<>();
Expand Down
16 changes: 16 additions & 0 deletions ...ctor/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveTableHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
Expand Down Expand Up @@ -51,6 +52,7 @@ public class HiveTableHandle implements ConnectorTableHandle {
private final String serializationLib;
private final String location;
private final List<String> partitionKeyNames;
private final Map<String, String> partitionKeyTypes;
private final Map<String, String> sdParameters;
private final Map<String, String> tableParameters;
// Whether the table's first column is a STRING, precomputed at handle build time (the metastore table is
Expand All @@ -72,6 +74,9 @@ private HiveTableHandle(Builder builder) {
this.partitionKeyNames = builder.partitionKeyNames != null
? Collections.unmodifiableList(builder.partitionKeyNames)
: Collections.emptyList();
this.partitionKeyTypes = builder.partitionKeyTypes != null
? Collections.unmodifiableMap(new HashMap<>(builder.partitionKeyTypes))
: Collections.emptyMap();
this.sdParameters = builder.sdParameters != null
? Collections.unmodifiableMap(builder.sdParameters)
: Collections.emptyMap();
Expand Down Expand Up @@ -116,6 +121,10 @@ public List<String> getPartitionKeyNames() {
return partitionKeyNames;
}

public Map<String, String> getPartitionKeyTypes() {
return partitionKeyTypes;
}

public Map<String, String> getSdParameters() {
return sdParameters;
}
Expand Down Expand Up @@ -182,6 +191,7 @@ public Builder toBuilder() {
b.serializationLib = this.serializationLib;
b.location = this.location;
b.partitionKeyNames = this.partitionKeyNames;
b.partitionKeyTypes = this.partitionKeyTypes;
b.sdParameters = this.sdParameters;
b.tableParameters = this.tableParameters;
b.firstColumnIsString = this.firstColumnIsString;
Expand All @@ -206,6 +216,7 @@ public static final class Builder {
private String serializationLib;
private String location;
private List<String> partitionKeyNames;
private Map<String, String> partitionKeyTypes;
private Map<String, String> sdParameters;
private Map<String, String> tableParameters;
private boolean firstColumnIsString;
Expand Down Expand Up @@ -238,6 +249,11 @@ public Builder partitionKeyNames(List<String> val) {
return this;
}

public Builder partitionKeyTypes(Map<String, String> val) {
this.partitionKeyTypes = val;
return this;
}

public Builder sdParameters(Map<String, String> val) {
this.sdParameters = val;
return this;
Expand Down
Loading
Loading