From f601804000e0f6def70097a025b2b6c76f81e428 Mon Sep 17 00:00:00 2001 From: zhaorongsheng Date: Wed, 9 Sep 2026 15:34:33 +0800 Subject: [PATCH] [improvement](hive) Push partition filters to HMS ### What problem does this PR solve? Issue Number: close #67724 Related PR: #67739 Problem Summary: Hive partition pruning could enumerate every HMS partition before a selective predicate reached the connector. This change keeps the plain-Hive latest snapshot lightweight, materializes a remote partition view only after the connector accepts the predicate, and retains full-list local pruning as the compatibility fallback. Consumers that require a complete partition view, including MTMV alignment and no-filter scan finalization, now rematerialize it instead of interpreting the lightweight pin as an empty table. Filter responses are bounded to prevent an oversized Thrift reply, and connector-filtered handles avoid batch name reconstruction so they keep the original HMS partition identity. ### Release note Improve planning latency for selective Hive partition queries when HMS supports get_partitions_by_filter. ### Check List (For Author) - Test: Unit Test - PluginDrivenMvccExternalTableTest (70 tests) - DISABLE_BUILD_UI=ON ./build.sh --fe - git diff --check - Behavior changed: Yes (selective Hive partition predicates use a connector-filtered view, while unsupported or oversized filters retain a safe full-list fallback) - Does this need documentation: No --- .../connector/hive/HiveConnectorMetadata.java | 234 +++++++++++++++--- .../connector/hive/HiveScanPlanProvider.java | 6 +- .../doris/connector/hive/HiveTableHandle.java | 16 ++ .../doris/connector/hive/HiveWriteUtils.java | 40 +++ ...ConnectorMetadataPartitionPruningTest.java | 108 +++++++- .../connector/hive/HiveScanBatchModeTest.java | 11 + .../doris/connector/hms/CachingHmsClient.java | 7 + .../apache/doris/connector/hms/HmsClient.java | 16 ++ .../doris/connector/hms/ThriftHmsClient.java | 16 ++ .../hive/metastore/HiveMetaStoreClient.java | 6 +- .../hms/ThriftHmsClientMaxPartsTest.java | 43 ++++ .../connector/spi/ConnectorCapability.java | 10 + .../spi/ConnectorPluginSurfaceTest.java | 6 +- .../resources/connector-plugin-surface.txt | 1 + fe/fe-connector/pom.xml | 2 +- .../mvcc/PluginDrivenMvccExternalTable.java | 37 ++- .../plugin/PluginDrivenExternalTable.java | 77 +++++- .../doris/datasource/scan/FileScanNode.java | 4 +- .../datasource/scan/PluginDrivenScanNode.java | 33 ++- .../rules/rewrite/PruneFileScanPartition.java | 49 +++- .../rewrite/QueryPartitionCollector.java | 10 +- .../trees/plans/logical/LogicalFileScan.java | 38 ++- .../PluginDrivenMvccExternalTableTest.java | 66 ++++- ...luginDrivenExternalTablePartitionTest.java | 30 +++ .../PluginDrivenScanNodeBatchModeTest.java | 15 ++ ...luginDrivenScanNodePartitionCountTest.java | 8 + ...ginDrivenScanNodePartitionPruningTest.java | 11 + 27 files changed, 818 insertions(+), 82 deletions(-) diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java index b933e3da1bd760..66937bf360e84c 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java @@ -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; @@ -472,11 +473,16 @@ public Optional getTableHandle( // Build partition key column names List partKeyNames = Collections.emptyList(); + Map partKeyTypes = Collections.emptyMap(); List 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) @@ -484,6 +490,7 @@ public Optional getTableHandle( .serializationLib(tableInfo.getSerializationLib()) .location(tableInfo.getLocation()) .partitionKeyNames(partKeyNames) + .partitionKeyTypes(partKeyTypes) .sdParameters(tableInfo.getSdParameters()) .tableParameters(tableInfo.getParameters()) .firstColumnIsString(firstColumnIsString(tableInfo)) @@ -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); } @@ -1168,46 +1178,14 @@ public Optional> applyFilter( return Optional.empty(); } - // Extract equality predicates on partition columns from the expression - Map> 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 allPartNames = hmsClient.listPartitionNames( - hiveHandle.getDbName(), hiveHandle.getTableName(), 100000); - List 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 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)); @@ -1234,9 +1212,9 @@ public List 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. * *

