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 fcaefecf41b156..a3a04600c9fdee 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 getPartitions(String dbName, String tblName, List 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 getExistingPartitions(String dbName, String tblName, List partitionNames) { + return getPartitions(dbName, tblName, partitionNames); + } + Table getTable(String dbName, String tblName); List 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 6e714e20a9b780..242dfc12a8c41a 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 @@ 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 HMSClientException(String format, Throwable cause, Object... msg) { 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 a727d4386f774a..2a7759faecb735 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 MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshCont return dlaTable.getPartitionSnapshot(partitionName, context, snapshot); } + @Override + public Map getPartitionSnapshots(Set partitionNames, + MTMVRefreshContext context, Optional 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 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 bfd39c59684e65..e8c12dd5623c46 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 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 MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshCont 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 getPartitionSnapshots(Set partitionNames, + Optional snapshot) throws AnalysisException { + try { + HiveExternalMetaCache.HivePartitionValues hivePartitionValues = + hmsTable.getHivePartitionValues(snapshot); + HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr() + .hive(hmsTable.getCatalog().getId()); + List resolvedNames = new ArrayList<>(partitionNames.size()); + List> resolvedValues = new ArrayList<>(partitionNames.size()); + for (String partitionName : partitionNames) { + Long partitionId = hivePartitionValues.getPartitionNameToIdMap().get(partitionName); + if (partitionId == null) { + continue; + } + List partitionValues = hivePartitionValues.getPartitionValuesMap().get(partitionId); + if (CollectionUtils.isEmpty(partitionValues)) { + continue; + } + resolvedNames.add(partitionName); + resolvedValues.add(partitionValues); + } + Map result = new LinkedHashMap<>(); + if (resolvedNames.isEmpty()) { + return result; + } + List 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 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 981b1f640138e3..cdf58a9adc6ac4 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 @@ private Map loadPartitions(Iterable 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 partitions = catalog.getClient().getExistingPartitions( nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName(), partitionNames); for (Partition partition : partitions) { StorageDescriptor sd = partition.getSd(); @@ -657,7 +660,30 @@ private List getAllPartitions(ExternalTable dorisTable, List partitions; if (withCache) { MetaCacheEntry 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 resolved = new HashMap<>(); + List 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 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 00000000000000..245f52b8bd6d8d --- /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 partitions = request.getPartitions(); + if (partitions.isEmpty()) { + HmsPartitionBatchStats stats = HmsPartitionBatchStats.builder() + .logicalElapsedNanos(System.nanoTime() - logicalStartNanos) + .build(); + return new HmsPartitionBatchResult(new ArrayList<>(), stats); + } + + List 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 batch = + partitions.subList(offset, offset + batchSize); + List 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 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 validateAndOrder( + List requested, + List returned, boolean allowMissing) { + int expectedValueCount = requested.get(0).getValues().size(); + Map, 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 ordered = new ArrayList<>(java.util.Collections.nCopies(requested.size(), null)); + Map, Integer> returnedCounts = new LinkedHashMap<>(); + if (returned == null) { + failure.invalid(""); + } else { + for (Partition partition : returned) { + if (partition == null) { + failure.invalid(""); + continue; + } + List 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, 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 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/HmsPartitionBatchResult.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchResult.java new file mode 100644 index 00000000000000..70c68933577070 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionBatchResult.java @@ -0,0 +1,42 @@ +// 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 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 partitions; + private final HmsPartitionBatchStats stats; + + public HmsPartitionBatchResult(List partitions, HmsPartitionBatchStats stats) { + this.partitions = Objects.requireNonNull(partitions, "partitions"); + this.stats = Objects.requireNonNull(stats, "stats"); + } + + public List getPartitions() { + return partitions; + } + + 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 00000000000000..1e2c476875a277 --- /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 00000000000000..62d292c1cda4dc --- /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 keysFromName(String partitionName) { + return parseParts(partitionName, null, true); + } + + static ParsedPartitionName parse(String partitionName, List expectedKeys) { + return new ParsedPartitionName(partitionName, + Collections.unmodifiableList(parseParts(partitionName, expectedKeys, false))); + } + + private static List parseParts( + String partitionName, List expectedKeys, boolean returnKeys) { + if (partitionName == null || partitionName.isEmpty()) { + throw new IllegalArgumentException("partition name must not be empty"); + } + List 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 values; + + private ParsedPartitionName(String name, List values) { + this.name = name; + this.values = values; + } + + String getName() { + return name; + } + + List 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 00000000000000..13efab2728e6e1 --- /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 partitions; + + HmsPartitionRequest(String dbName, String tableName, List 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 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 parsePartitions(List names) { + List parsedPartitions = new ArrayList<>(names.size()); + Set> identities = new HashSet<>(); + List 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 00000000000000..89cabf097cefd9 --- /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 missingSamples = new ArrayList<>(); + private final List duplicateSamples = new ArrayList<>(); + private final List unexpectedSamples = new ArrayList<>(); + private final List 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 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/HmsPartitionTransport.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionTransport.java new file mode 100644 index 00000000000000..afa951f7d9f8cb --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HmsPartitionTransport.java @@ -0,0 +1,29 @@ +// 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 java.util.List; + +/** Leaf transport contract: one invocation enters the configured getPartitionsByNames transport once. */ +@FunctionalInterface +interface HmsPartitionTransport { + List getPartitionsByNames( + String dbName, String tableName, List 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 46ca8ef3bc87ae..0ddb0e0f48ef3c 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 ThriftHMSCachedClient(HiveConf hiveConf, int poolSize, ExecutionAuthentic 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 Partition getPartition(String dbName, String tblName, List partit @Override public List getPartitions(String dbName, String tblName, List 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 getExistingPartitions(String dbName, String tblName, List 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 fetchPartitionsByNames(String dbName, String tblName, List 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 ef84d6071ca20b..d6f025308b7e4c 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 @@ -256,6 +256,17 @@ public void run() throws JobException { } Map 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 1b2dc56a9c8e58..98c279a6ac4cbf 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.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 tables, Set excludedTriggerTables) throws AnalysisException { MTMV mtmv = refreshContext.getMtmv(); @@ -109,7 +111,8 @@ public static boolean isMTMVPartitionSync(MTMVRefreshContext refreshContext, Str Set 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 static boolean isMTMVSync(MTMVRefreshContext context, Set throws AnalysisException { MTMV mtmv = context.getMtmv(); Set 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 static Map> getPartitionsUnSyncTables(MTMV mtmv) List partitionIds = mtmv.getPartitionIds(); Map> 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 getPartitionUnSyncTables(MTMVRefreshContext context, String partitionName) + private static List getPartitionUnSyncTables(MTMVRefreshContext context, + PreparedPartitionSnapshots partitionSnapshots, String partitionName) throws AnalysisException { MTMV mtmv = context.getMtmv(); Map> mappings = context.getByPartitionName(partitionName); @@ -273,8 +281,8 @@ private static List getPartitionUnSyncTables(MTMVRefreshContext context, if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE && pctTables.contains( pctTable)) { Set 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 static List getMTMVNeedRefreshPartitions(MTMVRefreshContext conte MTMV mtmv = context.getMtmv(); Set partitionNames = mtmv.getPartitionNames(); List 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 static List getMTMVNeedRefreshPartitions(MTMVRefreshContext conte * @return * @throws AnalysisException */ - public static boolean isSyncWithPartitions(MTMVRefreshContext context, String mtmvPartitionName, + public static boolean isSyncWithPartitions(MTMVRefreshContext context, + PreparedPartitionSnapshots partitionSnapshots, String mtmvPartitionName, Set pctPartitionNames, MTMVRelatedTableIf pctTable) throws AnalysisException { MTMV mtmv = context.getMtmv(); if (!pctTable.needAutoRefresh()) { @@ -339,8 +355,7 @@ public static boolean isSyncWithPartitions(MTMVRefreshContext context, String mt 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 static MTMVSnapshotIf getTableSnapshotFromContext(MTMVRelatedTableIf mtmv 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 static MTMVSnapshotIf getTableSnapshotFromContext(MTMVRelatedTableIf mtmv public static Map generatePartitionSnapshots(MTMVRefreshContext context, Set baseTables, Set partitionNames) throws AnalysisException { + PreparedPartitionSnapshots preparedSnapshots = context.preparePartitionSnapshots(partitionNames); Map 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 baseTables, Map> pctPartitionNames) + PreparedPartitionSnapshots partitionSnapshots, Set baseTables, + Map> pctPartitionNames) throws AnalysisException { MTMV mtmv = context.getMtmv(); MTMVRefreshPartitionSnapshot refreshPartitionSnapshot = new MTMVRefreshPartitionSnapshot(); @@ -594,8 +610,7 @@ private static MTMVRefreshPartitionSnapshot generatePartitionSnapshot(MTMVRefres 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 603f89bbec5d85..77fe3312119b9f 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 @@ 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 baseTableSnapshotCache = Maps.newHashMap(); + private final Map> partitionSnapshotCache = Maps.newHashMap(); + private final Map> missingPartitionSnapshotCache = Maps.newHashMap(); + private final Map> partitionSnapshotFailureCache = + Maps.newHashMap(); public MTMVRefreshContext(MTMV mtmv) { this.mtmv = mtmv; @@ -58,6 +69,78 @@ public Map getBaseTableSnapshotCache() { return baseTableSnapshotCache; } + /** Loads the union of mapped base partitions once per related table. */ + public PreparedPartitionSnapshots preparePartitionSnapshots(Set 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 mtmvPartitionNames) + throws AnalysisException { + return preparePartitionSnapshots(mtmvPartitionNames, true); + } + + private PreparedPartitionSnapshots preparePartitionSnapshots( + Set mtmvPartitionNames, boolean comparableOnly) + throws AnalysisException { + Map> namesByTable = new LinkedHashMap<>(); + Map tableInfos = comparableOnly + ? new LinkedHashMap<>() : Collections.emptyMap(); + for (String mtmvPartitionName : mtmvPartitionNames) { + for (Map.Entry> 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> entry : namesByTable.entrySet()) { + loadSnapshots(entry.getKey(), entry.getValue()); + } + return new PreparedPartitionSnapshots(this); + } + + private void loadSnapshots(MTMVRelatedTableIf table, Set partitionNames) throws AnalysisException { + Map cached = partitionSnapshotCache.computeIfAbsent( + table, ignored -> new LinkedHashMap<>()); + Set knownMissing = missingPartitionSnapshotCache.computeIfAbsent( + table, ignored -> new LinkedHashSet<>()); + Set missing = new LinkedHashSet<>(partitionNames); + missing.removeAll(cached.keySet()); + missing.removeAll(knownMissing); + if (missing.isEmpty()) { + return; + } + Map 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 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 static MTMVRefreshContext buildContext(MTMV mtmv) throws AnalysisExceptio return context; } + Optional 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 snapshots = context.partitionSnapshotCache.get(table); + if (snapshots != null && snapshots.containsKey(partitionName)) { + return snapshots.get(partitionName); + } + Map failures = context.partitionSnapshotFailureCache.get(table); + if (failures != null && failures.containsKey(partitionName)) { + throw failures.get(partitionName); + } + Set 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 7f38a0492c281c..ee0bbb8b6da159 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.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 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 getPartitionSnapshots(Set partitionNames, + MTMVRefreshContext context, Optional snapshot) throws AnalysisException { + Map 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 6dc0bd24da72dc..7a55fc12784a6f 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.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 static Collection getMTMVCanRewritePartitions(MTMV mtmv, Conne } Set mtmvNeedComparePartitions = null; MTMVRefreshContext refreshContext = null; + PreparedPartitionSnapshots partitionSnapshots = null; // check gracePeriod long gracePeriodMills = mtmv.getGracePeriod(); for (Partition partition : allPartitions) { @@ -95,8 +97,25 @@ public static Collection getMTMVCanRewritePartitions(MTMV mtmv, Conne if (!mtmvNeedComparePartitions.contains(partition.getName())) { continue; } + if (partitionSnapshots == null) { + Set 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 00000000000000..3421bd3609241b --- /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 batchSizes = new ArrayList<>(); + HmsPartitionBatchExecutor executor = executor(5000, (db, table, names) -> { + batchSizes.add(names.size()); + List result = infos(names); + Collections.reverse(result); + return result; + }); + + List names = names(120_000); + List 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 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 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 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 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 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 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 names) { + return new HmsPartitionRequest("db", "table", names); + } + + private static List 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 names(int count) { + return IntStream.range(0, count).mapToObj(i -> "p=" + i).collect(Collectors.toList()); + } + + private static List infos(List 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 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 24f0fcf88a8bc4..7a1b66d36f5814 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 baseTables = Sets.newHashSet(); @@ -150,6 +152,18 @@ public void setUp() throws NoSuchMethodException, SecurityException, AnalysisExc minTimes = 0; result = baseSnapshotIf; + context.prepareComparablePartitionSnapshots((Set) any); + minTimes = 0; + result = partitionSnapshots; + + context.preparePartitionSnapshots((Set) 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 void testIsMTMVSyncNotSync() { @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 void testIsSyncWithPartitionNotEqual() throws AnalysisException { } }; 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 void testIsSyncWithPartitionNotSync() throws AnalysisException { } }; 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 void testIsMTMVPartitionSyncWithImmutableExcludedTriggerTables() throws A }; Set 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 00000000000000..aa93128e625c1e --- /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>> 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.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>> 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>> 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>> 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>> 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>> 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.getArgument(0).hashCode())); + + Map 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 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>> 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 snapshots(Set names) { + Map 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 2387fd8b44abd5..e05727cccc70aa 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.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 void setUp() throws NoSuchMethodException, SecurityException, AnalysisExc minTimes = 0; result = true; - MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, anyString, + MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, + (PreparedPartitionSnapshots) any, anyString, (Set) any, (Set) any); minTimes = 0; @@ -143,6 +149,10 @@ public void setUp() throws NoSuchMethodException, SecurityException, AnalysisExc minTimes = 0; result = false; + refreshContext.prepareComparablePartitionSnapshots((Set) any); + minTimes = 0; + result = preparedPartitionSnapshots; + mtmv.canBeCandidate(); minTimes = 0; result = true; @@ -158,7 +168,8 @@ public void testGetMTMVCanRewritePartitionsForceConsistent() throws AnalysisExce minTimes = 0; result = 2L; - MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, anyString, + MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, + (PreparedPartitionSnapshots) any, anyString, (Set) any, (Set) any); minTimes = 0; @@ -189,7 +200,8 @@ public void testGetMTMVCanRewritePartitionsInGracePeriod() throws AnalysisExcept minTimes = 0; result = 2L; - MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, anyString, + MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, + (PreparedPartitionSnapshots) any, anyString, (Set) any, (Set) any); minTimes = 0; @@ -211,7 +223,8 @@ public void testGetMTMVCanRewritePartitionsNotInGracePeriod() throws AnalysisExc minTimes = 0; result = 1L; - MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, anyString, + MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, + (PreparedPartitionSnapshots) any, anyString, (Set) any, (Set) any); minTimes = 0; @@ -246,7 +259,8 @@ public void testGetMTMVCanRewritePartitionsDisableMaterializedViewRewrite() { public void testGetMTMVCanRewritePartitionsNotSync() throws AnalysisException { new Expectations() { { - MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, anyString, + MTMVPartitionUtil.isMTMVPartitionSync((MTMVRefreshContext) any, + (PreparedPartitionSnapshots) any, anyString, (Set) any, (Set) any); minTimes = 0;