github-actions[bot] commented on code in PR #67186: URL: https://github.com/apache/doris/pull/67186#discussion_r3894173628
########## fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsPartitionBatchExecutor.java: ########## @@ -0,0 +1,290 @@ +// 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++; + rpcItems += batchSize; + largestBatchSize = Math.max(largestBatchSize, batchSize); + smallestBatchSize = Math.min(smallestBatchSize, batchSize); + long rpcStartNanos = System.nanoTime(); + try { + List<HmsPartitionInfo> returned = transport.getPartitionsByNames( + request.getDbName(), request.getTableName(), batchNames); + result.addAll(validateAndOrder(batch, returned, allowMissing)); + offset += batchSize; + } catch (RemoteCallException e) { + if (batchSize <= minBatchSize || !failureClassifier.isDegradable(e)) { + throw finalBatchFailure(request, offset, batchSize, effectiveBatchSize, Review Comment: **[P2] Preserve the failed logical request's batch stats.** The terminal branches here throw before the only `HmsPartitionBatchStats` is built, and `HiveScanPlanProvider` records stats only after `getExistingPartitionsWithStats` returns. A first failed scan therefore drains an empty profile, while a later failure reports only prior successful requests and omits the RPC, latency, and fallback that caused the error. This is distinct from the new planning/dispatch finalizers, which can only drain already-buffered stats. Please carry the terminal stats on a typed failure (or equivalent callback), record them before rethrowing the primary error, and add first-failure and fallback-exhaustion profile tests for synchronous and batch planning. -- 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]
