github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3893397919
##########
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:
**[P2] Drain profiles when synchronous planning fails.** Hive can record
pruning or partition-RPC stats inside `planScan` and then fail during
filesystem, file-list, or ACID planning. Because this collection runs only
after `planScan` and the later post-processing succeed, the failed query loses
those diagnostics. Please collect from a failure-safe finalization path that
preserves the primary planning exception, and add a record-then-throw
synchronous planning test.
##########
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:
**[P2] Enforce the fallback budget during each retry RPC.** After the first
size-limit failure starts the fallback clock, this synchronous call can still
block until the independent HMS socket timeout, even when
`hive.hms_partitions_batch_fallback_timeout_ms` is much shorter. The checks
before/after the call only reject the result after the advertised budget has
already elapsed. Please apply the remaining budget to each retry (or provide
equivalent in-flight cancellation) and add a blocking-retry test that proves
the call returns within it.
##########
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:
**[P2] Drain batch profiles when dispatch stops early.** This collection
runs only when finished partitions equal the full partition list, but the
dispatcher stops submitting after a worker failure or query stop. If any
batches remain unsubmitted, no worker can reach that total, so successful
earlier HMS calls remain buffered and the failed query profile loses their
diagnostics. Please finalize against the submitted/outstanding work after
dispatch closes, collect once on success or failure, and add a
failure-before-last-submission 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]
