github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3870178433
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java:
##########
@@ -782,15 +861,47 @@ private static HmsPartitionInfo
convertPartition(Partition partition) {
// ========== Pool management ==========
- private PooledHmsClient borrowClient() {
+ private PooledHmsClient borrowClient(ConnectorOperationControl
operationControl) {
if (clientPool == null) {
return createFreshClient();
}
- try {
- return clientPool.borrowObject();
- } catch (Exception e) {
- throw new HmsClientException(withRootCause("Failed to borrow HMS
client "
- + "from pool: " + e.getMessage(), e), e);
+ long poolWaitDeadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(POOL_BORROW_TIMEOUT_MS);
+ while (true) {
+ operationControl.checkActive();
+ long operationRemainingMillis =
operationControl.remainingTimeMillis();
+ if (operationRemainingMillis <= 0) {
+ throw new ConnectorOperationAbortedException(
+
ConnectorOperationAbortedException.Reason.DEADLINE_EXCEEDED,
+ "HMS client pool wait deadline exceeded");
+ }
+ long poolRemainingNanos = poolWaitDeadlineNanos -
System.nanoTime();
+ if (poolRemainingNanos <= 0) {
+ throw new HmsClientException("Timed out waiting for an HMS
client from the pool");
+ }
+ long poolRemainingMillis = Math.max(1L,
+ TimeUnit.NANOSECONDS.toMillis(poolRemainingNanos));
+ long waitMillis = Math.min(POOL_BORROW_CHECK_MILLIS,
poolRemainingMillis);
+ if (operationRemainingMillis != Long.MAX_VALUE) {
+ waitMillis = Math.min(waitMillis, operationRemainingMillis);
+ }
+ try {
+ return clientPool.borrowObject(waitMillis);
Review Comment:
**[P1] Bound HMS client creation with the operation control.** On an empty
pool, Commons Pool 2.2 runs `HmsClientFactory.create()` synchronously inside
`borrowObject(waitMillis)` before the timed idle-object wait, so `waitMillis`
does not bound `createFreshClient()`; the pool-disabled branch calls it
directly as well. Kerberos login, DNS, or socket construction can therefore
remain stuck after KILL/deadline, before `HmsRemoteCallTracking` installs its
watchdog and before the next `checkActive()`. Please make creation
cancellable/deadline-aware (and destroy any client that completes late) for
both branches, with blocking-provider KILL/deadline tests.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java:
##########
@@ -155,4 +155,14 @@ default long allocateTransactionId() {
default ConnectorStatementScope getStatementScope() {
return ConnectorStatementScope.NONE;
}
+
+ /** Returns cooperative cancellation and deadline control for connector
metadata operations. */
Review Comment:
**[P2] Freeze the new session/control API in the plugin surface.**
`ConnectorPluginSurfaceTest.FROZEN_TYPES` does not include `ConnectorSession`
or the new control/observer/event/abort types, so the regenerated baseline
records `ConnectorContext#getMetadataAccessObserver()` but not these two
session methods or the callable contracts they expose. The separate metadata
baseline also omits return types. That leaves later removal/re-signing of this
new 6.0 surface invisible to the stated compatibility speed bump. This is
independent of whether 6.0 is still unpublished: please freeze these reachable
SPI types (or recursively freeze reachable SPI contracts), regenerate the
baseline, and assert the new methods are present.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsPartitionRequest.java:
##########
@@ -0,0 +1,238 @@
+// 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 org.apache.doris.connector.spi.ConnectorMetadataAccessObserver;
+import org.apache.doris.connector.spi.ConnectorOperationControl;
+import org.apache.doris.connector.spi.ConnectorSession;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Immutable logical HMS request data with request-scoped adaptive execution
state. */
+final class HmsPartitionRequest {
+
+ static final long NO_FALLBACK_START_NANOS = Long.MIN_VALUE;
+
+ private final String dbName;
+ private final String tableName;
+ private final List<String> partitionNames;
+ private final HmsPartitionAccessSource source;
+ private final ConnectorOperationControl operationControl;
+ private final ConnectorMetadataAccessObserver metadataAccessObserver;
+ private final PartitionChunkConsumer partitionChunkConsumer;
+ private final BatchExecutionState batchExecutionState;
+
+ private HmsPartitionRequest(Builder builder) {
+ this.dbName = builder.dbName;
+ this.tableName = builder.tableName;
+ this.partitionNames = Collections.unmodifiableList(new
ArrayList<>(builder.partitionNames));
+ this.source = builder.source;
+ this.operationControl = builder.operationControl;
+ this.metadataAccessObserver = builder.metadataAccessObserver;
+ this.partitionChunkConsumer = builder.partitionChunkConsumer;
+ this.batchExecutionState = builder.batchExecutionState;
+ }
+
+ static HmsPartitionRequest from(ConnectorSession session,
HmsPartitionAccessSource source,
+ String dbName, String tableName, List<String> partitionNames) {
+ return builder()
+ .database(dbName)
+ .table(tableName)
+ .partitionNames(partitionNames)
+ .source(source)
+ .operationControl(session == null
+ ? ConnectorOperationControl.NONE :
session.getOperationControl())
+ .metadataAccessObserver(session == null
+ ? ConnectorMetadataAccessObserver.NOOP :
session.getMetadataAccessObserver())
+ .build();
+ }
+
+ static Builder builder() {
+ return new Builder();
+ }
+
+ String getDbName() {
+ return dbName;
+ }
+
+ String getTableName() {
+ return tableName;
+ }
+
+ List<String> getPartitionNames() {
+ return partitionNames;
+ }
+
+ HmsPartitionAccessSource getSource() {
+ return source;
+ }
+
+ ConnectorOperationControl getOperationControl() {
+ return operationControl;
+ }
+
+ ConnectorOperationControl getEffectiveOperationControl() {
+ ConnectorOperationControl effective =
batchExecutionState.effectiveOperationControl.get();
+ return effective == null ? operationControl : effective;
+ }
+
+ void updateEffectiveOperationControl(ConnectorOperationControl
effectiveOperationControl) {
+
batchExecutionState.effectiveOperationControl.set(effectiveOperationControl);
+ }
+
+ ConnectorMetadataAccessObserver getMetadataAccessObserver() {
+ return metadataAccessObserver;
+ }
+
+ PartitionChunkConsumer getPartitionChunkConsumer() {
+ return partitionChunkConsumer;
+ }
+
+ int effectiveBatchSize(int configuredMaxBatchSize) {
+ return Math.min(configuredMaxBatchSize,
batchExecutionState.effectiveBatchSize.get());
+ }
+
+ void reduceEffectiveBatchSize(int batchSize) {
+ batchExecutionState.effectiveBatchSize.accumulateAndGet(batchSize,
Math::min);
+ }
+
+ long fallbackStartNanos() {
+ return batchExecutionState.fallbackStartNanos.get();
+ }
+
+ long startFallback(long startNanos) {
+
batchExecutionState.fallbackStartNanos.compareAndSet(NO_FALLBACK_START_NANOS,
startNanos);
+ return batchExecutionState.fallbackStartNanos.get();
+ }
+
+ /** Receives one fully validated and request-ordered physical chunk before
the next HMS chunk starts. */
+ @FunctionalInterface
+ interface PartitionChunkConsumer {
+ PartitionChunkConsumer NOOP = (partitionNames, partitions,
operationControl) -> { };
+
+ void accept(List<String> partitionNames, List<HmsPartitionInfo>
partitions,
+ ConnectorOperationControl operationControl);
+ }
+
+ static final class Builder {
+ private String dbName;
+ private String tableName;
+ private List<String> partitionNames;
+ private HmsPartitionAccessSource source =
HmsPartitionAccessSource.UNKNOWN;
+ private ConnectorOperationControl operationControl =
ConnectorOperationControl.NONE;
+ private ConnectorMetadataAccessObserver metadataAccessObserver =
ConnectorMetadataAccessObserver.NOOP;
+ private PartitionChunkConsumer partitionChunkConsumer =
PartitionChunkConsumer.NOOP;
+ private BatchExecutionState batchExecutionState = new
BatchExecutionState();
+
+ private Builder() {
+ }
+
+ Builder database(String dbName) {
+ this.dbName = dbName;
+ return this;
+ }
+
+ Builder table(String tableName) {
+ this.tableName = tableName;
+ return this;
+ }
+
+ Builder partitionNames(List<String> partitionNames) {
+ this.partitionNames = partitionNames;
+ return this;
+ }
+
+ Builder source(HmsPartitionAccessSource source) {
+ this.source = source;
+ return this;
+ }
+
+ Builder operationControl(ConnectorOperationControl operationControl) {
+ this.operationControl = operationControl;
+ return this;
+ }
+
+ Builder metadataAccessObserver(ConnectorMetadataAccessObserver
metadataAccessObserver) {
+ this.metadataAccessObserver = metadataAccessObserver;
+ return this;
+ }
+
+ Builder partitionChunkConsumer(PartitionChunkConsumer
partitionChunkConsumer) {
+ this.partitionChunkConsumer = partitionChunkConsumer;
+ return this;
+ }
+
+ Builder shareBatchExecutionWith(HmsPartitionRequest request) {
+ this.batchExecutionState = request.batchExecutionState;
+ return this;
+ }
+
+ HmsPartitionRequest build() {
+ requireName(dbName, "database");
+ requireName(tableName, "table");
+ Objects.requireNonNull(partitionNames, "partitionNames");
+ Objects.requireNonNull(source, "source");
+ Objects.requireNonNull(operationControl, "operationControl");
+ Objects.requireNonNull(metadataAccessObserver,
"metadataAccessObserver");
+ Objects.requireNonNull(partitionChunkConsumer,
"partitionChunkConsumer");
+ Objects.requireNonNull(batchExecutionState, "batchExecutionState");
+ Set<List<String>> identities = new HashSet<>();
+ List<String> partitionKeys = null;
+ for (int i = 0; i < partitionNames.size(); i++) {
+ if ((i & 1023) == 0) {
+ operationControl.checkActive();
+ }
+ String partitionName = partitionNames.get(i);
+ HmsPartitionIdentity.ParsedPartitionName parsed =
HmsPartitionIdentity.parse(partitionName);
Review Comment:
**[P2] Retain parsed identities across the cache-backed request.** This
builder validates every partition name with `HmsPartitionIdentity.parse()` and
then discards the result. The normal cold-cache path reparses all names during
cache lookup, registration, copied-window construction, raw validation, and
final reconstruction—`6N+C` parse/unescape passes for a fully cold sole owner
(roughly 720k canonicalizations for 120k names); even all hits take `3N`. The
added 120k test covers only the raw loader, so it misses this decorator cost.
Please carry immutable parsed keys/values on the request and share/slice them
through windows and validation, with a cache-backed large-request
parse-count/performance test.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java:
##########
@@ -245,6 +245,7 @@ public static boolean isMTMVSync(MTMVRefreshContext
context, Set<BaseTableInfo>
throws AnalysisException {
MTMV mtmv = context.getMtmv();
Set<String> partitionNames = mtmv.getPartitionNames();
+ context.preloadSnapshots(partitionNames, tables, excludeTables);
Review Comment:
**[P2] Check the persisted partition set before eager freshness loading.**
This preload fetches partition-detail freshness for the entire mapped PCT union
before `isSyncWithPartitions()` compares the current names with the persisted
`getPctSnapshots()` set. With a coarse mapping over 160k Hive partitions,
adding or dropping one partition already proves the MTMV stale locally, but
this order now issues the full HMS freshness request first (and can fail before
returning that deterministic stale result); the pre-PR path performed the set
gate before fetching snapshots. Please make the comparison two-phase—reject
name-set mismatches first, then preload only mappings that still need version
checks—and add large mismatch coverage asserting no partition-freshness call.
--
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]
