github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3879148549
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PreloadExternalMetadata.java:
##########
@@ -79,6 +84,22 @@ public ExternalMetadataPreloadResult
executePreload(StatementContext statementCo
TimeUtils.getElapsedTimeMs(preloadStartTime));
}
+ private void preloadCloudMtmvRefreshContexts(StatementContext
statementContext) {
+ if (Config.isNotCloudMode()) {
+ return;
+ }
+ for (MTMV mtmv : statementContext.getCandidateMTMVs()) {
+ if
(statementContext.getPreloadedMtmvRefreshContext(mtmv).isPresent()) {
+ continue;
+ }
+ try {
+ statementContext.putPreloadedMtmvRefreshContext(mtmv,
MTMVRefreshContext.buildContext(mtmv));
+ } catch (AnalysisException e) {
Review Comment:
[P1] Preserve the optional-MV failure boundary here. `buildContext` can
materialize an external PCT snapshot and throw an unchecked
`DorisConnectorException`, but this new unconditional cloud stage catches only
`AnalysisException`. A query over a healthy base table can now fail because an
otherwise optional candidate MTMV references an unavailable connector; the
later rewrite hook historically degrades such a candidate to no rewrite. Please
rethrow `ConnectorOperationAbortedException`, but isolate/log ordinary
connector failures per candidate, with a cloud collect-stage failure test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/PreloadExternalMetadata.java:
##########
@@ -58,6 +62,7 @@ public List<Rule> buildRules() {
*/
public ExternalMetadataPreloadResult executePreload(StatementContext
statementContext) {
long preloadStartTime = TimeUtils.getStartTimeMs();
+ preloadCloudMtmvRefreshContexts(statementContext);
Review Comment:
[P1] Build this cloud context from the statement's exact external snapshot.
This call runs before normal snapshot loading, so
`getAndCopyPartitionItems(empty)` materializes an unrecorded latest pin; later
freshness uses the statement pin (or materializes latest again), while locked
revalidation keeps the old external mapping. A partition added between those
reads can be absent from the mapping yet compared against a newer generation,
allowing an incomplete MTMV rewrite. Please install/reuse the candidate table
pin before building the context and add an interleaving test proving mapping,
freshness, and scan share it.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -1086,33 +1090,61 @@ static long scaleSampledSize(long sampledSize, int
totalPartitions, int sampledP
}
/**
- * Resolves the data locations to list: the table location for an
unpartitioned table, else every
- * partition's location (bounded by {@link #MAX_PARTITIONS_FOR_STATS}). A
partition or table with no
- * location contributes nothing.
+ * Resolves the data locations to list. For an estimate, sampling happens
on lightweight partition names
+ * before any partition object is requested. A non-positive sample size
means that the explicit file-size
+ * path needs every partition object. A partition or table with no
location contributes nothing.
*/
- private List<PartitionRef> resolvePartitionRefs(HiveTableHandle handle) {
+ private PartitionRefSelection resolvePartitionRefs(
+ ConnectorSession session, HiveTableHandle handle, int sampleSize) {
List<String> partKeyNames = handle.getPartitionKeyNames();
if (partKeyNames == null || partKeyNames.isEmpty()) {
String location = handle.getLocation();
- return (location == null || location.isEmpty())
+ List<PartitionRef> refs = (location == null || location.isEmpty())
? Collections.emptyList()
: Collections.singletonList(new PartitionRef(location,
Collections.emptyList()));
+ return new PartitionRefSelection(refs, refs.size(), refs.size(),
false);
}
List<String> partNames = hmsClient.listPartitionNames(
- handle.getDbName(), handle.getTableName(),
MAX_PARTITIONS_FOR_STATS);
+ handle.getDbName(), handle.getTableName(), ALL_PARTITIONS);
Review Comment:
[P1] Keep the full name listing under the new operation control. This path
first calls the sessionless `listPartitionNames(..., -1)`, whose Thrift
implementation uses `ConnectorOperationControl.NONE`; the session is only
applied to the later sampled `getPartitions` call. On a cold very-large table,
KILL or the statement deadline therefore cannot stop the largest HMS RPC in
this statistics path (and an already-cancelled cached request still
copies/shuffles the full list). Please add a session-aware name-list path and
use it here and in the parallel scan/write/freshness callers, with blocking
cold-list cancellation/deadline coverage.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java:
##########
@@ -674,17 +747,43 @@ public synchronized void close() throws IOException {
// ========== Internal execution framework ==========
private <T> T execute(HmsAction<T> action) {
+ return execute(ConnectorOperationControl.NONE, action);
+ }
+
+ private <T> T execute(ConnectorOperationControl operationControl,
HmsAction<T> action) {
if (closed) {
throw new HmsClientException("HMS client is closed");
}
- try (PooledHmsClient pooled = borrowClient()) {
+ operationControl.checkActive();
+ try (PooledHmsClient pooled = borrowClient(operationControl)) {
+ operationControl.checkActive();
+ T result;
try {
- return doAs(() -> action.call(pooled.client));
+ result = doAs(() -> {
Review Comment:
[P1] Install operation control before entering the outer Kerberos `doAs`.
The action that starts `HmsRemoteCallTracking.withTracker` runs only after
`AuthAction.execute`; Hive/Hudi authentication calls `getUGI` first, whose
synchronized keytab first-use/refresh can block on another login or the KDC.
KILL/deadline therefore has no watchdog here even though no wire call has
started. Put this authenticated setup under the bounded control lifecycle
(retaining the borrowed client until any abandoned auth task exits), recheck
inside the callable, and add blocking Hive/Hudi keytab first-use/refresh tests.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsRemoteCallTracking.java:
##########
@@ -0,0 +1,273 @@
+// 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.ConnectorOperationAbortedException;
+import org.apache.doris.connector.spi.ConnectorOperationControl;
+
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/** Bridges one logical client invocation to every raw HMS attempt made by
RetryingMetaStoreClient. */
+final class HmsRemoteCallTracking {
+
+ private static final long CONTROL_CHECK_MILLIS = 100L;
+ private static final ThreadLocal<Context> CURRENT = new ThreadLocal<>();
+ private static final ScheduledExecutorService CONTROL_WATCHDOG =
+ Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable,
"hms-operation-control-watchdog");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ private HmsRemoteCallTracking() {
+ }
+
+ static <T> T withTracker(HmsPartitionBatchLoader.RemoteCallTracker
tracker, int itemCount,
+ ConnectorOperationControl operationControl, Callable<T>
clientInvocation) throws Exception {
+ Context previous = CURRENT.get();
+ operationControl.checkActive();
+ Context context = new Context(tracker, itemCount, operationControl,
Thread.currentThread());
+ CURRENT.set(context);
+ ScheduledFuture<?> watchdog = operationControl ==
ConnectorOperationControl.NONE
+ ? null : CONTROL_WATCHDOG.scheduleWithFixedDelay(
+ context::checkOperation, CONTROL_CHECK_MILLIS,
CONTROL_CHECK_MILLIS, TimeUnit.MILLISECONDS);
+ try {
+ try {
+ T result = clientInvocation.call();
+ operationControl.checkActive();
+ return result;
+ } catch (Exception e) {
+ ConnectorOperationAbortedException abort = context.getAbort();
+ if (abort != null) {
+ throw abort;
+ }
+ if (causedByInterruptedException(e)) {
+ // RetryingMetaStoreClient's retry delay uses
Thread.sleep. A direct Future.cancel(true)
+ // can interrupt that sleep before the watchdog observes
the caller control. The Hive dynamic
+ // proxy wraps the undeclared InterruptedException in
UndeclaredThrowableException, and sleep
+ // clears the flag while throwing. Restore it and preserve
cancellation semantics; after a
+ // failed wire attempt the pooled client is ambiguous and
the specialized abort taints it.
+ Thread.currentThread().interrupt();
+ throw context.interruptedAbort();
+ }
+ throw e;
+ }
+ } finally {
+ context.finish();
+ if (watchdog != null) {
+ watchdog.cancel(false);
+ }
+ if (context.wasInterruptedByWatchdog()) {
+ Thread.interrupted();
+ }
+ if (previous == null) {
+ CURRENT.remove();
+ } else {
+ CURRENT.set(previous);
+ }
+ }
+ }
+
+ private static boolean causedByInterruptedException(Throwable failure) {
+ for (Throwable cause = failure; cause != null; cause =
cause.getCause()) {
+ if (cause instanceof InterruptedException) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ static <T> T trackWireAttempt(ThriftCall<T> wireAttempt) throws TException
{
+ Context context = CURRENT.get();
+ if (context == null) {
+ return wireAttempt.call();
+ }
+ context.operationControl.checkActive();
+ context.startWireAttempt();
+ try {
+ T result = context.tracker.call(context.itemCount,
wireAttempt::call);
+ context.finishWireAttempt(false);
+ context.operationControl.checkActive();
+ return result;
+ } catch (TException e) {
+ context.finishWireAttempt(true);
+ throw e;
+ } catch (RuntimeException e) {
+ context.finishWireAttempt(false);
+ throw e;
+ } catch (Exception e) {
+ context.finishWireAttempt(false);
+ throw new TException(e);
+ }
+ }
+
+ static void checkReconnectActive() {
+ Context context = CURRENT.get();
+ if (context != null) {
+ context.checkRetryPhaseActive();
+ }
+ }
+
+ static void markReconnectFailure() {
+ Context context = CURRENT.get();
+ if (context != null) {
+ context.markReconnectFailure();
+ }
+ }
+
+ static void markReconnectSuccess() {
+ Context context = CURRENT.get();
+ if (context != null) {
+ context.markReconnectSuccess();
+ }
+ }
+
+ static boolean shouldTaintClient(ConnectorOperationAbortedException abort)
{
+ return abort instanceof RetryPhaseOperationAbortedException;
+ }
+
+ @FunctionalInterface
+ interface ThriftCall<T> {
+ T call() throws TException;
+ }
+
+ private static final class Context {
+ private final HmsPartitionBatchLoader.RemoteCallTracker tracker;
+ private final int itemCount;
+ private final ConnectorOperationControl operationControl;
+ private final Thread invocationThread;
+ private ConnectorOperationAbortedException abort;
+ private boolean finished;
+ private boolean interruptedByWatchdog;
+ private boolean wireCallActive;
+ private boolean retryingAfterWireFailure;
+ private boolean clientUnsafeAfterReconnectFailure;
+
+ private Context(HmsPartitionBatchLoader.RemoteCallTracker tracker, int
itemCount,
+ ConnectorOperationControl operationControl, Thread
invocationThread) {
+ this.tracker = tracker;
+ this.itemCount = itemCount;
+ this.operationControl = operationControl;
+ this.invocationThread = invocationThread;
+ }
+
+ private void checkOperation() {
+ try {
+ operationControl.checkActive();
+ } catch (ConnectorOperationAbortedException e) {
+ synchronized (this) {
+ if (finished || abort != null) {
+ return;
+ }
+ if ((retryingAfterWireFailure ||
clientUnsafeAfterReconnectFailure) && !wireCallActive) {
Review Comment:
[P1] Treat initial proxy authentication as a pre-wire setup phase. Hive
3.1.3 calls `reloginExpiringKeytabUser()` before reconnect and before
`method.invoke` reaches the tracked raw client. While a due keytab renewal
blocks on UGI/KDC, all three phase flags are false, so this branch only records
a generic abort and KILL/deadline cannot return until login finishes. Model
that phase explicitly (or use bounded asynchronous abandonment while retaining
the borrowed client until exit) and add blocking first-attempt keytab
KILL/deadline tests.
##########
fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionBuilder.java:
##########
@@ -175,7 +194,55 @@ public ConnectorSession build() {
}
return new ConnectorSessionImpl(queryId, user, timeZone, locale,
catalogId, catalogName, catalogProperties, sessionProperties,
sid, cred,
- captureStatementScope());
+ captureStatementScope(), captureOperationControl(),
captureMetadataAccessObserver());
+ }
+
+ private ConnectorMetadataAccessObserver captureMetadataAccessObserver() {
+ if (metadataAccessObserver != null) {
+ return metadataAccessObserver;
+ }
+ ConnectContext ctx = connectContext != null ? connectContext :
ConnectContext.get();
+ if (ctx == null || !ctx.getSessionVariable().enableProfile()) {
+ return ConnectorMetadataAccessObserver.NOOP;
+ }
+ SummaryProfile profile = SummaryProfile.getSummaryProfile(ctx);
+ return profile == null ? ConnectorMetadataAccessObserver.NOOP
+ : event -> profile.recordConnectorMetadataAccess(catalogName,
event);
+ }
+
+ private ConnectorOperationControl captureOperationControl() {
+ if (operationControl != null) {
+ return operationControl;
+ }
+ ConnectContext ctx = connectContext != null ? connectContext :
ConnectContext.get();
+ if (ctx == null) {
+ return ConnectorOperationControl.NONE;
+ }
+ long startMillis = ctx.getStartTime() > 0 ? ctx.getStartTime() :
System.currentTimeMillis();
+ long deadlineMillis = startMillis + ctx.getExecTimeoutS() * 1000L;
+ // The connection may execute another statement later; bind
cancellation to this session's statement.
+ StmtExecutor originatingExecutor = ctx.getExecutor();
Review Comment:
[P1] Do not capture null cancellation state for parsed-statement background
tasks. `ConnectorRewriteGroupTask`, `InsertTask`, and `StreamingInsertTask`
construct `StmtExecutor(ctx, StatementBase)`, whose constructor never registers
itself on `ctx`; their cancel paths mark only that executor. A connector
session built during source/sink planning therefore captures null here and its
cache/pool/retry waits continue until the unrelated deadline. Register every
parsed executor before connector planning (centrally or at each task) and cover
cancellation while background rewrite/insert metadata is blocked.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/TrackingHiveMetaStoreClient.java:
##########
@@ -0,0 +1,58 @@
+// 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.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaHookLoader;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.List;
+
+/** Raw HMS client used under RetryingMetaStoreClient so every retry attempt
is observable. */
+public final class TrackingHiveMetaStoreClient extends HiveMetaStoreClient
implements IMetaStoreClient {
+
+ public TrackingHiveMetaStoreClient(Configuration conf, HiveMetaHookLoader
hookLoader, Boolean allowEmbedded)
+ throws MetaException {
+ super(conf, hookLoader, allowEmbedded);
+ }
+
+ @Override
+ public List<Partition> getPartitionsByNames(String dbName, String
tableName, List<String> partitionNames)
+ throws TException {
+ return HmsRemoteCallTracking.trackWireAttempt(
+ () -> super.getPartitionsByNames(dbName, tableName,
partitionNames));
+ }
+
+ @Override
+ public void reconnect() throws MetaException {
+ HmsRemoteCallTracking.checkReconnectActive();
Review Comment:
[P1] Track the pre-wire reconnect as an interruptible unsafe phase. Between
these two checks, `super.reconnect` may close the old transport and block in
URI resolution/Kerberos/transport open, but neither `retryingAfterWireFailure`
nor `clientUnsafeAfterReconnectFailure` is set. The watchdog therefore records
only a generic abort and cannot return KILL/deadline until setup finishes. Mark
reconnect active before entering `super`; if cancellation interrupts it
mid-transition, taint that client. Add a blocking socket-lifetime
pre-first-wire reconnect test.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java:
##########
@@ -85,8 +105,19 @@ public static Collection<Partition>
getMTMVCanRewritePartitions(MTMV mtmv, Conne
}
if (mtmvNeedComparePartitions == null) {
try {
- mtmvNeedComparePartitions =
getMtmvPartitionsByRelatedPartitions(mtmv, refreshContext,
- queryUsedPartitions);
+ mtmvNeedComparePartitions = Sets.newLinkedHashSet(
+ getMtmvPartitionsByRelatedPartitions(mtmv,
refreshContext, queryUsedPartitions));
+ mtmvNeedComparePartitions.retainAll(partitionsToCompare);
+ MTMVRefreshContext currentRefreshContext = refreshContext;
+ mtmvNeedComparePartitions.removeIf(
+ partitionName ->
!currentRefreshContext.persistedPartitionSetsMatch(partitionName));
+ if (mtmvNeedComparePartitions.isEmpty()) {
+ return res;
+ }
+ Set<TableNameInfo> excludeTables = forceConsistent
+ ? ImmutableSet.of() :
mtmv.getQueryRewriteConsistencyRelaxedTables();
+ refreshContext.preloadSnapshots(mtmvNeedComparePartitions,
Review Comment:
[P1] Do not let the async-MV hook swallow operation aborts from this
preload. The table boundary deliberately rethrows
`ConnectorOperationAbortedException`, but it passes this
`AnalysisException`-only catch and is then caught by
`createAsyncMaterializationContext`'s `catch (Exception)`, which returns an
empty MV list. If KILL/deadline fires here before a coordinator exists,
planning can continue into a local base plan despite the executor's cancel
flag. Please explicitly rethrow operation aborts at the hook boundary and add
`CANCELLED` and `DEADLINE_EXCEEDED` planner-hook tests.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -307,6 +308,8 @@ public enum TableFrom {
// Record mtmv and valid partitions map because this is time-consuming
behavior
private final Map<BaseTableInfo, Collection<Partition>>
mvCanRewritePartitionsMap = new HashMap<>();
+ // Cloud MTMV versions are loaded before planner table locks and
revalidated from their local caches later.
+ private final Map<BaseTableInfo, MTMVRefreshContext>
preloadedMtmvRefreshContexts = new HashMap<>();
Review Comment:
[P1] Clear this execution-scoped context map when a prepared statement
resets MVCC state. PREPARE and EXECUTE reuse the same `StatementContext`;
`resetMvccSnapshots` clears normal pins and the preload completion marker, but
not this map. The next EXECUTE reruns preload, skips `buildContext` here, and
can retain prior external `partitionItems` plus freshness caches while its scan
uses a fresh pin, allowing a stale MTMV rewrite. Clear/rebuild the map under
the execution's exact pin and add PREPARE-to-first-EXECUTE and two-EXECUTE
external-change tests.
--
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]
