CalvinKirs commented on code in PR #67186: URL: https://github.com/apache/doris/pull/67186#discussion_r3893685313
########## fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsPartitionBatchExecutor.java: ########## @@ -0,0 +1,332 @@ +// 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; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; + +/** 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 long fallbackTimeoutNanos; + private final HmsPartitionTransport transport; + private final FailureClassifier failureClassifier; + private final LongSupplier nanoTime; + + private HmsPartitionBatchExecutor(Builder builder) { + this.maxBatchSize = builder.maxBatchSize; + this.minBatchSize = builder.minBatchSize; + this.fallbackTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(builder.fallbackTimeoutMillis); + this.transport = builder.transport; + this.failureClassifier = builder.failureClassifier; + this.nanoTime = builder.nanoTime; + } + + 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; + boolean fallbackStarted = false; + long fallbackStartNanos = 0; + while (offset < partitions.size()) { + checkFallbackTimeout(request, offset, fallbackStarted, fallbackStartNanos); + 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( Review Comment: Addressed in 6903af8364d by scope reduction. The fallback-specific timeout property and between-attempt clock were removed because they could not bound an active synchronous Thrift RPC without reintroducing the full cancellation/client-taint lifecycle. Adaptive reductions remain bounded by the max-to-min batch-size ladder, while individual calls use the existing HMS connection/socket timeout. The PR description now states this boundary explicitly. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java: ########## @@ -1601,9 +1630,7 @@ public List<Split> getSplits(int numBackends) throws UserException { // queryId) and write them into the query profile. connector-agnostic: the connector supplies the // group/label/metrics, the engine only transcribes them (no source branch). Default empty for // connectors that don't harvest. Same thread as planScan, so the harvest is complete. - List<ConnectorScanProfile> scanProfiles = onPluginClassLoader(scanProvider, - () -> scanProvider.collectScanProfiles(connectorSession)); - appendConnectorScanProfiles(scanProfiles); + collectAndAppendConnectorScanProfiles(scanProvider); Review Comment: Fixed in 6903af8364d. Synchronous scan planning now finalizes connector profiles on both success and failure. If profile finalization itself fails while planning is already failing, that failure is suppressed onto the original planning exception instead of masking it. PluginDrivenScanNodeScanProfileTest covers record-then-throw and primary-failure preservation. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java: ########## @@ -1951,6 +1981,14 @@ public void startSplit(int numBackends) { splitAssignment.setException(batchException.get()); } if (numFinishedPartitions.addAndGet(curBatchSize) == allPartitions.size()) { + try { + List<ConnectorScanProfile> profiles = onPluginClassLoader(scanProvider, Review Comment: Fixed in 6903af8364d. Batch finalization now waits for dispatch to close and only the actually submitted tasks to finish, then drains profiles and finishes scheduling exactly once. It no longer compares completed partitions with the full logical table list, so an early failure/stop before the last submission retains completed HMS diagnostics. The focused test covers dispatch closing with unsubmitted logical work. -- 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]
