CalvinKirs commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3879685062
##########
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:
Fixed in 9487b21cb47. Before building each cloud MTMV refresh context, the
preload stage now installs/reuses the PCT table snapshot through
StatementContext.loadSnapshots. The added test verifies snapshot loading
precedes MTMVRefreshContext.buildContext, keeping mapping/freshness/scan on the
statement pin.
##########
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:
Fixed in 9487b21cb47. StatementContext.resetMvccSnapshots now clears
preloadedMtmvRefreshContexts together with the other execution-scoped snapshot
state. StatementContextTest verifies a preloaded MTMV context is removed on
reset.
##########
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:
Resolved by scope reduction in 9487b21cb47. The query cancellation watchdog
and its phase model were removed. Authentication cancellation is no longer
claimed and is explicitly outside this PR scope.
##########
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:
Resolved by scope reduction in 9487b21cb47. The asynchronous/watchdog
cancellation lifecycle around outer Kerberos authentication was removed.
Authentication cancellation is no longer claimed and is explicitly outside this
PR scope.
##########
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:
Resolved by scope reduction in 9487b21cb47. The reconnect
cancellation/watchdog lifecycle was removed together with query-level HMS
cancellation support. TrackingHiveMetaStoreClient now tracks only physical
getPartitionsByNames attempts for observability.
##########
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:
Resolved by scope reduction in 9487b21cb47. Query-level
cancellation/deadline propagation has been removed from this PR, including
ConnectorOperationControl and all session/control wiring. The PR description
now explicitly lists name listing and the complete HMS cancellation lifecycle
as out of scope.
##########
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:
Resolved by scope reduction in 9487b21cb47.
ConnectorOperationAbortedException and query-level cancellation/deadline
propagation were removed from this PR, so there is no operation-abort exception
for this hook to swallow.
##########
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:
Fixed in 9487b21cb47. The optional cloud MTMV candidate boundary now catches
AnalysisException and DorisConnectorException per candidate, logs the failure,
and continues base-table analysis. PreloadExternalMetadataTest covers this
connector-failure path.
##########
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:
Resolved by scope reduction in 9487b21cb47. Connector session cancellation
capture and the StmtExecutor cancellation state added by this PR were removed.
Background-task cancellation propagation is no longer claimed by this PR and is
explicitly out of scope.
--
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]