{@code lastModifiedMillis} is deliberately left {@link ConnectorPartitionInfo#UNKNOWN} (-1): * reading each partition's {@code transient_lastDdlTime} requires a {@code get_partitions_by_names} @@ -1261,6 +1239,15 @@ public List 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); } @@ -1288,6 +1275,78 @@ private List listPartitionsUncached(HiveTableHandle hive return result; } + private PartitionPruningResult prunePartitions(ConnectorSession session, HiveTableHandle hiveHandle, + ConnectorExpression expression) { + List partKeyNames = hiveHandle.getPartitionKeyNames(); + Map> partitionPredicates = extractPartitionPredicates(expression, partKeyNames); + if (partitionPredicates.isEmpty()) { + return null; + } + + String hmsFilter = buildHmsPartitionFilter(partKeyNames, hiveHandle.getPartitionKeyTypes(), + partitionPredicates); + if (hmsFilter != null) { + try { + List 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 allPartNames = hmsClient.listPartitionNames( + hiveHandle.getDbName(), hiveHandle.getTableName(), -1); + List 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 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 partitions; + private final HmsPartitionBatchStats batchStats; + + private PartitionPruningResult(List partitions, HmsPartitionBatchStats batchStats) { + this.partitions = partitions; + this.batchStats = batchStats; + } + } + + private static List toConnectorPartitionInfos(List partitions, + List partKeyNames) { + List result = new ArrayList<>(partitions.size()); + for (HmsPartitionInfo partition : partitions) { + List 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} @@ -1339,6 +1398,10 @@ private List collectPartitionNames(HiveTableHandle handle, boolean bypas */ private static Map toPartitionValueMap(String partitionName, List partKeyNames) { List values = HiveWriteUtils.toPartitionValues(partitionName); + return toPartitionValueMap(values, partKeyNames); + } + + private static Map toPartitionValueMap(List values, List partKeyNames) { if (partKeyNames == null || values.size() != partKeyNames.size()) { return Collections.emptyMap(); } @@ -2492,6 +2555,99 @@ private List prunePartitionNames(List allPartNames, return matched; } + private static String buildHmsPartitionFilter(List partKeyNames, Map partKeyTypes, + Map> partitionPredicates) { + List filters = new ArrayList<>(); + for (String partKeyName : partKeyNames) { + List values = partitionPredicates.get(partKeyName); + if (values == null || values.isEmpty()) { + continue; + } + if (!isHmsFilterIdentifier(partKeyName)) { + return null; + } + List 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 + "'"; + } + + 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 parsePartitionName(String partName, List partKeyNames) { Map values = new HashMap<>(); diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java index 0660619a399299..81773d2e2d7110 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java @@ -200,7 +200,11 @@ public List planScan(ConnectorSession session, ConnectorScan public boolean supportsBatchScan(ConnectorSession session, ConnectorTableHandle handle) { HiveTableHandle hiveHandle = (HiveTableHandle) handle; List partKeyNames = hiveHandle.getPartitionKeyNames(); - return partKeyNames != null && !partKeyNames.isEmpty() && !hiveHandle.isTransactional(); + // A connector-filtered handle already carries the exact HMS Partition objects. Keeping it on the normal + // scan path avoids reconstructing partition names from key/value pairs and fetching them again per batch, + // which can lose the metastore's locale-sensitive canonical key spelling. + return partKeyNames != null && !partKeyNames.isEmpty() && !hiveHandle.isTransactional() + && hiveHandle.getPrunedPartitions() == null; } /** diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveTableHandle.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveTableHandle.java index c92bc53b8645fc..02aabde6d3033a 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveTableHandle.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveTableHandle.java @@ -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; @@ -51,6 +52,7 @@ public class HiveTableHandle implements ConnectorTableHandle { private final String serializationLib; private final String location; private final List partitionKeyNames; + private final Map partitionKeyTypes; private final Map sdParameters; private final Map tableParameters; // Whether the table's first column is a STRING, precomputed at handle build time (the metastore table is @@ -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(); @@ -116,6 +121,10 @@ public List getPartitionKeyNames() { return partitionKeyNames; } + public Map getPartitionKeyTypes() { + return partitionKeyTypes; + } + public Map getSdParameters() { return sdParameters; } @@ -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; @@ -206,6 +216,7 @@ public static final class Builder { private String serializationLib; private String location; private List partitionKeyNames; + private Map partitionKeyTypes; private Map sdParameters; private Map tableParameters; private boolean firstColumnIsString; @@ -238,6 +249,11 @@ public Builder partitionKeyNames(List val) { return this; } + public Builder partitionKeyTypes(Map val) { + this.partitionKeyTypes = val; + return this; + } + public Builder sdParameters(Map val) { this.sdParameters = val; return this; diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWriteUtils.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWriteUtils.java index 80c5199d7f491a..2a26d17a7b0fff 100644 --- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWriteUtils.java +++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWriteUtils.java @@ -17,6 +17,7 @@ package org.apache.doris.connector.hive; +import org.apache.doris.connector.spi.scan.ConnectorPartitionValues; import org.apache.doris.thrift.THivePartitionUpdate; import org.apache.hadoop.fs.Path; @@ -26,6 +27,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Objects; @@ -203,6 +205,44 @@ static List toPartitionValues(String partitionName) { return result; } + /** Builds a metastore-rendered Hive partition name from declaration-order keys and values. */ + static String makePartName(List partKeys, List values) { + StringBuilder result = new StringBuilder(); + for (int index = 0; index < partKeys.size(); index++) { + if (index != 0) { + result.append('/'); + } + result.append(escapePathName(partKeys.get(index).toLowerCase(Locale.ROOT))) + .append('=') + .append(escapePathName(values.get(index))); + } + return result.toString(); + } + + private static String escapePathName(String path) { + if (path == null || path.isEmpty()) { + return ConnectorPartitionValues.NULL_PARTITION_NAME; + } + StringBuilder result = new StringBuilder(); + for (int index = 0; index < path.length(); index++) { + char character = path.charAt(index); + if (needsPathEscaping(character)) { + result.append('%').append(String.format("%02X", (int) character)); + } else { + result.append(character); + } + } + return result.toString(); + } + + private static boolean needsPathEscaping(char character) { + return character < ' ' + || character == '"' || character == '#' || character == '%' || character == '\'' + || character == '*' || character == '/' || character == ':' || character == '=' + || character == '?' || character == '\\' || character == 0x7F || character == '{' + || character == '[' || character == ']' || character == '^'; + } + /** * URL-decodes a Hive-escaped path component (e.g. "a%2Fb" -> "a/b"). Byte-faithful port of Hive's * {@code org.apache.hadoop.hive.common.FileUtils.unescapePathName}, inlined to avoid a hive-common diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataPartitionPruningTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataPartitionPruningTest.java index 907cc2126d7021..7ea55157e5a7c4 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataPartitionPruningTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataPartitionPruningTest.java @@ -21,6 +21,7 @@ import org.apache.doris.connector.hms.HmsDatabaseInfo; import org.apache.doris.connector.hms.HmsPartitionInfo; import org.apache.doris.connector.hms.HmsTableInfo; +import org.apache.doris.connector.spi.ConnectorPartitionInfo; import org.apache.doris.connector.spi.ConnectorType; import org.apache.doris.connector.spi.handle.ConnectorTableHandle; import org.apache.doris.connector.spi.pushdown.ConnectorAnd; @@ -99,6 +100,87 @@ public void testAndOfTwoPartitionColumnsPrunes() { prunedLocations(result)); } + @Test + public void testHmsFilterPrunesWithoutListingAllPartitionNames() { + HmsFilterClient client = new HmsFilterClient(Collections.singletonList("year=2024/month=01")); + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), new FakeConnectorContext()); + + Optional> result = metadata.applyFilter( + null, partitionedHandle(), new ConnectorFilterConstraint(and(eq("year", "2024"), eq("month", "01")))); + + Assertions.assertTrue(result.isPresent()); + Assertions.assertEquals(Collections.singletonList("year=2024/month=01"), prunedLocations(result)); + Assertions.assertEquals("(year = '2024' AND month = '01')", client.filter); + Assertions.assertFalse(client.wasListPartitionNamesCalled()); + } + + @Test + public void testHmsFilterRendersIntegralPartitionLiteralWithoutQuotes() { + HmsFilterClient client = new HmsFilterClient(Collections.singletonList("year=2024/month=01")); + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), new FakeConnectorContext()); + HiveTableHandle handle = new HiveTableHandle.Builder("db", "t", HiveTableType.HIVE) + .partitionKeyNames(PART_KEYS) + .partitionKeyTypes(Map.of("year", "INT", "month", "STRING")) + .build(); + + Optional> result = metadata.applyFilter( + null, handle, new ConnectorFilterConstraint(and(eq("year", "2024"), eq("month", "01")))); + + Assertions.assertTrue(result.isPresent()); + Assertions.assertEquals("(year = 2024 AND month = '01')", client.filter); + Assertions.assertFalse(client.wasListPartitionNamesCalled()); + } + + @Test + public void testFilteredPartitionViewUsesHmsFilterWithoutListingAllPartitionNames() { + HmsFilterClient client = new HmsFilterClient(Collections.singletonList("year=2024/month=01")); + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), new FakeConnectorContext()); + + List partitions = metadata.listPartitions(null, partitionedHandle(), + Optional.of(and(eq("year", "2024"), eq("month", "01")))); + + Assertions.assertEquals(1, partitions.size()); + Assertions.assertEquals("year=2024/month=01", partitions.get(0).getPartitionName()); + Assertions.assertEquals("(year = '2024' AND month = '01')", client.filter); + Assertions.assertFalse(client.wasListPartitionNamesCalled()); + } + + @Test + public void testPrunedHandleReusesFilteredPartitionView() { + HmsFilterClient client = new HmsFilterClient(Collections.singletonList("year=2024/month=01")); + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), new FakeConnectorContext()); + + Optional> result = metadata.applyFilter( + null, partitionedHandle(), new ConnectorFilterConstraint(and(eq("year", "2024"), eq("month", "01")))); + Assertions.assertTrue(result.isPresent()); + + List partitions = metadata.listPartitions(null, result.get().getHandle(), + Optional.empty()); + + Assertions.assertEquals(1, partitions.size()); + Assertions.assertEquals("year=2024/month=01", partitions.get(0).getPartitionName()); + Assertions.assertFalse(client.wasListPartitionNamesCalled()); + } + + @Test + public void testUnsupportedHmsFilterFallsBackToLocalPruning() { + FakeHmsClient client = new FakeHmsClient(PARTITIONS); + HiveConnectorMetadata metadata = new HiveConnectorMetadata( + client, HiveTestProperties.minimal(), new FakeConnectorContext()); + + Optional> result = metadata.applyFilter( + null, partitionedHandle(), new ConnectorFilterConstraint(eq("year", "2024"))); + + Assertions.assertTrue(result.isPresent()); + Assertions.assertEquals( + Arrays.asList("year=2024/month=01", "year=2024/month=02"), prunedLocations(result)); + Assertions.assertTrue(client.wasListPartitionNamesCalled()); + } + @Test public void testNonPartitionColumnInAndIsIgnored() { Optional> result = @@ -281,8 +363,9 @@ private static ConnectorAnd and(ConnectorExpression... children) { * whose location IS the partition name (so the pruning selection can be asserted). * The rest fail loud. */ - private static final class FakeHmsClient implements HmsClient { + private static class FakeHmsClient implements HmsClient { private final List partitionNames; + private boolean listPartitionNamesCalled; FakeHmsClient(List partitionNames) { this.partitionNames = partitionNames; @@ -290,15 +373,20 @@ private static final class FakeHmsClient implements HmsClient { @Override public List listPartitionNames(String dbName, String tableName, int maxParts) { + listPartitionNamesCalled = true; return partitionNames; } + boolean wasListPartitionNamesCalled() { + return listPartitionNamesCalled; + } + @Override public List getPartitions(String dbName, String tableName, List partNames) { List result = new ArrayList<>(); for (String name : partNames) { - result.add(new HmsPartitionInfo(Collections.emptyList(), name, + result.add(new HmsPartitionInfo(HiveWriteUtils.toPartitionValues(name), name, null, null, null, Collections.emptyMap())); } return result; @@ -343,4 +431,20 @@ public HmsPartitionInfo getPartition(String dbName, String tableName, List filteredPartitionNames; + private String filter; + + HmsFilterClient(List filteredPartitionNames) { + super(Collections.emptyList()); + this.filteredPartitionNames = filteredPartitionNames; + } + + @Override + public List listPartitionsByFilter(String dbName, String tableName, String filter) { + this.filter = filter; + return getPartitions(dbName, tableName, filteredPartitionNames); + } + } } diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java index 720d9b91066ab8..03269500ada3da 100644 --- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java +++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java @@ -115,6 +115,17 @@ public void supportsBatchScanIsFalseForTransactionalPartitionedTable() { Assertions.assertFalse(provider.supportsBatchScan(new FakeSession(), handle)); } + @Test + public void supportsBatchScanIsFalseForConnectorFilteredTable() { + HiveScanPlanProvider provider = provider(null, new CountingLister()); + HiveTableHandle handle = new HiveTableHandle.Builder("db", "t", HiveTableType.HIVE) + .partitionKeyNames(PART_KEYS) + .prunedPartitions(Collections.singletonList(part("year=2024/month=01"))) + .build(); + + Assertions.assertFalse(provider.supportsBatchScan(new FakeSession(), handle)); + } + // ==================== planScanForPartitionBatch: scoped to the batch, no duplication ==================== @Test diff --git a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java index 0b005b746bcac9..1efe27b1056a79 100644 --- a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java +++ b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java @@ -208,6 +208,13 @@ public List listPartitionNamesFresh(String dbName, String tableName, int return delegate.listPartitionNames(dbName, tableName, maxParts); } + @Override + public List listPartitionsByFilter(String dbName, String tableName, String filter) { + // A filter result has no bounded cache key and must reflect the predicate sent by the current query. + // Do not read or populate the full partition-name cache here. + return delegate.listPartitionsByFilter(dbName, tableName, filter); + } + @Override public List getPartitions(String dbName, String tableName, List partNames) { return getPartitionsWithStats(dbName, tableName, partNames).getPartitions(); diff --git a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsClient.java b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsClient.java index ff81a8cfee9be2..9bfbbac9f3319a 100644 --- a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsClient.java +++ b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsClient.java @@ -152,6 +152,22 @@ default List listPartitionNamesFresh(String dbName, String tableName, in return listPartitionNames(dbName, tableName, maxParts); } + /** + * Lists partitions matching an HMS filter expression. + * + *

This optional operation is used by selective scan planning to avoid enumerating every partition name + * before FE-side pruning. Implementations that do not support the metastore filter dialect keep the default + * and callers fall back to {@link #listPartitionNames(String, String, int)}.

+ * + * @param dbName database name + * @param tableName table name + * @param filter HMS {@code get_partitions_by_filter} expression + * @return matching partition metadata + */ + default List listPartitionsByFilter(String dbName, String tableName, String filter) { + throw new UnsupportedOperationException("listPartitionsByFilter is not supported by this client"); + } + /** * Get partition metadata by partition names. * diff --git a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java index 8a226c0d3d6dee..c69e490ac32a74 100644 --- a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java +++ b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java @@ -98,6 +98,7 @@ public class ThriftHmsClient implements HmsClient { private static final HiveMetaHookLoader DUMMY_HOOK_LOADER = tbl -> null; private static final long POOL_BORROW_TIMEOUT_MS = 60_000L; private static final int ADD_PARTITIONS_BATCH_SIZE = 20; + private static final int MAX_FILTERED_PARTITIONS = HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE; private static final String TRANSIENT_LAST_DDL_TIME = "transient_lastDdlTime"; private final HiveConf hiveConf; @@ -239,6 +240,21 @@ static short toThriftMaxParts(int maxParts) { return maxParts <= 0 ? (short) -1 : (short) maxParts; } + @Override + public List listPartitionsByFilter(String dbName, String tableName, String filter) { + List partitions = execute(client -> client.listPartitionsByFilter( + dbName, tableName, filter, (short) (MAX_FILTERED_PARTITIONS + 1))); + if (isFilteredPartitionResponseSaturated(partitions.size())) { + throw new HmsClientException("HMS partition filter matched more than " + + MAX_FILTERED_PARTITIONS + " partitions"); + } + return partitions.stream().map(ThriftHmsClient::convertPartition).collect(Collectors.toList()); + } + + static boolean isFilteredPartitionResponseSaturated(int partitionCount) { + return partitionCount > MAX_FILTERED_PARTITIONS; + } + @Override public List getPartitions(String dbName, String tableName, List partNames) { diff --git a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java index ca5113a57d8dee..bb80dbc83327ee 100644 --- a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java +++ b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java @@ -1701,8 +1701,10 @@ public List listPartitionsByFilter(String db_name, String tbl_name, @Override public List listPartitionsByFilter(String catName, String db_name, String tbl_name, String filter, int max_parts) throws TException { - List parts =client.get_partitions_by_filter(prependCatalogToDbName( - catName, db_name, conf), tbl_name, filter, shrinkMaxtoShort(max_parts)); + String databaseName = hiveVersion == HiveVersion.V1_0 || hiveVersion == HiveVersion.V2_0 + || hiveVersion == HiveVersion.V2_3 ? db_name : prependCatalogToDbName(catName, db_name, conf); + List parts =client.get_partitions_by_filter( + databaseName, tbl_name, filter, shrinkMaxtoShort(max_parts)); return deepCopyPartitions(filterHook.filterPartitions(parts)); } diff --git a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientMaxPartsTest.java b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientMaxPartsTest.java index 2231451a48527e..54ad36fe10ad74 100644 --- a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientMaxPartsTest.java +++ b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/ThriftHmsClientMaxPartsTest.java @@ -17,9 +17,15 @@ package org.apache.doris.connector.hms; +import org.apache.hadoop.hive.metastore.IMetaStoreClient; +import org.apache.hadoop.hive.metastore.api.Partition; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Collections; + /** * Tests {@link ThriftHmsClient#toThriftMaxParts}: the connector's {@code maxParts} contract mapped onto the * {@code short max_parts} that HMS {@code get_partition_names} accepts. @@ -56,4 +62,41 @@ public void testPositiveAboveShortNarrowsToUnbounded() { Assertions.assertEquals((short) 100000, mapped); Assertions.assertTrue(mapped < 0, "a value above Short.MAX_VALUE must narrow to a negative (unbounded) short"); } + + @Test + public void testFilteredPartitionResponseIsBounded() { + Assertions.assertFalse(ThriftHmsClient.isFilteredPartitionResponseSaturated( + HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE)); + Assertions.assertTrue(ThriftHmsClient.isFilteredPartitionResponseSaturated( + HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE + 1)); + } + + @Test + public void testFilteredPartitionResponseFallsBackWhenSaturated() throws Exception { + IMetaStoreClient metastore = (IMetaStoreClient) Proxy.newProxyInstance( + getClass().getClassLoader(), new Class[] {IMetaStoreClient.class}, (proxy, method, args) -> { + if ("listPartitionsByFilter".equals(method.getName())) { + Assertions.assertEquals("db", args[0]); + Assertions.assertEquals("tbl", args[1]); + Assertions.assertEquals("p=1", args[2]); + Assertions.assertEquals((short) (HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE + 1), args[3]); + return new ArrayList<>(Collections.nCopies( + HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE + 1, new Partition())); + } + return null; + }); + ThriftHmsClient client = new ThriftHmsClient(new HmsClientConfig(Collections.emptyMap(), 0), + new ThriftHmsClient.AuthAction() { + @Override + public T execute(java.util.concurrent.Callable callable) throws Exception { + return callable.call(); + } + }, hiveConf -> metastore, HmsTypeMapping.Options.DEFAULT); + try { + Assertions.assertThrows(HmsClientException.class, + () -> client.listPartitionsByFilter("db", "tbl", "p=1")); + } finally { + client.close(); + } + } } diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorCapability.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorCapability.java index a5d1dd67b2eba9..e6f10b7a8ff285 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorCapability.java +++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorCapability.java @@ -167,6 +167,16 @@ public enum ConnectorCapability { * whose scan path supports storage-level predicate pruning.

*/ SUPPORTS_STORAGE_PREDICATE_PRUNING, + /** + * Indicates that the connector can materialize a partitioned table's selected partition view directly from + * a connector predicate during Nereids partition pruning. The engine defers eager full partition + * materialization for such tables and asks the connector for the filtered view instead; when the connector + * cannot apply a predicate it must retain its existing full-list fallback. + * + *

Scope: per-table only. A heterogeneous connector such as Hive can support this for plain HMS + * tables while delegating sibling table formats to connectors with different partition semantics.

+ */ + SUPPORTS_CONNECTOR_PARTITION_PRUNING, /** * Indicates the connector's external metadata (schema / partitions / snapshot) can be pre-warmed * asynchronously by the planner before it takes the internal read lock, rather than loaded lazily diff --git a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java index 9b4cd0907243a4..1104ca590e2454 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java +++ b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java @@ -81,9 +81,9 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange() throws IOException Assertions.assertNotNull(in, "missing connector plugin API version resource"); version.load(in); } - // Storage predicate pruning and provider-level DDL validation both changed the public surface in - // major 7. An older FE must reject plugins using either addition before linking incompatible bytecode. - Assertions.assertEquals("7.0", version.getProperty("api.version")); + // Connector partition pruning changed the public surface in major 8. An older FE must reject plugins + // using the added capability before linking incompatible bytecode. + Assertions.assertEquals("8.0", version.getProperty("api.version")); } /** Root entry points plus provider/handle types returned to connector plugins. */ diff --git a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt index e46d06b20cb151..c2a5dc5f3dff55 100644 --- a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt +++ b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt @@ -20,6 +20,7 @@ org.apache.doris.connector.spi.Connector#preCreateValidation(org.apache.doris.co org.apache.doris.connector.spi.Connector#schemaCacheTtlSecondOverride():java.util.OptionalLong org.apache.doris.connector.spi.Connector#testConnection(org.apache.doris.connector.spi.ConnectorSession):org.apache.doris.connector.spi.ConnectorTestResult org.apache.doris.connector.spi.ConnectorCapability#enum:SUPPORTS_COLUMN_AUTO_ANALYZE +org.apache.doris.connector.spi.ConnectorCapability#enum:SUPPORTS_CONNECTOR_PARTITION_PRUNING org.apache.doris.connector.spi.ConnectorCapability#enum:SUPPORTS_METADATA_PRELOAD org.apache.doris.connector.spi.ConnectorCapability#enum:SUPPORTS_MVCC_SNAPSHOT org.apache.doris.connector.spi.ConnectorCapability#enum:SUPPORTS_NESTED_COLUMN_PRUNE diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml index e217ff5cf2790f..f744982603411a 100644 --- a/fe/fe-connector/pom.xml +++ b/fe/fe-connector/pom.xml @@ -55,7 +55,7 @@ under the License. of the latter two means bumping this property as well (and fe-extension-spi means bumping all five families). --> - 7.0 + 8.0 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java index d4223f9f7c60d2..919470516510cf 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java @@ -65,6 +65,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -109,6 +110,11 @@ public PluginDrivenMvccExternalTable(long id, String name, String remoteName, super(id, name, remoteName, catalog, db); } + @Override + public boolean supportsLatestSnapshotPreload() { + return supportsConnectorPartitionPruning(); + } + // ──────────────────── snapshot materialization ──────────────────── /** @@ -166,6 +172,16 @@ private PluginDrivenMvccSnapshot materializeLatest( ConnectorMvccSnapshot connectorSnapshot = existingFence.orElseGet( () -> metadata.beginQuerySnapshot(session, handle).orElseGet(this::emptySnapshot)); + // A connector that can materialize a partition predicate remotely must not populate the latest + // snapshot with every partition before Nereids has supplied that predicate. Plain Hive reaches this + // MVCC table class because the catalog also serves snapshot-capable sibling formats, but its latest + // pin is deliberately not a data snapshot and applySnapshot is a no-op. Keep only that lightweight + // query-begin pin here; PruneFileScanPartition will request the selected partition view later. + if (supportsConnectorPartitionPruning()) { + return new PluginDrivenMvccSnapshot(connectorSnapshot, + Collections.emptyMap(), Collections.emptyMap()); + } + // Range-view path (e.g. iceberg): thread the query's pin onto the handle FIRST (applySnapshot), so // the partition/freshness enumeration stays consistent with the data-scan pin, then ask the connector // for its range-aware view. A connector without a range view returns empty -> fall through to the @@ -671,6 +687,13 @@ static boolean schemaCacheDisabled(Connector connector) { @Override public Map getNameToPartitionItems(Optional snapshot) { + if (supportsConnectorPartitionPruning()) { + // The latest Hive query pin intentionally carries no partition map so selective scans can send a + // predicate to HMS first. Consumers that explicitly ask for a partition map (MTMV alignment, + // no-filter scan finalization, and a connector-declined pruning fallback) require the real full + // view instead of treating that query-only pin as an empty table. + return super.getNameToPartitionItems(snapshot); + } return getOrMaterialize(snapshot).getNameToPartitionItem(); } @@ -769,6 +792,13 @@ private Optional pinnedSnapshot(CatalogRelation scan) { public MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshContext context, Optional snapshot) throws AnalysisException { PluginDrivenMvccSnapshot pin = getOrMaterialize(snapshot); + if (supportsConnectorPartitionPruning() && pin.getConnectorSnapshot().isLastModifiedFreshness()) { + OptionalLong onDemand = queryPartitionFreshnessMillis(partitionName); + if (!onDemand.isPresent()) { + throw new AnalysisException("can not find partition: " + partitionName); + } + return new MTMVTimestampSnapshot(onDemand.getAsLong()); + } Long value = pin.getNameToLastModifiedMillis().get(partitionName); if (value == null) { throw new AnalysisException("can not find partition: " + partitionName); @@ -804,9 +834,10 @@ public Map getPartitionSnapshots(Set partitionNa PluginDrivenMvccSnapshot pin = getOrMaterialize(snapshot); Map snapshots = new LinkedHashMap<>(); if (pin.getConnectorSnapshot().isLastModifiedFreshness()) { - List existingPartitionNames = partitionNames.stream() - .filter(pin.getNameToLastModifiedMillis()::containsKey) - .collect(Collectors.toList()); + List existingPartitionNames = supportsConnectorPartitionPruning() + ? new ArrayList<>(partitionNames) + : partitionNames.stream().filter(pin.getNameToLastModifiedMillis()::containsKey) + .collect(Collectors.toList()); Map freshness = queryPartitionFreshnessMillis(existingPartitionNames); for (String partitionName : partitionNames) { Long value = freshness.get(partitionName); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java index dd0bc658e86330..2668f2c5ec2c2d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java @@ -39,6 +39,8 @@ import org.apache.doris.connector.spi.handle.WriteOperation; import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.connector.spi.pushdown.ConnectorExpression; +import org.apache.doris.connector.spi.pushdown.ConnectorFilterConstraint; +import org.apache.doris.connector.spi.pushdown.FilterApplicationResult; import org.apache.doris.connector.spi.write.ConnectorWritePlanProvider; import org.apache.doris.datasource.ExternalCatalog; import org.apache.doris.datasource.ExternalDatabase; @@ -51,6 +53,7 @@ import org.apache.doris.datasource.systable.PartitionsSysTable; import org.apache.doris.datasource.systable.PluginDrivenSysTable; import org.apache.doris.datasource.systable.SysTable; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.GlobalVariable; import org.apache.doris.statistics.analysis.AnalysisInfo; @@ -949,17 +952,57 @@ public boolean supportInternalPartitionPruned() { return true; } + @Override + public SelectedPartitions initSelectedPartitions(Optional snapshot) { + if (supportsConnectorPartitionPruning()) { + return SelectedPartitions.DEFERRED_PARTITION_PRUNING; + } + return super.initSelectedPartitions(snapshot); + } + + /** Whether this table defers partition materialization until Nereids supplies a connector predicate. */ + public boolean supportsConnectorPartitionPruning() { + return hasCapability(ConnectorCapability.SUPPORTS_CONNECTOR_PARTITION_PRUNING); + } + @Override public Map getNameToPartitionItems(Optional snapshot) { + return getNameToPartitionItems(snapshot, Optional.empty()); + } + + /** + * Builds the generic partition map from a connector-filtered partition view. Callers use this only after + * converting a Nereids predicate into the neutral connector expression grammar. + */ + public Optional> getNameToPartitionItemsByFilter(Optional snapshot, + ConnectorExpression partitionFilter) { List partitionColumns = getPartitionColumns(snapshot); if (partitionColumns.isEmpty()) { - return Collections.emptyMap(); + return Optional.empty(); } - List remoteNames = getSchemaCacheValue(snapshot) - .map(value -> ((PluginDrivenSchemaCacheValue) value).getPartitionColumnRemoteNames()) - .orElse(Collections.emptyList()); - List types = partitionColumns.stream().map(Column::getType).collect(Collectors.toList()); + PluginDrivenExternalCatalog pluginCatalog = (PluginDrivenExternalCatalog) catalog; + Connector connector = pluginCatalog.getConnector(); + ConnectorSession session = pluginCatalog.buildConnectorSession(); + ConnectorMetadata metadata = PluginDrivenMetadata.get(session, connector); + Optional handleOpt = resolveConnectorTableHandle(session, metadata); + if (!handleOpt.isPresent()) { + return Optional.empty(); + } + Optional> filterResult = metadata.applyFilter( + session, handleOpt.get(), new ConnectorFilterConstraint(partitionFilter)); + if (!filterResult.isPresent()) { + return Optional.empty(); + } + return Optional.of(buildNameToPartitionItems(snapshot, metadata, session, + filterResult.get().getHandle(), partitionColumns)); + } + private Map getNameToPartitionItems(Optional snapshot, + Optional partitionFilter) { + List partitionColumns = getPartitionColumns(snapshot); + if (partitionColumns.isEmpty()) { + return Collections.emptyMap(); + } PluginDrivenExternalCatalog pluginCatalog = (PluginDrivenExternalCatalog) catalog; Connector connector = pluginCatalog.getConnector(); ConnectorSession session = pluginCatalog.buildConnectorSession(); @@ -969,12 +1012,24 @@ public Map getNameToPartitionItems(Optional return Collections.emptyMap(); } - // One round-trip, no FE-side partition-value cache (per CACHE-P1: the cutover lists - // partitions per query instead of maintaining a second-level cache). The connector returns - // each partition's display name plus a raw-keyed value map; we extract values in - // partition-column order via the cached remote names. - List partitions = - metadata.listPartitions(session, handleOpt.get(), Optional.empty()); + return buildNameToPartitionItems(snapshot, metadata, session, handleOpt.get(), partitionColumns, + partitionFilter); + } + + private Map buildNameToPartitionItems(Optional snapshot, + ConnectorMetadata metadata, ConnectorSession session, ConnectorTableHandle handle, + List partitionColumns) { + return buildNameToPartitionItems(snapshot, metadata, session, handle, partitionColumns, Optional.empty()); + } + + private Map buildNameToPartitionItems(Optional snapshot, + ConnectorMetadata metadata, ConnectorSession session, ConnectorTableHandle handle, + List partitionColumns, Optional partitionFilter) { + List remoteNames = getSchemaCacheValue(snapshot) + .map(value -> ((PluginDrivenSchemaCacheValue) value).getPartitionColumnRemoteNames()) + .orElse(Collections.emptyList()); + List types = partitionColumns.stream().map(Column::getType).collect(Collectors.toList()); + List partitions = metadata.listPartitions(session, handle, partitionFilter); List partitionNames = new ArrayList<>(partitions.size()); List> partitionValues = new ArrayList<>(partitions.size()); for (ConnectorPartitionInfo partition : partitions) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanNode.java index 6874d929b67a0a..37422f07b64297 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanNode.java @@ -162,8 +162,8 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) { } output.append("inputSplitNum=").append(selectedSplitNum).append(", totalFileSize=") .append(totalFileSize).append(", scanRanges=").append(scanRangeLocations.size()).append("\n"); - output.append(prefix).append("partition=").append(selectedPartitionNum).append("/").append(totalPartitionNum) - .append("\n"); + output.append(prefix).append("partition=").append(selectedPartitionNum).append("/") + .append(totalPartitionNum < 0 ? "?" : totalPartitionNum).append("\n"); if (detailLevel == TExplainLevel.VERBOSE && !isBatch) { appendBackendScanRangeDetail(output, prefix); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java index 3229689f0175d4..063d75018e98a6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java @@ -30,6 +30,7 @@ import org.apache.doris.catalog.Column; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.MapType; +import org.apache.doris.catalog.PartitionItem; import org.apache.doris.catalog.StructType; import org.apache.doris.catalog.TableIf; import org.apache.doris.catalog.Type; @@ -425,7 +426,8 @@ static List resolveRequiredPartitions(SelectedPartitions selectedPartiti // MaxCompute table that genuinely has ZERO partitions this scan-all is row-equivalent to legacy's // unconditional empty short-circuit — MaxComputeScanPlanProvider.planScan returns no splits when // getFileNum() <= 0, still zero rows. - if (selectedPartitions.selectedPartitions.isEmpty() && selectedPartitions.totalPartitionNum == 0) { + if (selectedPartitions.selectedPartitions.isEmpty() && selectedPartitions.totalPartitionNum == 0 + && !selectedPartitions.hasPartitionPredicate) { return null; } // A predicate-driven connector re-plans through its SDK with the pushed predicate (its planScan @@ -457,7 +459,7 @@ static List resolveRequiredPartitions(SelectedPartitions selectedPartiti * must read ALL partitions, so it pushes no partition restriction.

*/ static long[] displayPartitionCounts(SelectedPartitions selectedPartitions) { - if (selectedPartitions == null || selectedPartitions == SelectedPartitions.NOT_PRUNED) { + if (selectedPartitions == null || selectedPartitions.isNotPruned()) { return null; } return new long[] { @@ -659,7 +661,7 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) { // line itself; the counts are populated from the Nereids pruning result in // getSplits()/startSplit() (see setSelectedPartitions). output.append(prefix).append("partition=").append(selectedPartitionNum) - .append("/").append(totalPartitionNum).append("\n"); + .append("/").append(totalPartitionNum < 0 ? "?" : totalPartitionNum).append("\n"); // FIX-E / FIX-R3-RESIDUAL (explain gap): the VERBOSE per-backend block (the backends: list, // per-file "path start/length" lines, and dataFileNum/deleteFileNum/deleteSplitNum) lives in // the parent FileScanNode but this override does not call super, so re-emit it under the SAME @@ -1143,12 +1145,35 @@ protected TFileAttributes getFileAttributes() throws UserException { protected void doFinalize() throws UserException { scanNodeProperties = null; cachedPropertiesResult = null; + materializeDeferredSelectedPartitions(); // Nereids prunes scan slots between init and finalize; fencing the init-time table-wide // tuple would reject old backends even when the executable scan no longer carries Variant. checkVariantBackendCompatibilityForCurrentScan(backendPolicy.getBackends()); super.doFinalize(); } + private void materializeDeferredSelectedPartitions() throws UserException { + if (!selectedPartitions.isDeferredPartitionPruning()) { + return; + } + // A logical filter materializes this state earlier in PruneFileScanPartition. Reaching finalize still + // deferred therefore means a no-filter full scan, which must recover the complete map before the + // batch-mode gate so it keeps the legacy asynchronous split-generation path. + PluginDrivenExternalTable table = (PluginDrivenExternalTable) getTargetTable(); + Optional snapshot = MvccUtil.getSnapshotFromContext(table, + Optional.ofNullable(getQueryTableSnapshot()), Optional.ofNullable(getScanParams())); + Map partitions = table.getNameToPartitionItems(snapshot); + selectedPartitions = materializeDeferredSelectedPartitions(selectedPartitions, partitions); + } + + static SelectedPartitions materializeDeferredSelectedPartitions(SelectedPartitions selectedPartitions, + Map partitions) { + if (!selectedPartitions.isDeferredPartitionPruning()) { + return selectedPartitions; + } + return new SelectedPartitions(partitions.size(), partitions, false); + } + @Override protected void convertPredicate() { // Attempt filter pushdown via the connector SPI @@ -1945,7 +1970,7 @@ private boolean computeBatchMode() { */ static boolean shouldUseBatchMode(SelectedPartitions selectedPartitions, boolean hasSlots, boolean supportsBatchScan, int numPartitionsInBatchMode) { - if (selectedPartitions == null || selectedPartitions == SelectedPartitions.NOT_PRUNED) { + if (selectedPartitions == null || selectedPartitions.isNotPruned()) { return false; } if (!hasSlots) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java index b3b121053c7913..dbbf1ac864e3b2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java @@ -18,7 +18,10 @@ package org.apache.doris.nereids.rules.rewrite; import org.apache.doris.catalog.PartitionItem; +import org.apache.doris.connector.spi.pushdown.ConnectorExpression; import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.connector.converter.NereidsToConnectorExpressionConverter; +import org.apache.doris.datasource.plugin.PluginDrivenExternalTable; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.rules.Rule; import org.apache.doris.nereids.rules.RuleType; @@ -114,21 +117,49 @@ private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable, } Map nameToPartitionItem = scan.getSelectedPartitions().selectedPartitions; + boolean connectorFilteredPartitions = false; + if (nameToPartitionItem.isEmpty() + && scan.getSelectedPartitions().isDeferredPartitionPruning() + && externalTable instanceof PluginDrivenExternalTable + && ((PluginDrivenExternalTable) externalTable).supportsConnectorPartitionPruning()) { + ConnectorExpression connectorPredicate = + NereidsToConnectorExpressionConverter.convert(filter.getPredicate()); + if (connectorPredicate != null) { + Optional> connectorPartitions = + ((PluginDrivenExternalTable) externalTable).getNameToPartitionItemsByFilter( + ctx.getStatementContext().getSnapshot(externalTable, + scan.getTableSnapshot(), scan.getScanParams()), connectorPredicate); + if (connectorPartitions.isPresent()) { + nameToPartitionItem = connectorPartitions.get(); + connectorFilteredPartitions = true; + } + } + } + if (!connectorFilteredPartitions && nameToPartitionItem.isEmpty() + && (scan.getSelectedPartitions().isNotPruned() + || scan.getSelectedPartitions().isDeferredPartitionPruning())) { + nameToPartitionItem = externalTable.getNameToPartitionItems( + ctx.getStatementContext().getSnapshot(externalTable, + scan.getTableSnapshot(), scan.getScanParams())); + } + final Map partitionItems = nameToPartitionItem; Optional> sortedPartitionRanges = Optional.empty(); boolean enableBinarySearch = ctx.getConnectContext() == null || ctx.getConnectContext().getSessionVariable().enableBinarySearchFilteringPartitions; - if (enableBinarySearch && !nameToPartitionItem.isEmpty()) { - sortedPartitionRanges = scan.getSelectedPartitions().sortedPartitionRanges - .or(() -> (Optional) externalTable.getSortedPartitionRanges(scan)) - .or(() -> Optional.ofNullable(SortedPartitionRanges.build(nameToPartitionItem))); + if (enableBinarySearch && !partitionItems.isEmpty()) { + sortedPartitionRanges = connectorFilteredPartitions + ? Optional.ofNullable(SortedPartitionRanges.build(partitionItems)) + : scan.getSelectedPartitions().sortedPartitionRanges + .or(() -> (Optional) externalTable.getSortedPartitionRanges(scan)) + .or(() -> Optional.ofNullable(SortedPartitionRanges.build(partitionItems))); } PartitionPruneResult result = PartitionPruner.pruneWithResult( - partitionSlots, filter.getPredicate(), nameToPartitionItem, ctx, + partitionSlots, filter.getPredicate(), partitionItems, ctx, PartitionTableType.EXTERNAL, sortedPartitionRanges); List prunedPartitions = new ArrayList<>(result.partitions); for (String name : prunedPartitions) { - PartitionItem item = nameToPartitionItem.get(name); + PartitionItem item = partitionItems.get(name); // Within THIS query, nameToPartitionItem and sortedPartitionRanges are built from the same // frozen map. On a cross-query cache HIT, sortedPartitionRanges instead reuses ranges built by // an earlier query keyed by the identical (snapshotId, schemaId) version token -- content is @@ -139,7 +170,9 @@ private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable, "pruned partition %s is missing in the selected partitions snapshot", name); selectedPartitionItems.put(name, item); } - return new SelectedPartitions(nameToPartitionItem.size(), selectedPartitionItems, true, - result.hasPartitionPredicate); + long totalPartitionNum = connectorFilteredPartitions + ? SelectedPartitions.UNKNOWN_TOTAL_PARTITION_NUM : partitionItems.size(); + return new SelectedPartitions(totalPartitionNum, selectedPartitionItems, true, + connectorFilteredPartitions || result.hasPartitionPredicate); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/QueryPartitionCollector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/QueryPartitionCollector.java index e0c833f95b14b1..bd1d24fa9da755 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/QueryPartitionCollector.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/QueryPartitionCollector.java @@ -75,9 +75,13 @@ public Void visitLogicalCatalogRelation(LogicalCatalogRelation catalogRelation, && ((ExternalTable) catalogRelation.getTable()).supportInternalPartitionPruned()) { LogicalFileScan logicalFileScan = (LogicalFileScan) catalogRelation; SelectedPartitions selectedPartitions = logicalFileScan.getSelectedPartitions(); - tablePartitions.addAll(selectedPartitions.selectedPartitions.keySet()); - tableUsedPartitionNameMap.put(table.getFullQualifiers(), - Pair.of(catalogRelation.getRelationId(), tablePartitions)); + if (selectedPartitions.isDeferredPartitionPruning()) { + tableUsedPartitionNameMap.put(table.getFullQualifiers(), PartitionCompensator.ALL_PARTITIONS); + } else { + tablePartitions.addAll(selectedPartitions.selectedPartitions.keySet()); + tableUsedPartitionNameMap.put(table.getFullQualifiers(), + Pair.of(catalogRelation.getRelationId(), tablePartitions)); + } } else { // not support get partition scene, we consider query all partitions from table tableUsedPartitionNameMap.put(table.getFullQualifiers(), PartitionCompensator.ALL_PARTITIONS); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java index d981074a2fadfd..01df1bed960fa1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java @@ -299,11 +299,25 @@ private boolean hasSameScanParams(Optional left, Optional> sortedPartitionRanges; + public final State state; + /** * Constructor for SelectedPartitions. */ @@ -352,12 +368,28 @@ public SelectedPartitions(long totalPartitionNum, Map sel public SelectedPartitions(long totalPartitionNum, Map selectedPartitions, boolean isPruned, boolean hasPartitionPredicate, Optional> sortedPartitionRanges) { + this(totalPartitionNum, selectedPartitions, isPruned, hasPartitionPredicate, sortedPartitionRanges, + State.MATERIALIZED); + } + + private SelectedPartitions(long totalPartitionNum, Map selectedPartitions, + boolean isPruned, boolean hasPartitionPredicate, + Optional> sortedPartitionRanges, State state) { this.totalPartitionNum = totalPartitionNum; this.selectedPartitions = ImmutableMap.copyOf(Objects.requireNonNull(selectedPartitions, "selectedPartitions is null")); this.isPruned = isPruned; this.hasPartitionPredicate = hasPartitionPredicate; this.sortedPartitionRanges = sortedPartitionRanges; + this.state = state; + } + + public boolean isNotPruned() { + return state == State.NOT_PRUNED; + } + + public boolean isDeferredPartitionPruning() { + return state == State.DEFERRED; } @Override @@ -371,6 +403,7 @@ public boolean equals(Object o) { SelectedPartitions that = (SelectedPartitions) o; return isPruned == that.isPruned && hasPartitionPredicate == that.hasPartitionPredicate + && state == that.state && Objects.equals( selectedPartitions.keySet(), that.selectedPartitions.keySet()) && Objects.equals( @@ -379,7 +412,8 @@ public boolean equals(Object o) { @Override public int hashCode() { - return Objects.hash(selectedPartitions, isPruned, hasPartitionPredicate, sortedPartitionRanges.isPresent()); + return Objects.hash(selectedPartitions, isPruned, hasPartitionPredicate, + sortedPartitionRanges.isPresent(), state); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java index ba82c074d541a6..a628546949dc28 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java @@ -567,6 +567,47 @@ public void testLoadSnapshotEmptyMaterializes() { "the latest pin must carry the materialized partition view"); } + @Test + public void testConnectorPartitionPruningLatestSnapshotDefersPartitionEnumeration() { + Fixture f = Fixture.connectorPartitionPruning(); + + Assertions.assertTrue(f.table.supportsLatestSnapshotPreload(), + "the lightweight connector snapshot must be preloaded before internal table locks are taken"); + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) f.table.loadSnapshot( + Optional.empty(), Optional.empty()); + + Assertions.assertEquals(PINNED_SNAPSHOT_ID, pin.getConnectorSnapshot().getSnapshotId()); + Assertions.assertTrue(pin.getNameToPartitionItem().isEmpty(), + "a connector-filtered table must not enumerate every partition while binding the latest snapshot"); + Mockito.verify(f.metadata, Mockito.never()).listPartitions( + Mockito.any(), Mockito.any(), Mockito.any()); + } + + @Test + public void testConnectorPartitionPruningMaterializesFullViewForExplicitConsumer() { + Fixture f = Fixture.connectorPartitionPruning(); + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + + Map partitions = f.table.getNameToPartitionItems(Optional.empty()); + + Assertions.assertEquals(2, partitions.size()); + Mockito.verify(f.metadata).listPartitions( + Mockito.eq(f.session), Mockito.eq(f.handle), Mockito.eq(Optional.empty())); + } + + @Test + public void testConnectorPartitionPruningUsesOnDemandMtmvFreshness() throws AnalysisException { + Fixture f = Fixture.connectorPartitionPruning(); + flagPinLastModified(f); + Mockito.when(f.metadata.getPartitionFreshnessMillis(Mockito.any(), Mockito.any(), + Mockito.eq("dt=2024-01-01"))).thenReturn(OptionalLong.of(TS_2024_01_01)); + + MTMVTimestampSnapshot snapshot = (MTMVTimestampSnapshot) f.table.getPartitionSnapshot( + "dt=2024-01-01", null, Optional.empty()); + + Assertions.assertEquals(TS_2024_01_01, snapshot.getSnapshotVersion()); + } + @Test public void testInitialLatestPartitionAccountingUsesPinnedHandle() { Fixture f = Fixture.partitioned(); @@ -1334,6 +1375,13 @@ private static ConnectorPartitionInfo cpi(String name, long lastModifiedMillis) ConnectorPartitionInfo.UNKNOWN, orderedValuesOf(name), Collections.emptyList()); } + private static ConnectorPartitionInfo cpiWithPartitionValueMap(String name, long lastModifiedMillis) { + List values = orderedValuesOf(name); + return new ConnectorPartitionInfo(name, Collections.singletonMap("dt", values.get(0)), Collections.emptyMap(), + ConnectorPartitionInfo.UNKNOWN, ConnectorPartitionInfo.UNKNOWN, lastModifiedMillis, + ConnectorPartitionInfo.UNKNOWN, values, Collections.emptyList()); + } + /** * Like {@link #cpi} but with the ordered values supplied EXPLICITLY rather than derived from the * rendered name — needed for the spec-evolution shapes, where a row's value count legitimately @@ -1433,6 +1481,12 @@ static Fixture partitioned() { cpi("dt=2024-02-02", TS_2024_02_02))); } + static Fixture connectorPartitionPruning() { + return build(Arrays.asList( + cpiWithPartitionValueMap("dt=2024-01-01", TS_2024_01_01), + cpiWithPartitionValueMap("dt=2024-02-02", TS_2024_02_02)), false, Type.DATEV2, true); + } + static Fixture with(List partitions) { return build(partitions, false); } @@ -1476,11 +1530,16 @@ static Fixture rangeView(ConnectorMvccPartitionView view) { } private static Fixture build(List partitions, boolean timeTravel) { - return build(partitions, timeTravel, Type.DATEV2); + return build(partitions, timeTravel, Type.DATEV2, false); } private static Fixture build(List partitions, boolean timeTravel, Type partitionColType) { + return build(partitions, timeTravel, partitionColType, false); + } + + private static Fixture build(List partitions, boolean timeTravel, + Type partitionColType, boolean connectorPartitionPruning) { ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class); ConnectorSession session = Mockito.mock(ConnectorSession.class); Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE); @@ -1551,6 +1610,11 @@ protected Optional getLatestSchemaCacheValue() { // canned value so the real getSchemaCacheValue() override is exercised. return Optional.of(latestCacheValue); } + + @Override + public boolean supportsConnectorPartitionPruning() { + return connectorPartitionPruning; + } }; return new Fixture(table, metadata, handle, pinnedHandle, session, latestCacheValue, resolvedSnapshot); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTablePartitionTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTablePartitionTest.java index 34f0c0033cf869..eac0bc569c48f0 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTablePartitionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTablePartitionTest.java @@ -34,6 +34,7 @@ import org.apache.doris.datasource.ExternalDatabase; import org.apache.doris.datasource.SchemaCacheValue; import org.apache.doris.datasource.SessionContext; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -111,6 +112,35 @@ public void testNonPartitionedTableReportsNoPartitionsButStillOptsIntoPruning() + "queries silently return zero rows (FIX-NONPART-PRUNE-DATALOSS)"); } + @Test + public void testConnectorPartitionPruningDefersInitialPartitionMaterialization() { + List schema = Arrays.asList( + new Column("year", PrimitiveType.INT), new Column("val", PrimitiveType.INT)); + PluginDrivenSchemaCacheValue cacheValue = new PluginDrivenSchemaCacheValue( + schema, Collections.singletonList(schema.get(0)), Collections.singletonList("year")); + PluginDrivenExternalTable table = new PluginDrivenExternalTable(1L, "tbl", "REMOTE_TBL", + new TestablePluginCatalog("max_compute", Mockito.mock(ConnectorMetadata.class), noneScopedSession()), + mockDb("REMOTE_DB")) { + @Override + protected synchronized void makeSureInitialized() { + // no-op: skip Env-backed catalog/db init + } + + @Override + public Optional getSchemaCacheValue() { + return Optional.of(cacheValue); + } + + @Override + public boolean supportsConnectorPartitionPruning() { + return true; + } + }; + + Assertions.assertSame(SelectedPartitions.DEFERRED_PARTITION_PRUNING, + table.initSelectedPartitions(Optional.empty())); + } + // ==================== getNameToPartitionItems (raw remote-name addressing) ==================== @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java index ba546ecda19c5d..95414827d4a230 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java @@ -104,6 +104,21 @@ public void testNoPredicatePartitionedTableBatches() { PluginDrivenScanNode.shouldUseBatchMode(noPredicateFullScan, true, true, THRESHOLD)); } + @Test + public void testDeferredPartitionViewMaterializationRestoresFullScanBatchEligibility() { + Map items = new LinkedHashMap<>(); + for (int index = 0; index < THRESHOLD; index++) { + items.put("pt=" + index, Mockito.mock(PartitionItem.class)); + } + + SelectedPartitions materialized = PluginDrivenScanNode.materializeDeferredSelectedPartitions( + SelectedPartitions.DEFERRED_PARTITION_PRUNING, items); + + Assertions.assertNotSame(SelectedPartitions.DEFERRED_PARTITION_PRUNING, materialized); + Assertions.assertTrue( + PluginDrivenScanNode.shouldUseBatchMode(materialized, true, true, THRESHOLD)); + } + @Test public void testNoSlotsNeverBatches() { // No required slots (e.g. count-only) -> not batch. Pins the hasSlots guard. diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionCountTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionCountTest.java index 8dea308d07bc47..eb3f0da48fa046 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionCountTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionCountTest.java @@ -99,6 +99,14 @@ public void testPrunedToZeroReportsZeroOverTotal() { PluginDrivenScanNode.displayPartitionCounts(prunedToZero)); } + @Test + public void testUnknownTotalIsPreservedForExplain() { + SelectedPartitions pruned = new SelectedPartitions( + SelectedPartitions.UNKNOWN_TOTAL_PARTITION_NUM, items(2), true, true); + Assertions.assertArrayEquals(new long[] {2, SelectedPartitions.UNKNOWN_TOTAL_PARTITION_NUM}, + PluginDrivenScanNode.displayPartitionCounts(pruned)); + } + // FIX-L12 — guards resolveSelectedPartitionNum, which prefers the connector's real scanned-partition // count (distinct native partitions after the connector's SDK manifest/residual/transform pruning) over // the engine's Nereids declared-column prune count, so partition=N/M and sql_block_rule reflect what is diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionPruningTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionPruningTest.java index e7ebc4eef1a23a..a6628d9ec568da 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionPruningTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionPruningTest.java @@ -118,6 +118,17 @@ public void testPrunedToZeroWithoutIgnoreStillShortCircuits() { Assertions.assertTrue(result.isEmpty()); } + @Test + public void testPrunedToZeroWithUnknownTotalStillShortCircuits() { + SelectedPartitions emptyPruned = new SelectedPartitions( + SelectedPartitions.UNKNOWN_TOTAL_PARTITION_NUM, Collections.emptyMap(), true, true); + + List result = PluginDrivenScanNode.resolveRequiredPartitions(emptyPruned); + + Assertions.assertNotNull(result); + Assertions.assertTrue(result.isEmpty()); + } + @Test public void testPrunedToZeroWithIgnoreShortCircuitScansAll() { // A predicate-driven connector (paimon: ConnectorScanPlanProvider.ignorePartitionPruneShortCircuit()