This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f6ec58f4013 branch-4.1: [improvement](hive) Batch Hive metastore
partition access (#68197)
f6ec58f4013 is described below
commit f6ec58f4013b4e37e5292f669500164a4a43b490
Author: Calvin Kirs <[email protected]>
AuthorDate: Mon Sep 21 11:18:45 2026 +0800
branch-4.1: [improvement](hive) Batch Hive metastore partition access
(#68197)
https://github.com/apache/doris/pull/67186
---
.../doris/datasource/hive/HMSCachedClient.java | 9 +
.../doris/datasource/hive/HMSClientException.java | 11 +
.../doris/datasource/hive/HMSExternalTable.java | 10 +
.../apache/doris/datasource/hive/HiveDlaTable.java | 53 ++++
.../datasource/hive/HiveExternalMetaCache.java | 30 ++-
.../datasource/hive/HmsPartitionBatchExecutor.java | 252 ++++++++++++++++++
...Exception.java => HmsPartitionBatchResult.java} | 25 +-
.../datasource/hive/HmsPartitionBatchStats.java | 151 +++++++++++
.../datasource/hive/HmsPartitionIdentity.java | 98 +++++++
.../doris/datasource/hive/HmsPartitionRequest.java | 74 ++++++
.../hive/HmsPartitionResultException.java | 98 +++++++
...ntException.java => HmsPartitionTransport.java} | 16 +-
.../datasource/hive/ThriftHMSCachedClient.java | 74 +++++-
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 11 +
.../org/apache/doris/mtmv/MTMVPartitionUtil.java | 53 ++--
.../org/apache/doris/mtmv/MTMVRefreshContext.java | 113 +++++++++
.../org/apache/doris/mtmv/MTMVRelatedTableIf.java | 19 ++
.../org/apache/doris/mtmv/MTMVRewriteUtil.java | 21 +-
.../hive/HmsPartitionBatchExecutorTest.java | 282 +++++++++++++++++++++
.../apache/doris/mtmv/MTMVPartitionUtilTest.java | 27 +-
.../doris/mtmv/MTMVRefreshContextBatchTest.java | 239 +++++++++++++++++
.../org/apache/doris/mtmv/MTMVRewriteUtilTest.java | 24 +-
22 files changed, 1631 insertions(+), 59 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSCachedClient.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSCachedClient.java
index fcaefecf41b..a3a04600c9f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSCachedClient.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSCachedClient.java
@@ -59,6 +59,15 @@ public interface HMSCachedClient {
List<Partition> getPartitions(String dbName, String tblName, List<String>
partitionNames);
+ /**
+ * Gets the subset of the requested partitions that exist. Unlike {@link
#getPartitions}, a name the
+ * metastore no longer has is a normal answer and is simply absent from
the result. The compatibility
+ * default preserves implementations whose {@code getPartitions} already
returns only what exists.
+ */
+ default List<Partition> getExistingPartitions(String dbName, String
tblName, List<String> partitionNames) {
+ return getPartitions(dbName, tblName, partitionNames);
+ }
+
Table getTable(String dbName, String tblName);
List<FieldSchema> getSchema(String dbName, String tblName);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
index 6e714e20a9b..242dfc12a8c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
@@ -20,6 +20,8 @@ package org.apache.doris.datasource.hive;
import org.apache.doris.common.util.Util;
public class HMSClientException extends RuntimeException {
+ private HmsPartitionBatchStats partitionBatchStats;
+
public HMSClientException(String format, Throwable cause, Object... msg) {
super(String.format(format, msg) + (cause == null ? "" : ". reason: "
+ Util.getRootCauseMessage(cause)),
cause);
@@ -28,4 +30,13 @@ public class HMSClientException extends RuntimeException {
public HMSClientException(String format, Object... msg) {
super(String.format(format, msg));
}
+
+ public HmsPartitionBatchStats getPartitionBatchStats() {
+ return partitionBatchStats;
+ }
+
+ HMSClientException withPartitionBatchStats(HmsPartitionBatchStats stats) {
+ this.partitionBatchStats = stats;
+ return this;
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
index a727d4386f7..2a7759faecb 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java
@@ -1058,6 +1058,16 @@ public class HMSExternalTable extends ExternalTable
implements MTMVRelatedTableI
return dlaTable.getPartitionSnapshot(partitionName, context, snapshot);
}
+ @Override
+ public Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String>
partitionNames,
+ MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
throws AnalysisException {
+ makeSureInitialized();
+ if (dlaTable instanceof HiveDlaTable) {
+ return ((HiveDlaTable)
dlaTable).getPartitionSnapshots(partitionNames, snapshot);
+ }
+ return MTMVRelatedTableIf.super.getPartitionSnapshots(partitionNames,
context, snapshot);
+ }
+
@Override
public MTMVSnapshotIf getTableSnapshot(MTMVRefreshContext context,
Optional<MvccSnapshot> snapshot)
throws AnalysisException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveDlaTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveDlaTable.java
index bfd39c59684..e8c12dd5623 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveDlaTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveDlaTable.java
@@ -32,7 +32,9 @@ import org.apache.doris.mtmv.MTMVTimestampSnapshot;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -80,6 +82,57 @@ public class HiveDlaTable extends HMSDlaTable {
return new MTMVTimestampSnapshot(hivePartition.getLastModifiedTime());
}
+ /**
+ * Bulk form of {@link #getPartitionSnapshot}: resolves every requested
name against one partition-value
+ * listing and loads all cache misses through one batched HMS request
instead of one RPC per partition.
+ * A name absent from the listing is omitted (the refresh context reports
it per partition); any other
+ * failure is normalized to the checked AnalysisException this MTMV
boundary declares, so the
+ * transparent-rewrite path degrades per-MV instead of a raw runtime
exception disabling every MV
+ * candidate at the planner hook.
+ */
+ Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String>
partitionNames,
+ Optional<MvccSnapshot> snapshot) throws AnalysisException {
+ try {
+ HiveExternalMetaCache.HivePartitionValues hivePartitionValues =
+ hmsTable.getHivePartitionValues(snapshot);
+ HiveExternalMetaCache cache =
Env.getCurrentEnv().getExtMetaCacheMgr()
+ .hive(hmsTable.getCatalog().getId());
+ List<String> resolvedNames = new
ArrayList<>(partitionNames.size());
+ List<List<String>> resolvedValues = new
ArrayList<>(partitionNames.size());
+ for (String partitionName : partitionNames) {
+ Long partitionId =
hivePartitionValues.getPartitionNameToIdMap().get(partitionName);
+ if (partitionId == null) {
+ continue;
+ }
+ List<String> partitionValues =
hivePartitionValues.getPartitionValuesMap().get(partitionId);
+ if (CollectionUtils.isEmpty(partitionValues)) {
+ continue;
+ }
+ resolvedNames.add(partitionName);
+ resolvedValues.add(partitionValues);
+ }
+ Map<String, MTMVSnapshotIf> result = new LinkedHashMap<>();
+ if (resolvedNames.isEmpty()) {
+ return result;
+ }
+ List<HivePartition> partitions =
cache.getAllPartitionsWithCache(hmsTable, resolvedValues);
+ if (partitions.size() != resolvedNames.size()) {
+ throw new AnalysisException("Invalid HMS partition result:
requested="
+ + resolvedNames.size() + ", returned=" +
partitions.size());
+ }
+ for (int i = 0; i < resolvedNames.size(); i++) {
+ result.put(resolvedNames.get(i),
+ new
MTMVTimestampSnapshot(partitions.get(i).getLastModifiedTime()));
+ }
+ return result;
+ } catch (AnalysisException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new AnalysisException("failed to load partition snapshots
for "
+ + hmsTable.getName() + ": " + e.getMessage(), e);
+ }
+ }
+
@Override
public MTMVSnapshotIf getTableSnapshot(MTMVRefreshContext context,
Optional<MvccSnapshot> snapshot)
throws AnalysisException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
index 981b1f64013..cdf58a9adc6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java
@@ -482,7 +482,10 @@ public class HiveExternalMetaCache extends
AbstractExternalMetaCache {
return sb.toString();
}).collect(Collectors.toList());
- List<Partition> partitions = catalog.getClient().getPartitions(
+ // Lenient existence semantics: a partition dropped remotely since the
name list was captured is
+ // simply absent from the result (the historical getPartitionsByNames
behavior); the client batches
+ // and validates the physical RPCs internally.
+ List<Partition> partitions = catalog.getClient().getExistingPartitions(
nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName(),
partitionNames);
for (Partition partition : partitions) {
StorageDescriptor sd = partition.getSd();
@@ -657,7 +660,30 @@ public class HiveExternalMetaCache extends
AbstractExternalMetaCache {
List<HivePartition> partitions;
if (withCache) {
MetaCacheEntry<PartitionCacheKey, HivePartition> partitionEntry =
this.partitionEntry.get(catalogId);
- partitions =
keys.stream().map(partitionEntry::get).collect(Collectors.toList());
+ // Serve hits from the cache and aggregate every miss into ONE
bulk load (batched inside the
+ // client) instead of one single-partition RPC per missed key.
+ Map<PartitionCacheKey, HivePartition> resolved = new HashMap<>();
+ List<PartitionCacheKey> misses = new ArrayList<>();
+ for (PartitionCacheKey key : keys) {
+ HivePartition hit = partitionEntry.getIfPresent(key);
+ if (hit != null) {
+ resolved.put(key, hit);
+ } else {
+ misses.add(key);
+ }
+ }
+ if (!misses.isEmpty()) {
+ Map<PartitionCacheKey, HivePartition> loaded =
loadPartitions(misses);
+ loaded.forEach(partitionEntry::put);
+ resolved.putAll(loaded);
+ }
+ partitions = new ArrayList<>(keys.size());
+ for (PartitionCacheKey key : keys) {
+ HivePartition partition = resolved.get(key);
+ // A key the bulk load did not return (partition dropped
remotely) falls back to the
+ // per-key loader, preserving the original single-load
not-found error for that partition.
+ partitions.add(partition != null ? partition :
partitionEntry.get(key));
+ }
} else {
partitions = new ArrayList<>(loadPartitions(keys).values());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutor.java
new file mode 100644
index 00000000000..245f52b8bd6
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutor.java
@@ -0,0 +1,252 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import org.apache.hadoop.hive.metastore.api.Partition;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/** Splits one logical partition request into bounded, validated HMS RPCs. */
+final class HmsPartitionBatchExecutor {
+
+ static final class RemoteCallException extends HMSClientException {
+ RemoteCallException(String messageDetail, Throwable cause) {
+ // The two-argument parent constructor treats the message as a
format string; route the detail
+ // through an explicit %s argument so partition names containing
'%' cannot break formatting.
+ super("Remote HMS partition operation failed: %s", cause,
messageDetail);
+ }
+ }
+
+ private final int maxBatchSize;
+ private final HmsPartitionTransport transport;
+
+ HmsPartitionBatchExecutor(int maxBatchSize, HmsPartitionTransport
transport) {
+ if (maxBatchSize <= 0) {
+ throw new IllegalArgumentException("invalid HMS partition batch
size");
+ }
+ this.maxBatchSize = maxBatchSize;
+ this.transport = java.util.Objects.requireNonNull(transport,
"transport");
+ }
+
+ HmsPartitionBatchResult executeExistingWithStats(HmsPartitionRequest
request) {
+ return executeWithStats(request, true);
+ }
+
+ HmsPartitionBatchResult executeWithStats(HmsPartitionRequest request) {
+ return executeWithStats(request, false);
+ }
+
+ private HmsPartitionBatchResult executeWithStats(HmsPartitionRequest
request, boolean allowMissing) {
+ long logicalStartNanos = System.nanoTime();
+ List<HmsPartitionIdentity.ParsedPartitionName> partitions =
request.getPartitions();
+ if (partitions.isEmpty()) {
+ HmsPartitionBatchStats stats = HmsPartitionBatchStats.builder()
+ .logicalElapsedNanos(System.nanoTime() - logicalStartNanos)
+ .build();
+ return new HmsPartitionBatchResult(new ArrayList<>(), stats);
+ }
+
+ List<Partition> result = new ArrayList<>(partitions.size());
+ int offset = 0;
+ int effectiveBatchSize = maxBatchSize;
+ int transportInvocations = 0;
+ int fallbackCount = 0;
+ long transportItems = 0;
+ long transportElapsedNanos = 0;
+ long maxTransportElapsedNanos = 0;
+ int largestBatchSize = 0;
+ int smallestBatchSize = Integer.MAX_VALUE;
+ while (offset < partitions.size()) {
+ int batchSize = Math.min(effectiveBatchSize, partitions.size() -
offset);
+ List<HmsPartitionIdentity.ParsedPartitionName> batch =
+ partitions.subList(offset, offset + batchSize);
+ List<String> batchNames = new ArrayList<>(batch.size());
+ for (HmsPartitionIdentity.ParsedPartitionName partition : batch) {
+ batchNames.add(partition.getName());
+ }
+ transportInvocations++;
+ transportItems += batchSize;
+ largestBatchSize = Math.max(largestBatchSize, batchSize);
+ smallestBatchSize = Math.min(smallestBatchSize, batchSize);
+ long transportStartNanos = System.nanoTime();
+ HMSClientException terminalFailure = null;
+ try {
+ List<Partition> returned = transport.getPartitionsByNames(
+ request.getDbName(), request.getTableName(),
batchNames);
+ result.addAll(validateAndOrder(batch, returned, allowMissing));
+ offset += batchSize;
+ } catch (RemoteCallException e) {
+ if (batchSize == 1 || !isDegradableRemoteFailure(e)) {
+ terminalFailure = finalBatchFailure(request, offset,
batchSize, effectiveBatchSize,
+ transportInvocations, fallbackCount, e);
+ } else {
+ effectiveBatchSize = Math.max(1, batchSize / 2);
+ fallbackCount++;
+ }
+ } catch (HMSClientException e) {
+ terminalFailure = e;
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ terminalFailure = new HMSClientException(
+ "Unexpected checked failure fetching HMS partitions",
e);
+ } finally {
+ long elapsedNanos = System.nanoTime() - transportStartNanos;
+ transportElapsedNanos += elapsedNanos;
+ maxTransportElapsedNanos = Math.max(maxTransportElapsedNanos,
elapsedNanos);
+ }
+ if (terminalFailure != null) {
+ throw terminalFailure.withPartitionBatchStats(buildStats(
+ partitions.size(), transportInvocations,
transportItems,
+ largestBatchSize, smallestBatchSize,
+ fallbackCount, System.nanoTime() - logicalStartNanos,
+ transportElapsedNanos, maxTransportElapsedNanos));
+ }
+ }
+ HmsPartitionBatchStats stats = buildStats(
+ partitions.size(), transportInvocations, transportItems,
largestBatchSize, smallestBatchSize,
+ fallbackCount, System.nanoTime() - logicalStartNanos,
+ transportElapsedNanos, maxTransportElapsedNanos);
+ return new HmsPartitionBatchResult(result, stats);
+ }
+
+ private static HmsPartitionBatchStats buildStats(
+ int requestedItems, int invocations, long transportItems, int
largestBatchSize,
+ int smallestBatchSize, int fallbackCount, long logicalElapsedNanos,
+ long transportElapsedNanos, long maxTransportElapsedNanos) {
+ return HmsPartitionBatchStats.builder()
+ .requestedItems(requestedItems)
+ .transportInvocations(invocations)
+ .transportItems(transportItems)
+ .largestBatchSize(largestBatchSize)
+ .smallestBatchSize(smallestBatchSize)
+ .fallbackCount(fallbackCount)
+ .logicalElapsedNanos(logicalElapsedNanos)
+ .transportElapsedNanos(transportElapsedNanos)
+ .maxTransportElapsedNanos(maxTransportElapsedNanos)
+ .build();
+ }
+
+ private static List<Partition> validateAndOrder(
+ List<HmsPartitionIdentity.ParsedPartitionName> requested,
+ List<Partition> returned, boolean allowMissing) {
+ int expectedValueCount = requested.get(0).getValues().size();
+ Map<List<String>, Integer> expected = new HashMap<>();
+ for (int i = 0; i < requested.size(); i++) {
+ expected.put(requested.get(i).getValues(), i);
+ }
+
+ HmsPartitionResultException.Builder failure =
HmsPartitionResultException.builder(
+ requested.size(), returned == null ? 0 : returned.size());
+ List<Partition> ordered = new
ArrayList<>(java.util.Collections.nCopies(requested.size(), null));
+ Map<List<String>, Integer> returnedCounts = new LinkedHashMap<>();
+ if (returned == null) {
+ failure.invalid("<null response>");
+ } else {
+ for (Partition partition : returned) {
+ if (partition == null) {
+ failure.invalid("<null partition>");
+ continue;
+ }
+ List<String> identity = partition.getValues();
+ if (identity == null || identity.size() != expectedValueCount)
{
+ failure.invalid(String.valueOf(identity));
+ continue;
+ }
+ returnedCounts.merge(identity, 1, Integer::sum);
+ Integer index = expected.get(identity);
+ if (index != null && ordered.get(index) == null) {
+ ordered.set(index, partition);
+ }
+ }
+ }
+ for (HmsPartitionIdentity.ParsedPartitionName partition : requested) {
+ if (!allowMissing &&
!returnedCounts.containsKey(partition.getValues())) {
+ failure.missing(partition.getName());
+ }
+ }
+ for (Map.Entry<List<String>, Integer> entry :
returnedCounts.entrySet()) {
+ if (!expected.containsKey(entry.getKey())) {
+ failure.unexpected(entry.getKey().toString());
+ }
+ if (entry.getValue() > 1) {
+ failure.duplicate(entry.getKey().toString());
+ }
+ }
+ if (failure.hasMismatches()) {
+ throw failure.build();
+ }
+ if (!allowMissing) {
+ return ordered;
+ }
+ List<Partition> existing = new ArrayList<>(returnedCounts.size());
+ for (Partition partition : ordered) {
+ if (partition != null) {
+ existing.add(partition);
+ }
+ }
+ return existing;
+ }
+
+ private HMSClientException finalBatchFailure(HmsPartitionRequest request,
int offset,
+ int failedBatchSize, int effectiveBatchSize, int
transportInvocations, int fallbackCount,
+ RemoteCallException failure) {
+ return new HMSClientException(
+ "HMS partition batch request failed: db=%s, table=%s,
requested=%d, offset=%d, "
+ + "failedBatchSize=%d, effectiveBatchSize=%d, "
+ + "transportInvocations=%d, "
+ + "fallbacks=%d: %s",
+ failure,
+ request.getDbName(), request.getTableName(),
request.getPartitions().size(), offset,
+ failedBatchSize, effectiveBatchSize, transportInvocations,
fallbackCount,
+ failure.getMessage());
+ }
+
+ private static boolean isDegradableRemoteFailure(RemoteCallException
failure) {
+ boolean thriftFailure = false;
+ boolean sizeFailure = false;
+ for (Throwable current = failure.getCause(); current != null; current
= current.getCause()) {
+ thriftFailure |= current instanceof TException;
+ String message = current.getMessage();
+ if (message != null) {
+ String normalized = message.toLowerCase(Locale.ROOT);
+ sizeFailure |= normalized.contains("message size")
+ || normalized.contains("max message")
+ || normalized.contains("maxmessagesize")
+ || normalized.contains("frame too large")
+ || normalized.contains("request too large")
+ || normalized.contains("payload too large")
+ || normalized.contains("too many partitions")
+ || normalized.contains("partition limit")
+ ||
normalized.contains("hive.metastore.limit.partition.request")
+ || (normalized.contains("partitions scanned")
+ && normalized.contains("exceeds limit"))
+ || (normalized.contains("frame size")
+ && normalized.contains("larger than max
length"));
+ }
+ }
+ return thriftFailure && sizeFailure;
+ }
+
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchResult.java
similarity index 54%
copy from
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
copy to
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchResult.java
index 6e714e20a9b..70c68933577 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchResult.java
@@ -17,15 +17,26 @@
package org.apache.doris.datasource.hive;
-import org.apache.doris.common.util.Util;
+import org.apache.hadoop.hive.metastore.api.Partition;
-public class HMSClientException extends RuntimeException {
- public HMSClientException(String format, Throwable cause, Object... msg) {
- super(String.format(format, msg) + (cause == null ? "" : ". reason: "
+ Util.getRootCauseMessage(cause)),
- cause);
+import java.util.List;
+import java.util.Objects;
+
+/** Partition objects and the physical HMS batching statistics that produced
them. */
+public final class HmsPartitionBatchResult {
+ private final List<Partition> partitions;
+ private final HmsPartitionBatchStats stats;
+
+ public HmsPartitionBatchResult(List<Partition> partitions,
HmsPartitionBatchStats stats) {
+ this.partitions = Objects.requireNonNull(partitions, "partitions");
+ this.stats = Objects.requireNonNull(stats, "stats");
+ }
+
+ public List<Partition> getPartitions() {
+ return partitions;
}
- public HMSClientException(String format, Object... msg) {
- super(String.format(format, msg));
+ public HmsPartitionBatchStats getStats() {
+ return stats;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchStats.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchStats.java
new file mode 100644
index 00000000000..1e2c476875a
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchStats.java
@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import java.io.Serializable;
+
+/** Immutable execution statistics for one logical HMS partition-object
request. */
+public final class HmsPartitionBatchStats implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final int requestedItems;
+ private final int transportInvocations;
+ private final long transportItems;
+ private final int largestBatchSize;
+ private final int smallestBatchSize;
+ private final int fallbackCount;
+ private final long logicalElapsedNanos;
+ private final long transportElapsedNanos;
+ private final long maxTransportElapsedNanos;
+
+ private HmsPartitionBatchStats(Builder builder) {
+ this.requestedItems = builder.requestedItems;
+ this.transportInvocations = builder.transportInvocations;
+ this.transportItems = builder.transportItems;
+ this.largestBatchSize = builder.largestBatchSize;
+ this.smallestBatchSize = builder.smallestBatchSize;
+ this.fallbackCount = builder.fallbackCount;
+ this.logicalElapsedNanos = builder.logicalElapsedNanos;
+ this.transportElapsedNanos = builder.transportElapsedNanos;
+ this.maxTransportElapsedNanos = builder.maxTransportElapsedNanos;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public int getRequestedItems() {
+ return requestedItems;
+ }
+
+ public int getTransportInvocations() {
+ return transportInvocations;
+ }
+
+ public long getTransportItems() {
+ return transportItems;
+ }
+
+ public int getLargestBatchSize() {
+ return largestBatchSize;
+ }
+
+ public int getSmallestBatchSize() {
+ return smallestBatchSize;
+ }
+
+ public int getFallbackCount() {
+ return fallbackCount;
+ }
+
+ public long getLogicalElapsedNanos() {
+ return logicalElapsedNanos;
+ }
+
+ public long getTransportElapsedNanos() {
+ return transportElapsedNanos;
+ }
+
+ public long getMaxTransportElapsedNanos() {
+ return maxTransportElapsedNanos;
+ }
+
+ public static final class Builder {
+ private int requestedItems;
+ private int transportInvocations;
+ private long transportItems;
+ private int largestBatchSize;
+ private int smallestBatchSize;
+ private int fallbackCount;
+ private long logicalElapsedNanos;
+ private long transportElapsedNanos;
+ private long maxTransportElapsedNanos;
+
+ private Builder() {
+ }
+
+ public Builder requestedItems(int requestedItems) {
+ this.requestedItems = requestedItems;
+ return this;
+ }
+
+ public Builder transportInvocations(int transportInvocations) {
+ this.transportInvocations = transportInvocations;
+ return this;
+ }
+
+ public Builder transportItems(long transportItems) {
+ this.transportItems = transportItems;
+ return this;
+ }
+
+ public Builder largestBatchSize(int largestBatchSize) {
+ this.largestBatchSize = largestBatchSize;
+ return this;
+ }
+
+ public Builder smallestBatchSize(int smallestBatchSize) {
+ this.smallestBatchSize = smallestBatchSize;
+ return this;
+ }
+
+ public Builder fallbackCount(int fallbackCount) {
+ this.fallbackCount = fallbackCount;
+ return this;
+ }
+
+ public Builder logicalElapsedNanos(long logicalElapsedNanos) {
+ this.logicalElapsedNanos = logicalElapsedNanos;
+ return this;
+ }
+
+ public Builder transportElapsedNanos(long transportElapsedNanos) {
+ this.transportElapsedNanos = transportElapsedNanos;
+ return this;
+ }
+
+ public Builder maxTransportElapsedNanos(long maxTransportElapsedNanos)
{
+ this.maxTransportElapsedNanos = maxTransportElapsedNanos;
+ return this;
+ }
+
+ public HmsPartitionBatchStats build() {
+ return new HmsPartitionBatchStats(this);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionIdentity.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionIdentity.java
new file mode 100644
index 00000000000..62d292c1cda
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionIdentity.java
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import org.apache.hadoop.hive.common.FileUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+
+final class HmsPartitionIdentity {
+
+ private HmsPartitionIdentity() {
+ }
+
+ static List<String> keysFromName(String partitionName) {
+ return parseParts(partitionName, null, true);
+ }
+
+ static ParsedPartitionName parse(String partitionName, List<String>
expectedKeys) {
+ return new ParsedPartitionName(partitionName,
+ Collections.unmodifiableList(parseParts(partitionName,
expectedKeys, false)));
+ }
+
+ private static List<String> parseParts(
+ String partitionName, List<String> expectedKeys, boolean
returnKeys) {
+ if (partitionName == null || partitionName.isEmpty()) {
+ throw new IllegalArgumentException("partition name must not be
empty");
+ }
+ List<String> parts = new ArrayList<>();
+ int segmentStart = 0;
+ int keyIndex = 0;
+ while (segmentStart < partitionName.length()) {
+ int segmentEnd = partitionName.indexOf('/', segmentStart);
+ if (segmentEnd < 0) {
+ segmentEnd = partitionName.length();
+ }
+ int separator = partitionName.indexOf('=', segmentStart);
+ if (separator <= segmentStart || separator >= segmentEnd) {
+ throw new IllegalArgumentException("invalid partition name: "
+ partitionName);
+ }
+ String key =
FileUtils.unescapePathName(partitionName.substring(segmentStart, separator))
+ .toLowerCase(Locale.ROOT);
+ if (expectedKeys != null
+ && (keyIndex >= expectedKeys.size() ||
!expectedKeys.get(keyIndex).equals(key))) {
+ throw new IllegalArgumentException("inconsistent partition
keys in request: " + partitionName);
+ }
+ parts.add(returnKeys ? key
+ :
FileUtils.unescapePathName(partitionName.substring(separator + 1, segmentEnd)));
+ keyIndex++;
+ if (segmentEnd == partitionName.length()) {
+ break;
+ }
+ segmentStart = segmentEnd + 1;
+ if (segmentStart == partitionName.length()) {
+ throw new IllegalArgumentException("invalid partition name: "
+ partitionName);
+ }
+ }
+ if (expectedKeys != null && keyIndex != expectedKeys.size()) {
+ throw new IllegalArgumentException("inconsistent partition keys in
request: " + partitionName);
+ }
+ return parts;
+ }
+
+ static final class ParsedPartitionName {
+ private final String name;
+ private final List<String> values;
+
+ private ParsedPartitionName(String name, List<String> values) {
+ this.name = name;
+ this.values = values;
+ }
+
+ String getName() {
+ return name;
+ }
+
+ List<String> getValues() {
+ return values;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionRequest.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionRequest.java
new file mode 100644
index 00000000000..13efab2728e
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionRequest.java
@@ -0,0 +1,74 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/** Immutable logical input for one HMS partition-object request. */
+final class HmsPartitionRequest {
+
+ private final String dbName;
+ private final String tableName;
+ private final List<HmsPartitionIdentity.ParsedPartitionName> partitions;
+
+ HmsPartitionRequest(String dbName, String tableName, List<String>
partitionNames) {
+ requireName(dbName, "database");
+ requireName(tableName, "table");
+ this.dbName = dbName;
+ this.tableName = tableName;
+ this.partitions =
parsePartitions(Objects.requireNonNull(partitionNames, "partitionNames"));
+ }
+
+ String getDbName() {
+ return dbName;
+ }
+
+ String getTableName() {
+ return tableName;
+ }
+
+ List<HmsPartitionIdentity.ParsedPartitionName> getPartitions() {
+ return partitions;
+ }
+
+ private static void requireName(String value, String field) {
+ if (value == null || value.isEmpty()) {
+ throw new IllegalArgumentException(field + " must not be empty");
+ }
+ }
+
+ private static List<HmsPartitionIdentity.ParsedPartitionName>
parsePartitions(List<String> names) {
+ List<HmsPartitionIdentity.ParsedPartitionName> parsedPartitions = new
ArrayList<>(names.size());
+ Set<List<String>> identities = new HashSet<>();
+ List<String> partitionKeys = names.isEmpty()
+ ? Collections.emptyList() :
HmsPartitionIdentity.keysFromName(names.get(0));
+ for (String name : names) {
+ HmsPartitionIdentity.ParsedPartitionName parsed =
HmsPartitionIdentity.parse(name, partitionKeys);
+ if (!identities.add(parsed.getValues())) {
+ throw new IllegalArgumentException("duplicate partition
identity in request: " + name);
+ }
+ parsedPartitions.add(parsed);
+ }
+ return Collections.unmodifiableList(parsedPartitions);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionResultException.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionResultException.java
new file mode 100644
index 00000000000..89cabf097ce
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionResultException.java
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import java.util.ArrayList;
+import java.util.List;
+
+final class HmsPartitionResultException extends HMSClientException {
+
+ private static final int MAX_SAMPLES_PER_TYPE = 10;
+ private static final int MAX_SAMPLE_LENGTH = 256;
+
+ private HmsPartitionResultException(Builder builder) {
+ super("Invalid HMS partition result: requested=%d, returned=%d, "
+ + "missing=%d, duplicate=%d, unexpected=%d,
invalid=%d, "
+ + "missingSamples=%s, duplicateSamples=%s,
unexpectedSamples=%s, invalidSamples=%s",
+ builder.requestedCount, builder.returnedCount,
+ builder.missingCount, builder.duplicateCount,
builder.unexpectedCount, builder.invalidCount,
+ builder.missingSamples, builder.duplicateSamples,
+ builder.unexpectedSamples, builder.invalidSamples);
+ }
+
+ static Builder builder(int requestedCount, int returnedCount) {
+ return new Builder(requestedCount, returnedCount);
+ }
+
+ static final class Builder {
+ private final int requestedCount;
+ private final int returnedCount;
+ private final List<String> missingSamples = new ArrayList<>();
+ private final List<String> duplicateSamples = new ArrayList<>();
+ private final List<String> unexpectedSamples = new ArrayList<>();
+ private final List<String> invalidSamples = new ArrayList<>();
+ private int missingCount;
+ private int duplicateCount;
+ private int unexpectedCount;
+ private int invalidCount;
+
+ private Builder(int requestedCount, int returnedCount) {
+ this.requestedCount = requestedCount;
+ this.returnedCount = returnedCount;
+ }
+
+ Builder missing(String sample) {
+ missingCount++;
+ addSample(missingSamples, sample);
+ return this;
+ }
+
+ Builder duplicate(String sample) {
+ duplicateCount++;
+ addSample(duplicateSamples, sample);
+ return this;
+ }
+
+ Builder unexpected(String sample) {
+ unexpectedCount++;
+ addSample(unexpectedSamples, sample);
+ return this;
+ }
+
+ Builder invalid(String sample) {
+ invalidCount++;
+ addSample(invalidSamples, sample);
+ return this;
+ }
+
+ boolean hasMismatches() {
+ return missingCount + duplicateCount + unexpectedCount +
invalidCount > 0;
+ }
+
+ HmsPartitionResultException build() {
+ return new HmsPartitionResultException(this);
+ }
+
+ private static void addSample(List<String> samples, String sample) {
+ if (samples.size() < MAX_SAMPLES_PER_TYPE) {
+ samples.add(sample.length() <= MAX_SAMPLE_LENGTH
+ ? sample : sample.substring(0, MAX_SAMPLE_LENGTH - 3)
+ "...");
+ }
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionTransport.java
similarity index 66%
copy from
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
copy to
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionTransport.java
index 6e714e20a9b..afa951f7d9f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSClientException.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionTransport.java
@@ -17,15 +17,13 @@
package org.apache.doris.datasource.hive;
-import org.apache.doris.common.util.Util;
+import org.apache.hadoop.hive.metastore.api.Partition;
-public class HMSClientException extends RuntimeException {
- public HMSClientException(String format, Throwable cause, Object... msg) {
- super(String.format(format, msg) + (cause == null ? "" : ". reason: "
+ Util.getRootCauseMessage(cause)),
- cause);
- }
+import java.util.List;
- public HMSClientException(String format, Object... msg) {
- super(String.format(format, msg));
- }
+/** Leaf transport contract: one invocation enters the configured
getPartitionsByNames transport once. */
+@FunctionalInterface
+interface HmsPartitionTransport {
+ List<Partition> getPartitionsByNames(
+ String dbName, String tableName, List<String> partitionNames)
throws Exception;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/ThriftHMSCachedClient.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/ThriftHMSCachedClient.java
index 46ca8ef3bc8..0ddb0e0f48e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/ThriftHMSCachedClient.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/ThriftHMSCachedClient.java
@@ -98,6 +98,11 @@ public class ThriftHMSCachedClient implements
HMSCachedClient {
private final HiveConf hiveConf;
private final ExecutionAuthenticator executionAuthenticator;
private final MetaStoreClientProvider metaStoreClientProvider;
+ private final int partitionBatchSize;
+
+ /** Maximum partition names sent by one getPartitionsByNames RPC; a
`hive.` catalog property. */
+ public static final String PARTITION_BATCH_SIZE_KEY =
"hive.hms_partitions_batch_size_per_rpc";
+ public static final int DEFAULT_PARTITION_BATCH_SIZE = 5000;
public ThriftHMSCachedClient(HiveConf hiveConf, int poolSize,
ExecutionAuthenticator executionAuthenticator) {
this(hiveConf, poolSize, executionAuthenticator, new
DefaultMetaStoreClientProvider());
@@ -111,6 +116,29 @@ public class ThriftHMSCachedClient implements
HMSCachedClient {
this.metaStoreClientProvider =
Preconditions.checkNotNull(metaStoreClientProvider, "metaStoreClientProvider");
this.clientPool = poolSize == 0 ? null
: new GenericObjectPool<>(new ThriftHMSClientFactory(),
createPoolConfig(poolSize));
+ this.partitionBatchSize = parsePartitionBatchSize(hiveConf);
+ }
+
+ private static int parsePartitionBatchSize(HiveConf hiveConf) {
+ return parsePartitionBatchSize(hiveConf == null ? null :
hiveConf.get(PARTITION_BATCH_SIZE_KEY));
+ }
+
+ static int parsePartitionBatchSize(String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return DEFAULT_PARTITION_BATCH_SIZE;
+ }
+ int parsed;
+ try {
+ parsed = Integer.parseInt(value.trim());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ PARTITION_BATCH_SIZE_KEY + " must be a positive integer,
got " + value, e);
+ }
+ if (parsed <= 0) {
+ throw new IllegalArgumentException(
+ PARTITION_BATCH_SIZE_KEY + " must be a positive integer,
got " + value);
+ }
+ return parsed;
}
@Override
@@ -344,23 +372,47 @@ public class ThriftHMSCachedClient implements
HMSCachedClient {
@Override
public List<Partition> getPartitions(String dbName, String tblName,
List<String> partitionNames) {
+ // Strict form: every requested name must come back, exactly once, in
request order. One batch
+ // executor owns bounded chunking, adaptive size fallback and response
validation; this method
+ // only supplies the leaf transport that performs one physical
getPartitionsByNames per attempt.
+ return newPartitionBatchExecutor()
+ .executeWithStats(new HmsPartitionRequest(dbName, tblName,
partitionNames))
+ .getPartitions();
+ }
+
+ @Override
+ public List<Partition> getExistingPartitions(String dbName, String
tblName, List<String> partitionNames) {
+ // Lenient form for callers racing remote DROPs (cache bulk loads,
freshness probes): a missing
+ // name is a normal answer and is omitted;
duplicate/unexpected/invalid responses still fail.
+ return newPartitionBatchExecutor()
+ .executeExistingWithStats(new HmsPartitionRequest(dbName,
tblName, partitionNames))
+ .getPartitions();
+ }
+
+ private HmsPartitionBatchExecutor newPartitionBatchExecutor() {
+ return new HmsPartitionBatchExecutor(partitionBatchSize,
this::fetchPartitionsByNames);
+ }
+
+ private List<Partition> fetchPartitionsByNames(String dbName, String
tblName, List<String> partitionNames) {
+ if (isClosed) {
+ throw new HMSClientException("HMS client is closed");
+ }
try (ThriftHMSClient client = getClient()) {
try {
return ugiDoAs(() ->
client.client.getPartitionsByNames(dbName, tblName, partitionNames));
} catch (Exception e) {
client.setThrowable(e);
- throw e;
- }
+ // Everything reaching this catch crossed into the remote call
(pool/auth setup failures
+ // throw from getClient() before this block), so classify it
for the batch executor's
+ // adaptive-fallback ladder. ugiDoAs wraps the real failure in
a RuntimeException; unwrap
+ // so thrift message-size causes stay visible to the
degradable classification.
+ Throwable cause = e instanceof RuntimeException &&
e.getCause() != null ? e.getCause() : e;
+ throw new
HmsPartitionBatchExecutor.RemoteCallException(cause.getMessage(), cause);
+ }
+ } catch (HMSClientException e) {
+ throw e;
} catch (Exception e) {
- // Avoid printing too much log
- String partitionNamesMsg;
- if (partitionNames.size() <= 3) {
- partitionNamesMsg = partitionNames.toString();
- } else {
- partitionNamesMsg = partitionNames.subList(0, 3) + "... total:
" + partitionNames.size();
- }
- throw new HMSClientException("failed to get partitions for table
%s in db %s with value [%s]", e, tblName,
- dbName, partitionNamesMsg);
+ throw new HMSClientException("failed to get partitions for table
%s in db %s", e, tblName, dbName);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index 02366f011e1..9aceec751ef 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -257,6 +257,17 @@ public class MTMVTask extends AbstractTask {
}
Map<TableIf, String> tableWithPartKey = getIncrementalTableMap();
this.completedPartitions = Lists.newCopyOnWriteArrayList();
+ try {
+ // Snapshot persistence happens after refresh partitions are
split into execution groups. Load the
+ // complete union here so the default one-partition group size
cannot turn a large Hive MTMV into
+ // one metadata request per MV partition;
generatePartitionSnapshots reuses this context cache.
+
context.preparePartitionSnapshots(Sets.newHashSet(needRefreshPartitions));
+ } catch (Exception e) {
+ // Preloading is only a batching optimization. Retrying
through the existing per-group load below
+ // preserves completed-group progress when a later chunk of
the union fails.
+ LOG.warn("Failed to preload partition snapshots for mv={},
taskId={}; "
+ + "falling back to per-group loading", mtmv.getName(),
getTaskId(), e);
+ }
int refreshPartitionNum = mtmv.getRefreshPartitionNum();
long execNum = (needRefreshPartitions.size() /
refreshPartitionNum) + ((needRefreshPartitions.size()
% refreshPartitionNum) > 0 ? 1 : 0);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
index 1b2dc56a9c8..98c279a6ac4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
@@ -37,6 +37,7 @@ import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
import org.apache.doris.datasource.mvcc.MvccUtil;
import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVRefreshContext.PreparedPartitionSnapshots;
import org.apache.doris.rpc.RpcException;
import com.google.common.base.Preconditions;
@@ -92,7 +93,8 @@ public class MTMVPartitionUtil {
* @return
* @throws AnalysisException
*/
- public static boolean isMTMVPartitionSync(MTMVRefreshContext
refreshContext, String partitionName,
+ public static boolean isMTMVPartitionSync(MTMVRefreshContext
refreshContext,
+ PreparedPartitionSnapshots partitionSnapshots, String
partitionName,
Set<BaseTableInfo> tables,
Set<TableName> excludedTriggerTables) throws AnalysisException {
MTMV mtmv = refreshContext.getMtmv();
@@ -109,7 +111,8 @@ public class MTMVPartitionUtil {
Set<String> relatedPartitionNames =
partitionMappings.getOrDefault(pctTable, Sets.newHashSet());
// if follow base table, not need compare with related table,
only should compare with related partition
excludedTriggerTablesToCheck.add(new TableName(pctTable));
- if (!isSyncWithPartitions(refreshContext, partitionName,
relatedPartitionNames, pctTable)) {
+ if (!isSyncWithPartitions(
+ refreshContext, partitionSnapshots, partitionName,
relatedPartitionNames, pctTable)) {
return false;
}
}
@@ -227,8 +230,10 @@ public class MTMVPartitionUtil {
throws AnalysisException {
MTMV mtmv = context.getMtmv();
Set<String> partitionNames = mtmv.getPartitionNames();
+ PreparedPartitionSnapshots partitionSnapshots =
+ context.prepareComparablePartitionSnapshots(partitionNames);
for (String partitionName : partitionNames) {
- if (!isMTMVPartitionSync(context, partitionName, tables,
+ if (!isMTMVPartitionSync(context, partitionSnapshots,
partitionName, tables,
excludeTables)) {
return false;
}
@@ -248,14 +253,17 @@ public class MTMVPartitionUtil {
List<Long> partitionIds = mtmv.getPartitionIds();
Map<Long, List<String>> res = Maps.newHashMap();
MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ PreparedPartitionSnapshots partitionSnapshots =
+
context.prepareComparablePartitionSnapshots(mtmv.getPartitionNames());
for (Long partitionId : partitionIds) {
String partitionName =
mtmv.getPartitionOrAnalysisException(partitionId).getName();
- res.put(partitionId, getPartitionUnSyncTables(context,
partitionName));
+ res.put(partitionId, getPartitionUnSyncTables(context,
partitionSnapshots, partitionName));
}
return res;
}
- private static List<String> getPartitionUnSyncTables(MTMVRefreshContext
context, String partitionName)
+ private static List<String> getPartitionUnSyncTables(MTMVRefreshContext
context,
+ PreparedPartitionSnapshots partitionSnapshots, String
partitionName)
throws AnalysisException {
MTMV mtmv = context.getMtmv();
Map<MTMVRelatedTableIf, Set<String>> mappings =
context.getByPartitionName(partitionName);
@@ -273,8 +281,8 @@ public class MTMVPartitionUtil {
if (mtmv.getMvPartitionInfo().getPartitionType() !=
MTMVPartitionType.SELF_MANAGE && pctTables.contains(
pctTable)) {
Set<String> pctPartitions = mappings.getOrDefault(pctTable,
Sets.newHashSet());
- boolean isSyncWithPartition = isSyncWithPartitions(context,
partitionName,
- pctPartitions, pctTable);
+ boolean isSyncWithPartition = isSyncWithPartitions(
+ context, partitionSnapshots, partitionName,
pctPartitions, pctTable);
if (!isSyncWithPartition) {
res.add(pctTable.getName());
}
@@ -298,9 +306,16 @@ public class MTMVPartitionUtil {
MTMV mtmv = context.getMtmv();
Set<String> partitionNames = mtmv.getPartitionNames();
List<String> res = Lists.newArrayList();
+ PreparedPartitionSnapshots partitionSnapshots;
+ try {
+ partitionSnapshots =
context.prepareComparablePartitionSnapshots(partitionNames);
+ } catch (AnalysisException e) {
+ LOG.warn("preload partition snapshots failed", e);
+ return Lists.newArrayList(partitionNames);
+ }
for (String partitionName : partitionNames) {
try {
- if (!isMTMVPartitionSync(context, partitionName, baseTables,
+ if (!isMTMVPartitionSync(context, partitionSnapshots,
partitionName, baseTables,
mtmv.getExcludedTriggerTables())) {
res.add(partitionName);
}
@@ -321,7 +336,8 @@ public class MTMVPartitionUtil {
* @return
* @throws AnalysisException
*/
- public static boolean isSyncWithPartitions(MTMVRefreshContext context,
String mtmvPartitionName,
+ public static boolean isSyncWithPartitions(MTMVRefreshContext context,
+ PreparedPartitionSnapshots partitionSnapshots, String
mtmvPartitionName,
Set<String> pctPartitionNames, MTMVRelatedTableIf pctTable) throws
AnalysisException {
MTMV mtmv = context.getMtmv();
if (!pctTable.needAutoRefresh()) {
@@ -339,8 +355,7 @@ public class MTMVPartitionUtil {
return true;
}
for (String pctPartitionName : pctPartitionNames) {
- MTMVSnapshotIf pctCurrentSnapshot = pctTable
- .getPartitionSnapshot(pctPartitionName, context,
MvccUtil.getSnapshotFromContext(pctTable));
+ MTMVSnapshotIf pctCurrentSnapshot =
partitionSnapshots.get(pctTable, pctPartitionName);
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("isSyncWithPartitions mvName is %s\n,
mtmvPartitionName is %s\n, "
+ "mtmv refreshSnapshot is %s\n,
pctPartitionName is %s\n, "
@@ -551,8 +566,8 @@ public class MTMVPartitionUtil {
if (baseTableSnapshotCache.containsKey(baseTableInfo)) {
return baseTableSnapshotCache.get(baseTableInfo);
}
- MTMVSnapshotIf baseTableCurrentSnapshot =
mtmvRelatedTableIf.getTableSnapshot(context,
- MvccUtil.getSnapshotFromContext(mtmvRelatedTableIf));
+ MTMVSnapshotIf baseTableCurrentSnapshot =
mtmvRelatedTableIf.getTableSnapshot(
+ context, context.resolveSnapshot(mtmvRelatedTableIf));
baseTableSnapshotCache.put(baseTableInfo, baseTableCurrentSnapshot);
return baseTableCurrentSnapshot;
}
@@ -569,18 +584,19 @@ public class MTMVPartitionUtil {
public static Map<String, MTMVRefreshPartitionSnapshot>
generatePartitionSnapshots(MTMVRefreshContext context,
Set<BaseTableInfo> baseTables, Set<String> partitionNames)
throws AnalysisException {
+ PreparedPartitionSnapshots preparedSnapshots =
context.preparePartitionSnapshots(partitionNames);
Map<String, MTMVRefreshPartitionSnapshot> res = Maps.newHashMap();
for (String partitionName : partitionNames) {
- res.put(partitionName,
- generatePartitionSnapshot(context, baseTables,
-
context.getPartitionMappings().get(partitionName)));
+ res.put(partitionName, generatePartitionSnapshot(context,
preparedSnapshots, baseTables,
+ context.getPartitionMappings().get(partitionName)));
}
return res;
}
private static MTMVRefreshPartitionSnapshot
generatePartitionSnapshot(MTMVRefreshContext context,
- Set<BaseTableInfo> baseTables, Map<MTMVRelatedTableIf,
Set<String>> pctPartitionNames)
+ PreparedPartitionSnapshots partitionSnapshots, Set<BaseTableInfo>
baseTables,
+ Map<MTMVRelatedTableIf, Set<String>> pctPartitionNames)
throws AnalysisException {
MTMV mtmv = context.getMtmv();
MTMVRefreshPartitionSnapshot refreshPartitionSnapshot = new
MTMVRefreshPartitionSnapshot();
@@ -594,8 +610,7 @@ public class MTMVPartitionUtil {
continue;
}
for (String pctPartitionName : oneTablePartitionNames) {
- MTMVSnapshotIf partitionSnapshot =
pctTable.getPartitionSnapshot(pctPartitionName, context,
- MvccUtil.getSnapshotFromContext(pctTable));
+ MTMVSnapshotIf partitionSnapshot =
partitionSnapshots.get(pctTable, pctPartitionName);
pctSnapshot.put(pctPartitionName, partitionSnapshot);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
index 603f89bbec5..77fe3312119 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java
@@ -19,10 +19,17 @@ package org.apache.doris.mtmv;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import org.apache.doris.datasource.mvcc.MvccUtil;
import com.google.common.collect.Maps;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
public class MTMVRefreshContext {
@@ -33,6 +40,10 @@ public class MTMVRefreshContext {
// Hence, the results are cached at this stage.
// The value is loaded/cached on the first fetch
private Map<BaseTableInfo, MTMVSnapshotIf> baseTableSnapshotCache =
Maps.newHashMap();
+ private final Map<MTMVRelatedTableIf, Map<String, MTMVSnapshotIf>>
partitionSnapshotCache = Maps.newHashMap();
+ private final Map<MTMVRelatedTableIf, Set<String>>
missingPartitionSnapshotCache = Maps.newHashMap();
+ private final Map<MTMVRelatedTableIf, Map<String, AnalysisException>>
partitionSnapshotFailureCache =
+ Maps.newHashMap();
public MTMVRefreshContext(MTMV mtmv) {
this.mtmv = mtmv;
@@ -58,6 +69,78 @@ public class MTMVRefreshContext {
return baseTableSnapshotCache;
}
+ /** Loads the union of mapped base partitions once per related table. */
+ public PreparedPartitionSnapshots preparePartitionSnapshots(Set<String>
mtmvPartitionNames)
+ throws AnalysisException {
+ return preparePartitionSnapshots(mtmvPartitionNames, false);
+ }
+
+ /** Loads only mappings whose persisted partition-name set still matches
and needs version comparison. */
+ public PreparedPartitionSnapshots
prepareComparablePartitionSnapshots(Set<String> mtmvPartitionNames)
+ throws AnalysisException {
+ return preparePartitionSnapshots(mtmvPartitionNames, true);
+ }
+
+ private PreparedPartitionSnapshots preparePartitionSnapshots(
+ Set<String> mtmvPartitionNames, boolean comparableOnly)
+ throws AnalysisException {
+ Map<MTMVRelatedTableIf, Set<String>> namesByTable = new
LinkedHashMap<>();
+ Map<MTMVRelatedTableIf, BaseTableInfo> tableInfos = comparableOnly
+ ? new LinkedHashMap<>() : Collections.emptyMap();
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry
+ : getByPartitionName(mtmvPartitionName).entrySet()) {
+ if (!entry.getKey().needAutoRefresh()) {
+ continue;
+ }
+ if (comparableOnly) {
+ BaseTableInfo tableInfo =
tableInfos.computeIfAbsent(entry.getKey(), BaseTableInfo::new);
+ if (!Objects.equals(entry.getValue(),
mtmv.getRefreshSnapshot()
+ .getPctSnapshots(mtmvPartitionName, tableInfo))) {
+ continue;
+ }
+ }
+ namesByTable.computeIfAbsent(entry.getKey(), ignored -> new
LinkedHashSet<>())
+ .addAll(entry.getValue());
+ }
+ }
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry :
namesByTable.entrySet()) {
+ loadSnapshots(entry.getKey(), entry.getValue());
+ }
+ return new PreparedPartitionSnapshots(this);
+ }
+
+ private void loadSnapshots(MTMVRelatedTableIf table, Set<String>
partitionNames) throws AnalysisException {
+ Map<String, MTMVSnapshotIf> cached =
partitionSnapshotCache.computeIfAbsent(
+ table, ignored -> new LinkedHashMap<>());
+ Set<String> knownMissing =
missingPartitionSnapshotCache.computeIfAbsent(
+ table, ignored -> new LinkedHashSet<>());
+ Set<String> missing = new LinkedHashSet<>(partitionNames);
+ missing.removeAll(cached.keySet());
+ missing.removeAll(knownMissing);
+ if (missing.isEmpty()) {
+ return;
+ }
+ Map<String, MTMVSnapshotIf> loaded = table.getPartitionSnapshots(
+ missing, this, resolveSnapshot(table));
+ if (loaded == null || loaded.containsKey(null) ||
loaded.containsValue(null)
+ || !missing.containsAll(loaded.keySet())) {
+ throw new AnalysisException("Invalid partition snapshot result for
table " + table.getName()
+ + ": requestedCount=" + missing.size() + ", returnedCount="
+ + (loaded == null ? "null" : loaded.size()));
+ }
+ cached.putAll(loaded);
+ Set<String> stillMissing = new LinkedHashSet<>(missing);
+ stillMissing.removeAll(loaded.keySet());
+ knownMissing.addAll(stillMissing);
+ }
+
+ void recordPartitionSnapshotFailure(
+ MTMVRelatedTableIf table, String partitionName, AnalysisException
failure) {
+ partitionSnapshotFailureCache.computeIfAbsent(table, ignored -> new
LinkedHashMap<>())
+ .put(partitionName, failure);
+ }
+
public static MTMVRefreshContext buildContext(MTMV mtmv) throws
AnalysisException {
MTMVRefreshContext context = new MTMVRefreshContext(mtmv);
context.partitionMappings = mtmv.calculatePartitionMappings();
@@ -65,4 +148,34 @@ public class MTMVRefreshContext {
return context;
}
+ Optional<MvccSnapshot> resolveSnapshot(MTMVRelatedTableIf table) {
+ return MvccUtil.getSnapshotFromContext(table);
+ }
+
+ /** Read-only access to partition snapshots that have already been loaded
as one bulk operation. */
+ public static final class PreparedPartitionSnapshots {
+ private final MTMVRefreshContext context;
+
+ private PreparedPartitionSnapshots(MTMVRefreshContext context) {
+ this.context = context;
+ }
+
+ public MTMVSnapshotIf get(MTMVRelatedTableIf table, String
partitionName) throws AnalysisException {
+ Map<String, MTMVSnapshotIf> snapshots =
context.partitionSnapshotCache.get(table);
+ if (snapshots != null && snapshots.containsKey(partitionName)) {
+ return snapshots.get(partitionName);
+ }
+ Map<String, AnalysisException> failures =
context.partitionSnapshotFailureCache.get(table);
+ if (failures != null && failures.containsKey(partitionName)) {
+ throw failures.get(partitionName);
+ }
+ Set<String> missing =
context.missingPartitionSnapshotCache.get(table);
+ if (missing != null && missing.contains(partitionName)) {
+ throw new AnalysisException("can not find partition: " +
partitionName);
+ }
+ throw new AnalysisException("Partition snapshot was not prepared:
table=" + table.getName()
+ + ", partition=" + partitionName);
+ }
+ }
+
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedTableIf.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedTableIf.java
index 7f38a0492c2..ee0bbb8b6da 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedTableIf.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedTableIf.java
@@ -25,6 +25,7 @@ import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -85,6 +86,24 @@ public interface MTMVRelatedTableIf extends TableIf {
MTMVSnapshotIf getPartitionSnapshot(String partitionName,
MTMVRefreshContext context,
Optional<MvccSnapshot> snapshot) throws AnalysisException;
+ /**
+ * Loads partition snapshots in bulk when the table supports it. The
compatibility default retains the
+ * original one-at-a-time behavior while retaining failures by partition
name in the refresh context;
+ * plugin-driven external tables override it to reach connector batching.
+ */
+ default Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String>
partitionNames,
+ MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
throws AnalysisException {
+ Map<String, MTMVSnapshotIf> snapshots = new LinkedHashMap<>();
+ for (String partitionName : partitionNames) {
+ try {
+ snapshots.put(partitionName,
getPartitionSnapshot(partitionName, context, snapshot));
+ } catch (AnalysisException e) {
+ context.recordPartitionSnapshotFailure(this, partitionName, e);
+ }
+ }
+ return snapshots;
+ }
+
/**
* getTableSnapshot
* It is best to use the version. If there is no version, use the last
update time
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java
index 6dc0bd24da7..7a55fc12784 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Pair;
import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVRefreshContext.PreparedPartitionSnapshots;
import org.apache.doris.qe.ConnectContext;
import com.google.common.annotations.VisibleForTesting;
@@ -65,6 +66,7 @@ public class MTMVRewriteUtil {
}
Set<String> mtmvNeedComparePartitions = null;
MTMVRefreshContext refreshContext = null;
+ PreparedPartitionSnapshots partitionSnapshots = null;
// check gracePeriod
long gracePeriodMills = mtmv.getGracePeriod();
for (Partition partition : allPartitions) {
@@ -95,8 +97,25 @@ public class MTMVRewriteUtil {
if (!mtmvNeedComparePartitions.contains(partition.getName())) {
continue;
}
+ if (partitionSnapshots == null) {
+ Set<String> partitionsToPreload = Sets.newHashSet();
+ for (Partition candidate : allPartitions) {
+ boolean withinGracePeriod = gracePeriodMills > 0
+ && currentTimeMills <=
candidate.getVisibleVersionTime() + gracePeriodMills
+ && !forceConsistent;
+ if (!withinGracePeriod &&
mtmvNeedComparePartitions.contains(candidate.getName())) {
+ partitionsToPreload.add(candidate.getName());
+ }
+ }
+ try {
+ partitionSnapshots =
refreshContext.prepareComparablePartitionSnapshots(partitionsToPreload);
+ } catch (AnalysisException e) {
+ LOG.warn("preload partition snapshots failed", e);
+ return res;
+ }
+ }
try {
- if (MTMVPartitionUtil.isMTMVPartitionSync(refreshContext,
partition.getName(),
+ if (MTMVPartitionUtil.isMTMVPartitionSync(refreshContext,
partitionSnapshots, partition.getName(),
mtmvRelation.getBaseTablesOneLevelAndFromView(),
forceConsistent ? ImmutableSet.of() :
mtmv.getQueryRewriteConsistencyRelaxedTables())) {
res.add(partition);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutorTest.java
new file mode 100644
index 00000000000..3421bd36092
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsPartitionBatchExecutorTest.java
@@ -0,0 +1,282 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.hive;
+
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class HmsPartitionBatchExecutorTest {
+
+ @Test
+ public void boundsLargeRequestAndRestoresOrder() {
+ List<Integer> batchSizes = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(5000, (db, table, names)
-> {
+ batchSizes.add(names.size());
+ List<Partition> result = infos(names);
+ Collections.reverse(result);
+ return result;
+ });
+
+ List<String> names = names(120_000);
+ List<Partition> result =
executor.executeWithStats(request(names)).getPartitions();
+
+ Assertions.assertEquals(24, batchSizes.size());
+ Assertions.assertTrue(batchSizes.stream().allMatch(size -> size ==
5000));
+
Assertions.assertEquals(names.stream().map(HmsPartitionBatchExecutorTest::values)
+ .collect(Collectors.toList()),
+
result.stream().map(Partition::getValues).collect(Collectors.toList()));
+ }
+
+ @Test
+ public void handlesEmptyExactAndTrailingBatches() {
+ List<Integer> batchSizes = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(3, (db, table, names) ->
{
+ batchSizes.add(names.size());
+ return infos(names);
+ });
+
+
Assertions.assertTrue(executor.executeWithStats(request(Collections.emptyList())).getPartitions().isEmpty());
+ Assertions.assertEquals(6,
executor.executeWithStats(request(names(6))).getPartitions().size());
+ Assertions.assertEquals(7,
executor.executeWithStats(request(names(7))).getPartitions().size());
+ Assertions.assertEquals(Arrays.asList(3, 3, 3, 3, 1), batchSizes);
+ }
+
+ @Test
+ public void halvesOversizeBatchAndReusesSafeSize() {
+ List<Integer> attempts = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(8, (db, table, names) ->
{
+ attempts.add(names.size());
+ if (names.size() > 2) {
+ throw remoteFailure("frame too large");
+ }
+ return infos(names);
+ });
+
+ Assertions.assertEquals(10,
executor.executeWithStats(request(names(10))).getPartitions().size());
+ Assertions.assertEquals(Arrays.asList(8, 4, 2, 2, 2, 2, 2), attempts);
+ }
+
+ @Test
+ public void reportsPhysicalBatchExecutionStats() {
+ HmsPartitionBatchExecutor executor = executor(4, (db, table, names) ->
{
+ if (names.size() > 2) {
+ throw remoteFailure("frame too large");
+ }
+ return infos(names);
+ });
+
+ HmsPartitionBatchResult result =
executor.executeWithStats(request(names(5)));
+ HmsPartitionBatchStats stats = result.getStats();
+
+ Assertions.assertEquals(5, result.getPartitions().size());
+ Assertions.assertEquals(5, stats.getRequestedItems());
+ Assertions.assertEquals(4, stats.getTransportInvocations());
+ Assertions.assertEquals(9, stats.getTransportItems());
+ Assertions.assertEquals(4, stats.getLargestBatchSize());
+ Assertions.assertEquals(1, stats.getSmallestBatchSize());
+ Assertions.assertEquals(1, stats.getFallbackCount());
+ Assertions.assertTrue(stats.getLogicalElapsedNanos() >=
stats.getTransportElapsedNanos());
+ Assertions.assertTrue(stats.getTransportElapsedNanos() >=
stats.getMaxTransportElapsedNanos());
+ }
+
+ @Test
+ public void minimumBatchFailurePropagates() {
+ List<Integer> attempts = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(2, (db, table, names) ->
{
+ attempts.add(names.size());
+ throw remoteFailure("max message size reached");
+ });
+
+ HMSClientException failure = Assertions.assertThrows(
+ HMSClientException.class, () ->
executor.executeWithStats(request(names(2))));
+ Assertions.assertEquals(Arrays.asList(2, 1), attempts);
+
Assertions.assertTrue(failure.getMessage().contains("failedBatchSize=1"));
+
Assertions.assertTrue(failure.getMessage().contains("transportInvocations=2"));
+ HmsPartitionBatchStats stats = failure.getPartitionBatchStats();
+ Assertions.assertNotNull(stats);
+ Assertions.assertEquals(2, stats.getRequestedItems());
+ Assertions.assertEquals(2, stats.getTransportInvocations());
+ Assertions.assertEquals(3, stats.getTransportItems());
+ Assertions.assertEquals(2, stats.getLargestBatchSize());
+ Assertions.assertEquals(1, stats.getSmallestBatchSize());
+ Assertions.assertEquals(1, stats.getFallbackCount());
+ Assertions.assertTrue(stats.getLogicalElapsedNanos() >=
stats.getTransportElapsedNanos());
+ }
+
+ @Test
+ public void ordinaryTransportFailureDoesNotFallback() {
+ List<Integer> attempts = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(8, (db, table, names) ->
{
+ attempts.add(names.size());
+ throw remoteFailure("connection refused");
+ });
+
+ HMSClientException failure = Assertions.assertThrows(
+ HMSClientException.class, () ->
executor.executeWithStats(request(names(8))));
+ Assertions.assertEquals(Collections.singletonList(8), attempts);
+ HmsPartitionBatchStats stats = failure.getPartitionBatchStats();
+ Assertions.assertNotNull(stats);
+ Assertions.assertEquals(8, stats.getRequestedItems());
+ Assertions.assertEquals(1, stats.getTransportInvocations());
+ Assertions.assertEquals(8, stats.getTransportItems());
+ Assertions.assertEquals(8, stats.getLargestBatchSize());
+ Assertions.assertEquals(8, stats.getSmallestBatchSize());
+ Assertions.assertEquals(0, stats.getFallbackCount());
+ }
+
+ @Test
+ public void halvesForHivePartitionRequestLimitMessage() {
+ List<Integer> attempts = new ArrayList<>();
+ HmsPartitionBatchExecutor executor = executor(4, (db, table, names) ->
{
+ attempts.add(names.size());
+ if (names.size() > 2) {
+ throw remoteFailure("Number of partitions scanned (4) exceeds
limit (2). "
+ + "This is controlled on the metastore server by
hive.metastore.limit.partition.request");
+ }
+ return infos(names);
+ });
+
+ Assertions.assertEquals(4,
executor.executeWithStats(request(names(4))).getPartitions().size());
+ Assertions.assertEquals(Arrays.asList(4, 2, 2), attempts);
+ }
+
+ @Test
+ public void reportsAllResultMismatchCategoriesPrecisely() {
+ HmsPartitionBatchExecutor executor = executor(10, (db, table, names)
-> Arrays.asList(
+ info("a"), info("c"), info("c")));
+
+ HmsPartitionResultException failure = Assertions.assertThrows(
+ HmsPartitionResultException.class,
+ () -> executor.executeWithStats(request(Arrays.asList("p=a",
"p=b"))));
+ Assertions.assertTrue(failure.getMessage().contains("missing=1"));
+ Assertions.assertTrue(failure.getMessage().contains("duplicate=1"));
+ Assertions.assertTrue(failure.getMessage().contains("unexpected=1"));
+
Assertions.assertTrue(failure.getMessage().contains("missingSamples=[p=b]"));
+
Assertions.assertTrue(failure.getMessage().contains("duplicateSamples=[[c]]"));
+
Assertions.assertTrue(failure.getMessage().contains("unexpectedSamples=[[c]]"));
+ }
+
+ @Test
+ public void existingPartitionModeOmitsOnlyMissingResults() {
+ HmsPartitionBatchExecutor executor = executor(10, (db, table, names) ->
+ Arrays.asList(info("c"), info("a")));
+
+ List<Partition> existing = executor.executeExistingWithStats(
+ request(Arrays.asList("p=a", "p=b", "p=c"))).getPartitions();
+
+ Assertions.assertEquals(Arrays.asList("a", "c"), existing.stream()
+ .map(partition ->
partition.getValues().get(0)).collect(Collectors.toList()));
+ }
+
+ @Test
+ public void
existingPartitionModeStillRejectsUnexpectedAndDuplicateResults() {
+ HmsPartitionBatchExecutor executor = executor(10, (db, table, names) ->
+ Arrays.asList(info("a"), info("a"), info("unexpected")));
+
+ HmsPartitionResultException failure = Assertions.assertThrows(
+ HmsPartitionResultException.class,
+ () ->
executor.executeExistingWithStats(request(Arrays.asList("p=a", "p=missing"))));
+
+ Assertions.assertTrue(failure.getMessage().contains("missing=0"));
+ Assertions.assertTrue(failure.getMessage().contains("duplicate=1"));
+ Assertions.assertTrue(failure.getMessage().contains("unexpected=1"));
+ }
+
+ @Test
+ public void rejectsNullAndMalformedResults() {
+ HmsPartitionBatchExecutor nullResponse = executor(10, (db, table,
names) -> null);
+ HmsPartitionResultException nullFailure = Assertions.assertThrows(
+ HmsPartitionResultException.class,
+ () ->
nullResponse.executeWithStats(request(Collections.singletonList("p=a"))));
+ Assertions.assertTrue(nullFailure.getMessage().contains("invalid=1"));
+
+ HmsPartitionBatchExecutor malformed = executor(10, (db, table, names)
-> Collections.singletonList(
+ info(Arrays.asList("a", "extra"))));
+ HmsPartitionResultException malformedFailure = Assertions.assertThrows(
+ HmsPartitionResultException.class,
+ () ->
malformed.executeWithStats(request(Collections.singletonList("p=a"))));
+
Assertions.assertTrue(malformedFailure.getMessage().contains("invalid=1"));
+
Assertions.assertTrue(malformedFailure.getMessage().contains("missing=1"));
+ }
+
+ @Test
+ public void validatesLogicalRequest() {
+ Assertions.assertEquals(Collections.singletonList(""), values("p="));
+ Assertions.assertEquals(Arrays.asList("a/b", "x=y"),
+ values("p=a%2Fb/q=x%3Dy"));
+ Assertions.assertDoesNotThrow(() -> request(Arrays.asList("P=a",
"p=b")));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
request(Arrays.asList("p=a", "q=b")));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
request(Collections.singletonList("p=a/")));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
request(Arrays.asList("p=a", "p=a")));
+ }
+
+ @Test
+ public void parsesAndValidatesConfiguration() {
+ Assertions.assertEquals(5000,
ThriftHMSCachedClient.parsePartitionBatchSize((String) null));
+ Assertions.assertEquals(5000,
ThriftHMSCachedClient.parsePartitionBatchSize(" "));
+ Assertions.assertEquals(321,
ThriftHMSCachedClient.parsePartitionBatchSize("321"));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> ThriftHMSCachedClient.parsePartitionBatchSize("0"));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () ->
ThriftHMSCachedClient.parsePartitionBatchSize("not-a-number"));
+ }
+
+ private static HmsPartitionBatchExecutor executor(int batchSize,
HmsPartitionTransport transport) {
+ return new HmsPartitionBatchExecutor(batchSize, transport);
+ }
+
+ private static HmsPartitionRequest request(List<String> names) {
+ return new HmsPartitionRequest("db", "table", names);
+ }
+
+ private static List<String> values(String name) {
+ return
request(Collections.singletonList(name)).getPartitions().get(0).getValues();
+ }
+
+ private static HmsPartitionBatchExecutor.RemoteCallException
remoteFailure(String message) {
+ return new HmsPartitionBatchExecutor.RemoteCallException(
+ "remote failure", new
shade.doris.hive.org.apache.thrift.TException(message));
+ }
+
+ private static List<String> names(int count) {
+ return IntStream.range(0, count).mapToObj(i -> "p=" +
i).collect(Collectors.toList());
+ }
+
+ private static List<Partition> infos(List<String> names) {
+ return
names.stream().map(HmsPartitionBatchExecutorTest::values).map(HmsPartitionBatchExecutorTest::info)
+ .collect(Collectors.toList());
+ }
+
+ private static Partition info(String value) {
+ return info(Collections.singletonList(value));
+ }
+
+ private static Partition info(List<String> values) {
+ Partition partition = new Partition();
+ partition.setValues(new ArrayList<>(values));
+ return partition;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
index 24f0fcf88a8..7a1b66d36f5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
@@ -69,6 +69,8 @@ public class MTMVPartitionUtilTest {
@Mocked
private MTMVRefreshContext context;
@Mocked
+ private MTMVRefreshContext.PreparedPartitionSnapshots partitionSnapshots;
+ @Mocked
private MTMVBaseVersions versions;
private Set<BaseTableInfo> baseTables = Sets.newHashSet();
@@ -150,6 +152,18 @@ public class MTMVPartitionUtilTest {
minTimes = 0;
result = baseSnapshotIf;
+ context.prepareComparablePartitionSnapshots((Set<String>) any);
+ minTimes = 0;
+ result = partitionSnapshots;
+
+ context.preparePartitionSnapshots((Set<String>) any);
+ minTimes = 0;
+ result = partitionSnapshots;
+
+ partitionSnapshots.get((MTMVRelatedTableIf) any, anyString);
+ minTimes = 0;
+ result = baseSnapshotIf;
+
refreshSnapshot.equalsWithPct(anyString, anyString,
(MTMVSnapshotIf) any,
(BaseTableInfo) any);
minTimes = 0;
@@ -208,7 +222,8 @@ public class MTMVPartitionUtilTest {
@Test
public void testIsSyncWithPartition() throws AnalysisException {
boolean isSyncWithPartition = MTMVPartitionUtil
- .isSyncWithPartitions(context, "name1",
Sets.newHashSet("name2"), baseOlapTable);
+ .isSyncWithPartitions(context, partitionSnapshots, "name1",
+ Sets.newHashSet("name2"), baseOlapTable);
Assert.assertTrue(isSyncWithPartition);
}
@@ -222,7 +237,8 @@ public class MTMVPartitionUtilTest {
}
};
boolean isSyncWithPartition = MTMVPartitionUtil
- .isSyncWithPartitions(context, "name1",
Sets.newHashSet("name2"), baseOlapTable);
+ .isSyncWithPartitions(context, partitionSnapshots, "name1",
+ Sets.newHashSet("name2"), baseOlapTable);
Assert.assertFalse(isSyncWithPartition);
}
@@ -237,7 +253,8 @@ public class MTMVPartitionUtilTest {
}
};
boolean isSyncWithPartition = MTMVPartitionUtil
- .isSyncWithPartitions(context, "name1",
Sets.newHashSet("name2"), baseOlapTable);
+ .isSyncWithPartitions(context, partitionSnapshots, "name1",
+ Sets.newHashSet("name2"), baseOlapTable);
Assert.assertFalse(isSyncWithPartition);
}
@@ -262,8 +279,8 @@ public class MTMVPartitionUtilTest {
};
Set<TableName> excludedTriggerTables = ImmutableSet.of();
- boolean isMTMVPartitionSync =
MTMVPartitionUtil.isMTMVPartitionSync(context, "name1", baseTables,
- excludedTriggerTables);
+ boolean isMTMVPartitionSync =
MTMVPartitionUtil.isMTMVPartitionSync(context, partitionSnapshots,
+ "name1", baseTables, excludedTriggerTables);
Assert.assertTrue(isMTMVPartitionSync);
Assert.assertTrue(excludedTriggerTables.isEmpty());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRefreshContextBatchTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRefreshContextBatchTest.java
new file mode 100644
index 00000000000..aa93128e625
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRefreshContextBatchTest.java
@@ -0,0 +1,239 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.mtmv;
+
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.mtmv.MTMVRefreshContext.PreparedPartitionSnapshots;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+public class MTMVRefreshContextBatchTest {
+
+ @Test
+ public void
aggregatesOneHundredSixtyThousandMappedPartitionsIntoOneLogicalLoad()
+ throws AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings = new
LinkedHashMap<>();
+ for (int i = 0; i < 160_000; i++) {
+ mappings.put("mv" + i, Collections.singletonMap(table,
Collections.singleton("p" + i)));
+ }
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ Mockito.when(refreshSnapshot.getPctSnapshots(Mockito.anyString(),
Mockito.any()))
+ .thenAnswer(invocation -> Collections.singleton(
+ "p" + invocation.<String>getArgument(0).substring(2)));
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.any(), Mockito.any()))
+ .thenAnswer(invocation ->
snapshots(invocation.getArgument(0)));
+
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ PreparedPartitionSnapshots prepared =
context.prepareComparablePartitionSnapshots(mappings.keySet());
+
+ Mockito.verify(table).getPartitionSnapshots(
+ Mockito.argThat(names -> names.size() == 160_000),
Mockito.same(context), Mockito.any());
+ Assertions.assertNotNull(prepared.get(table, "p159999"));
+ Mockito.verify(table, Mockito.times(1))
+ .getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any());
+ }
+
+ @Test
+ public void comparablePreloadSkipsLocallyMismatchedPartitionSets() throws
AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings =
Collections.singletonMap(
+ "mv", Collections.singletonMap(table,
Collections.singleton("current")));
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ Mockito.when(refreshSnapshot.getPctSnapshots(Mockito.eq("mv"),
Mockito.any()))
+ .thenReturn(Collections.singleton("persisted"));
+
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ PreparedPartitionSnapshots prepared =
+
context.prepareComparablePartitionSnapshots(Collections.singleton("mv"));
+
+ Mockito.verify(table, Mockito.never())
+ .getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any());
+ Assertions.assertThrows(AnalysisException.class, () ->
prepared.get(table, "current"));
+ }
+
+ @Test
+ public void persistencePreloadLoadsMappingsRegardlessOfPersistedState()
throws AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings = new
LinkedHashMap<>();
+ mappings.put("mv1", Collections.singletonMap(table,
Collections.singleton("p1")));
+ mappings.put("mv2", Collections.singletonMap(table,
Collections.singleton("p2")));
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.any(), Mockito.any()))
+ .thenAnswer(invocation ->
snapshots(invocation.getArgument(0)));
+
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ context.preparePartitionSnapshots(mappings.keySet());
+
+ Mockito.verify(table).getPartitionSnapshots(
+ Mockito.argThat(names -> names.equals(new
LinkedHashSet<>(Arrays.asList("p1", "p2")))),
+ Mockito.same(context), Mockito.any());
+ Mockito.verifyNoInteractions(refreshSnapshot);
+ }
+
+ @Test
+ public void cachesOverlappingLoadsAndRequestsOnlyMissingSnapshots() throws
AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings = new
LinkedHashMap<>();
+ mappings.put("mv1", Collections.singletonMap(table,
Collections.singleton("p1")));
+ mappings.put("mv2", Collections.singletonMap(table,
Collections.singleton("p2")));
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any()))
+ .thenAnswer(invocation ->
snapshots(invocation.getArgument(0)));
+
+ PreparedPartitionSnapshots first =
context.preparePartitionSnapshots(Collections.singleton("mv1"));
+ PreparedPartitionSnapshots second =
context.preparePartitionSnapshots(mappings.keySet());
+ Assertions.assertNotNull(first.get(table, "p1"));
+ Assertions.assertNotNull(second.get(table, "p1"));
+ Assertions.assertNotNull(second.get(table, "p2"));
+
+ Mockito.verify(table).getPartitionSnapshots(
+ Mockito.eq(Collections.singleton("p1")),
Mockito.same(context), Mockito.any());
+ Mockito.verify(table).getPartitionSnapshots(
+ Mockito.eq(Collections.singleton("p2")),
Mockito.same(context), Mockito.any());
+ }
+
+ @Test
+ public void defersMissingBulkResultsToThePartitionThatConsumesThem()
throws AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings = new
LinkedHashMap<>();
+ mappings.put("bad", Collections.singletonMap(table,
Collections.singleton("missing")));
+ mappings.put("good", Collections.singletonMap(table,
Collections.singleton("present")));
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any()))
+ .thenReturn(Collections.singletonMap("present", new
MTMVTimestampSnapshot(1L)));
+
+ PreparedPartitionSnapshots prepared =
context.preparePartitionSnapshots(mappings.keySet());
+
+ Assertions.assertNotNull(prepared.get(table, "present"));
+ AnalysisException failure =
Assertions.assertThrows(AnalysisException.class,
+ () -> prepared.get(table, "missing"));
+ Assertions.assertTrue(failure.getMessage().contains("can not find
partition: missing"));
+ context.preparePartitionSnapshots(mappings.keySet());
+ Mockito.verify(table, Mockito.times(1))
+ .getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any());
+ }
+
+ @Test
+ public void rejectsUnexpectedBulkResults() throws AnalysisException {
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ MTMVRefreshSnapshot refreshSnapshot =
Mockito.mock(MTMVRefreshSnapshot.class);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings =
Collections.singletonMap(
+ "mv", Collections.singletonMap(table,
Collections.singleton("requested")));
+ configureContext(mtmv, table, refreshSnapshot, mappings);
+ MTMVRefreshContext context = MTMVRefreshContext.buildContext(mtmv);
+ Mockito.when(table.getPartitionSnapshots(Mockito.anySet(),
Mockito.same(context), Mockito.any()))
+ .thenReturn(Collections.singletonMap("unexpected", new
MTMVTimestampSnapshot(1L)));
+
+ Assertions.assertThrows(AnalysisException.class,
+ () ->
context.preparePartitionSnapshots(Collections.singleton("mv")));
+ }
+
+ @Test
+ public void defaultBulkAdapterPreservesNonBulkImplementations() throws
AnalysisException {
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class,
Mockito.CALLS_REAL_METHODS);
+ MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
+ Mockito.when(table.getPartitionSnapshot(Mockito.anyString(),
Mockito.same(context), Mockito.any()))
+ .thenAnswer(invocation -> new
MTMVTimestampSnapshot(invocation.<String>getArgument(0).hashCode()));
+
+ Map<String, MTMVSnapshotIf> snapshots = table.getPartitionSnapshots(
+ new LinkedHashSet<>(Arrays.asList("p1", "p2")), context,
Optional.empty());
+
+ Assertions.assertEquals(Arrays.asList("p1", "p2"), new
ArrayList<>(snapshots.keySet()));
+ Mockito.verify(table, Mockito.times(2))
+ .getPartitionSnapshot(Mockito.anyString(),
Mockito.same(context), Mockito.any());
+ }
+
+ @Test
+ public void defaultBulkAdapterDefersPerPartitionFailures() throws
AnalysisException {
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class,
Mockito.CALLS_REAL_METHODS);
+ MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
+ Mockito.when(table.getPartitionSnapshot(Mockito.eq("missing"),
Mockito.same(context), Mockito.any()))
+ .thenThrow(new AnalysisException("missing snapshot"));
+ Mockito.when(table.getPartitionSnapshot(Mockito.eq("present"),
Mockito.same(context), Mockito.any()))
+ .thenReturn(new MTMVTimestampSnapshot(1L));
+
+ Map<String, MTMVSnapshotIf> snapshots = table.getPartitionSnapshots(
+ new LinkedHashSet<>(Arrays.asList("missing", "present")),
context, Optional.empty());
+
+ Assertions.assertEquals(Collections.singleton("present"),
snapshots.keySet());
+ Mockito.verify(context).recordPartitionSnapshotFailure(
+ Mockito.same(table), Mockito.eq("missing"),
Mockito.any(AnalysisException.class));
+ }
+
+ private static void configureContext(MTMV mtmv, MTMVRelatedTableIf table,
+ MTMVRefreshSnapshot refreshSnapshot,
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>> mappings) throws
AnalysisException {
+ MTMVPartitionInfo partitionInfo =
Mockito.mock(MTMVPartitionInfo.class);
+ Mockito.when(mtmv.calculatePartitionMappings())
+ .thenReturn(mappings);
+ Mockito.when(mtmv.getRelation()).thenReturn(null);
+ Mockito.when(mtmv.getMvPartitionInfo()).thenReturn(partitionInfo);
+ Mockito.when(mtmv.getRefreshSnapshot()).thenReturn(refreshSnapshot);
+ Mockito.when(table.needAutoRefresh()).thenReturn(true);
+ configureTableIdentity(table);
+ Mockito.when(partitionInfo.getPartitionType())
+
.thenReturn(MTMVPartitionInfo.MTMVPartitionType.FOLLOW_BASE_TABLE);
+
Mockito.when(partitionInfo.getPctTables()).thenReturn(Collections.singleton(table));
+ }
+
+ private static void configureTableIdentity(MTMVRelatedTableIf table) {
+ DatabaseIf database = Mockito.mock(DatabaseIf.class);
+ CatalogIf catalog = Mockito.mock(CatalogIf.class);
+ Mockito.when(table.getName()).thenReturn("table");
+ Mockito.when(table.getDatabase()).thenReturn(database);
+ Mockito.when(database.getFullName()).thenReturn("database");
+ Mockito.when(database.getCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("catalog");
+ }
+
+ private static Map<String, MTMVSnapshotIf> snapshots(Set<String> names) {
+ Map<String, MTMVSnapshotIf> snapshots = new LinkedHashMap<>();
+ for (String name : names) {
+ snapshots.put(name, new MTMVTimestampSnapshot(name.hashCode()));
+ }
+ return snapshots;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRewriteUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRewriteUtilTest.java
index 2387fd8b44a..e05727cccc7 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRewriteUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVRewriteUtilTest.java
@@ -29,6 +29,7 @@ import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import org.apache.doris.mtmv.MTMVRefreshContext.PreparedPartitionSnapshots;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
import org.apache.doris.qe.ConnectContext;
@@ -74,6 +75,10 @@ public class MTMVRewriteUtilTest {
private MTMVPartitionUtil mtmvPartitionUtil;
@Mocked
private MTMVUtil mtmvUtil;
+ @Mocked
+ private MTMVRefreshContext refreshContext;
+ @Mocked
+ private PreparedPartitionSnapshots preparedPartitionSnapshots;
private long currentTimeMills = 3L;
@Before
@@ -133,7 +138,8 @@ public class MTMVRewriteUtilTest {
minTimes = 0;
result = true;
- MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext)
any, anyString,
+ MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any,
+ (PreparedPartitionSnapshots) any, anyString,
(Set<BaseTableInfo>) any,
(Set<TableName>) any);
minTimes = 0;
@@ -143,6 +149,10 @@ public class MTMVRewriteUtilTest {
minTimes = 0;
result = false;
+
refreshContext.prepareComparablePartitionSnapshots((Set<String>) any);
+ minTimes = 0;
+ result = preparedPartitionSnapshots;
+
mtmv.canBeCandidate();
minTimes = 0;
result = true;
@@ -158,7 +168,8 @@ public class MTMVRewriteUtilTest {
minTimes = 0;
result = 2L;
- MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext)
any, anyString,
+ MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any,
+ (PreparedPartitionSnapshots) any, anyString,
(Set<BaseTableInfo>) any,
(Set<TableName>) any);
minTimes = 0;
@@ -189,7 +200,8 @@ public class MTMVRewriteUtilTest {
minTimes = 0;
result = 2L;
- MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext)
any, anyString,
+ MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any,
+ (PreparedPartitionSnapshots) any, anyString,
(Set<BaseTableInfo>) any,
(Set<TableName>) any);
minTimes = 0;
@@ -211,7 +223,8 @@ public class MTMVRewriteUtilTest {
minTimes = 0;
result = 1L;
- MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext)
any, anyString,
+ MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any,
+ (PreparedPartitionSnapshots) any, anyString,
(Set<BaseTableInfo>) any,
(Set<TableName>) any);
minTimes = 0;
@@ -246,7 +259,8 @@ public class MTMVRewriteUtilTest {
public void testGetMTMVCanRewritePartitionsNotSync() throws
AnalysisException {
new Expectations() {
{
- MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext)
any, anyString,
+ MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any,
+ (PreparedPartitionSnapshots) any, anyString,
(Set<BaseTableInfo>) any,
(Set<TableName>) any);
minTimes = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
