github-actions[bot] commented on code in PR #67186: URL: https://github.com/apache/doris/pull/67186#discussion_r3901694295
########## fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsPartitionBatchExecutor.java: ########## @@ -0,0 +1,311 @@ +// 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.connector.hms; + +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 { + + @FunctionalInterface + interface FailureClassifier { + boolean isDegradable(Throwable failure); + } + + static final class RemoteCallException extends HmsClientException { + RemoteCallException(String message, Throwable cause) { + super(message, cause); + } + } + + private final int maxBatchSize; + private final int minBatchSize; + private final HmsPartitionTransport transport; + private final FailureClassifier failureClassifier; + + private HmsPartitionBatchExecutor(Builder builder) { + this.maxBatchSize = builder.maxBatchSize; + this.minBatchSize = builder.minBatchSize; + this.transport = builder.transport; + this.failureClassifier = builder.failureClassifier; + } + + static Builder builder() { + return new Builder(); + } + + List<HmsPartitionInfo> execute(HmsPartitionRequest request) { + return executeWithStats(request).getPartitions(); + } + + List<HmsPartitionInfo> executeExisting(HmsPartitionRequest request) { + return executeExistingWithStats(request).getPartitions(); + } + + 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<HmsPartitionInfo> result = new ArrayList<>(partitions.size()); + int offset = 0; + int effectiveBatchSize = maxBatchSize; + int attempts = 0; + int fallbackCount = 0; + long rpcItems = 0; + long rpcElapsedNanos = 0; + long maxRpcElapsedNanos = 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()); + } + attempts++; Review Comment: **[P2] Do not report transport invocations as physical HMS RPCs.** These counters and the RPC timer start before `transport.getPartitionsByNames`; the pooled path can then fail in `borrowClient`, fresh-client creation, or outer authentication without calling HMS at all, while the default `RetryingMetaStoreClient` can perform multiple wire attempts inside one invocation. The Query Profile can therefore show one `RpcAttempt`/all `RpcItems` for zero wire calls, or undercount retries, and `RpcElapsedTime` includes setup/pool wait despite the new API documenting physical-attempt statistics. Instrument actual client attempts (including retries), or rename/separate these as batch-invocation and setup metrics, with pre-wire-failure and retry coverage. ########## fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java: ########## @@ -96,8 +98,25 @@ public static Collection<Partition> getMTMVCanRewritePartitions(MTMV mtmv, Conne 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()); + } + } Review Comment: **[P2] Preserve per-partition rewrite failure isolation.** This rewrite-wide preload sits outside the loop's existing checked-failure boundary. With `mv1 -> p1` and `mv2 -> p2`, if `p1` disappears after the pin, Hive's omission-tolerant bulk result can still contain `p2`, but `getPartitionSnapshots` throws for `p1` and this catch returns before `mv2` is evaluated. The previous scalar lookup threw inside the per-partition `try`, so only `mv1` was skipped. Please retain per-name bulk outcomes (for example, defer missing-name errors to `PreparedPartitionSnapshots#get`) and add a two-partition pin/drop test proving the unaffected partition remains rewritable. ########## fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java: ########## @@ -1182,17 +1183,19 @@ public Optional<FilterApplicationResult<ConnectorTableHandle>> applyFilter( return Optional.empty(); } - List<HmsPartitionInfo> prunedPartitions = matchedPartNames.isEmpty() - ? Collections.emptyList() - : hmsClient.getPartitions(hiveHandle.getDbName(), Review Comment: **[P2] Preserve failed pruning stats before the scan provider exists.** A selective equality-pruning request can perform HMS batches and then throw a stats-bearing `HmsClientException` here, before a new handle or `HiveScanPlanProvider` is created. `convertPredicate` propagates that failure, so the synchronous/batch finalizers never run and the Query Profile omits the request that aborted planning. The existing fixes cover successful pruning handoff and failures inside `planScan`, not this earlier boundary. Establish the scan-scoped profile owner before filter pushdown (or otherwise publish the attached stats while preserving the primary exception), and add a production-chain failing-prune test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
