github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3870997121
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java:
##########
@@ -87,6 +88,10 @@ public static Collection<Partition>
getMTMVCanRewritePartitions(MTMV mtmv, Conne
try {
mtmvNeedComparePartitions =
getMtmvPartitionsByRelatedPartitions(mtmv, refreshContext,
queryUsedPartitions);
+ Set<TableNameInfo> excludeTables = forceConsistent
+ ? ImmutableSet.of() :
mtmv.getQueryRewriteConsistencyRelaxedTables();
+ refreshContext.preloadSnapshots(mtmvNeedComparePartitions,
Review Comment:
**[P2] Preserve rewrite's zero-RPC stale short circuits.** This preload runs
from `InitMaterializationContextHook.afterRewrite` after
`StatementContext.lock()`, before each partition reaches the persisted PCT-name
check or the query-selected `contains()` gate. Before this patch, a name-set
mismatch or an empty selected comparison set performed no freshness lookup; it
can now issue the full comparison-union/non-PCT snapshot request (up to the
large Hive batches this PR targets) while all planner read locks remain held
until `planWithoutLock()` returns. That extends the existing lock-held metadata
window to candidates that were previously decided locally. Please apply the
deterministic gates first, preload only the survivors outside the planner-lock
interval with atomic local-state revalidation, and add blocking rewrite tests
that assert mismatched and empty-selection candidates issue no request while
write locks remain acquirable.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java:
##########
@@ -226,49 +296,559 @@ public List<HmsPartitionInfo> getPartitions(String
dbName, String tableName, Lis
}
}
if (missNames != null) {
- // Capture the invalidation generation BEFORE the delegate RPC so
a REFRESH (flush) that races this
- // in-flight cold-cache fetch does not get silently undone by
re-caching the pre-refresh partitions.
- // The pre-D2 code went through partitionsCache.get(key, loader)
-> getWithManualLoad, which had this
- // guard; the per-partition put must restore it
(getTable/listPartitionNames/getTableColumnStatistics
- // still use the guarded get path). The delegate results still
populate the RESULT list directly,
- // preserving the misparse->never-drop safety (only the CACHE put
is generation-guarded).
- long generation = partitionsCache.invalidationGeneration();
- for (HmsPartitionInfo info : delegate.getPartitions(dbName,
tableName, missNames)) {
- partitionsCache.putIfNotInvalidatedSince(
- generation, new PartitionKey(dbName, tableName,
info.getValues()), info);
- result.add(info);
+ loadMissingPartitions(request, missNames, resultByIdentity);
+ }
+ List<HmsPartitionInfo> result = new ArrayList<>(partNames.size());
+ for (int i = 0; i < partNames.size(); i++) {
+
checkOperationActivePeriodically(request.getEffectiveOperationControl(), i);
+ String name = partNames.get(i);
+ List<String> identity = HmsPartitionIdentity.fromName(name);
+ HmsPartitionInfo partition = resultByIdentity.get(identity);
+ if (partition == null) {
+ throw HmsPartitionResultException.builder(partNames.size(),
resultByIdentity.size())
+ .missing(name)
+ .build();
}
+ result.add(partition);
}
+ checkOperationActive(request.getEffectiveOperationControl());
return result;
}
- /**
- * Splits a Hive partition name ("c1=a/c2=b") into its ordered values
("a", "b"), unescaping each via
- * Hive's {@code FileUtils} (already a hms-module dependency — {@code
HmsEventParser} uses it). Semantics
- * match the write path's {@code HiveWriteUtils.toPartitionValues}, so
scan and write correlate partitions
- * identically. Only used to build the per-partition LOOKUP key: a parse
that diverges from the stored
- * partition's own values just misses and re-fetches (never a
wrong/dropped partition), so this is a
- * hit-rate optimization, not a correctness dependency.
- */
- private static List<String> toPartitionValues(String partitionName) {
- List<String> values = new ArrayList<>();
- int start = 0;
+ private void loadMissingPartitions(HmsPartitionRequest request,
List<String> initialMissNames,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity) {
+ if (!partitionsCache.isEffectiveEnabled()) {
+ loadAndCacheMissingPartitions(
+ request, initialMissNames,
partitionsCache.invalidationGeneration(), resultByIdentity);
+ return;
+ }
+ for (int offset = 0; offset < initialMissNames.size(); offset +=
partitionLoadWindowSize) {
+ checkOperationActive(request.getEffectiveOperationControl());
+ int end = Math.min(offset + partitionLoadWindowSize,
initialMissNames.size());
+ loadMissingPartitionWindow(request,
initialMissNames.subList(offset, end), resultByIdentity);
+ }
+ }
+
+ private void loadMissingPartitionWindow(HmsPartitionRequest request,
List<String> initialMissNames,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity) {
+ ConnectorOperationControl operationControl =
request.getEffectiveOperationControl();
+ List<String> pendingNames = initialMissNames;
+ while (!pendingNames.isEmpty()) {
+ checkOperationActive(operationControl);
+ PartitionLoadBatch ownedBatch = new PartitionLoadBatch();
+ List<PartitionLoadRegistration> owned = new ArrayList<>();
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting =
new IdentityHashMap<>();
+ acquirePartitionLoadSlot(request, pendingNames.size());
+ try {
+ try {
+ for (int i = 0; i < pendingNames.size(); i++) {
+ checkOperationActivePeriodically(operationControl, i);
+ registerPartitionLoad(
+ request, pendingNames.get(i), ownedBatch,
resultByIdentity, owned, waiting);
+ }
+ afterPartitionLoadRegistrationForTest();
+ if (!owned.isEmpty()) {
+ loadOwnedPartitions(request, ownedBatch, owned,
resultByIdentity);
+ }
+ ownedBatch.future.complete(PartitionLoadOutcome.success());
+ } catch (RuntimeException | Error e) {
+
ownedBatch.future.complete(PartitionLoadOutcome.failure(e));
+ throw e;
+ } finally {
+ releaseOwnedPartitionLoads(ownedBatch);
+ }
+ } finally {
+ // A pure waiter does not consume HMS capacity. Release the
owner-registration/load budget before
+ // waiting so one slow identity cannot block unrelated cold
loads while pool clients are idle.
+ partitionLoadSlots.release();
+ }
+ List<String> retryNames = new ArrayList<>();
+ for (Map.Entry<PartitionLoadBatch,
List<PartitionLoadRegistration>> entry : waiting.entrySet()) {
+ consumeWaitingBatch(request, entry.getKey(), entry.getValue(),
resultByIdentity, retryNames);
+ }
+ pendingNames = retryNames;
+ }
+ }
+
+ /** Test seam for observing registrations without changing the production
coordination contract. */
+ void afterPartitionLoadRegistrationForTest() {
+ }
+
+ int inFlightPartitionLoadCountForTest() {
+ return inFlightPartitionLoads.size();
+ }
+
+ private void registerPartitionLoad(HmsPartitionRequest request, String
partitionName,
+ PartitionLoadBatch ownedBatch, Map<List<String>, HmsPartitionInfo>
resultByIdentity,
+ List<PartitionLoadRegistration> owned,
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting) {
+ List<String> values = HmsPartitionIdentity.fromName(partitionName);
+ PartitionKey key = new PartitionKey(request.getDbName(),
request.getTableName(), values);
+ ReentrantLock stateLock = partitionStateLock(request.getDbName(),
request.getTableName());
+ acquirePartitionStateLock(stateLock,
request.getEffectiveOperationControl());
+ try {
+ HmsPartitionInfo hit = partitionsCache.getIfPresent(key);
+ if (hit != null) {
+ resultByIdentity.put(values, hit);
+ return;
+ }
+ while (true) {
+ PartitionLoadBatch existing =
inFlightPartitionLoads.putIfAbsent(key, ownedBatch);
+ if (existing == null) {
+ break;
+ }
+ if (!existing.isInvalidated(key)) {
+ waiting.computeIfAbsent(existing, ignored -> new
ArrayList<>())
+ .add(new PartitionLoadRegistration(partitionName,
key));
+ return;
+ }
+ if (inFlightPartitionLoads.replace(key, existing, ownedBatch))
{
+ break;
+ }
+ }
+ ownedBatch.claimedKeys.add(key);
+ // Close the cache-check/register race: a previous owner may have
filled the cache and removed its
+ // future after our first cache check but before this putIfAbsent.
+ hit = partitionsCache.getIfPresent(key);
+ if (hit != null) {
+ ownedBatch.resolvedPartitions.put(key, hit);
+ resultByIdentity.put(values, hit);
+ ownedBatch.claimedKeys.remove(key);
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ return;
+ }
+ owned.add(new PartitionLoadRegistration(partitionName, key));
+ } finally {
+ stateLock.unlock();
+ }
+ }
+
+ private void loadOwnedPartitions(HmsPartitionRequest request,
PartitionLoadBatch ownedBatch,
+ List<PartitionLoadRegistration> owned, Map<List<String>,
HmsPartitionInfo> resultByIdentity) {
+ List<String> ownedNames = new ArrayList<>(owned.size());
+ Map<List<String>, PartitionLoadRegistration> ownedByIdentity = new
HashMap<>();
+ for (PartitionLoadRegistration registration : owned) {
+ ownedNames.add(registration.partitionName);
+ ownedByIdentity.put(registration.key.values, registration);
+ }
+ HmsPartitionRequest missRequest = copiedPartitionRequest(request,
ownedNames)
+ .partitionChunkConsumer((chunkNames, chunkPartitions,
effectiveControl) -> publishOwnedPartitions(
+ request, chunkPartitions, resultByIdentity,
ownedBatch, ownedByIdentity, effectiveControl))
+ .build();
+ List<HmsPartitionInfo> loaded = delegate.getPartitions(missRequest);
+ ConnectorOperationControl effectiveControl =
request.getEffectiveOperationControl();
+ checkOperationActive(effectiveControl);
+ if (ownedByIdentity.isEmpty()) {
+ return;
+ }
+ if (ownedByIdentity.size() != ownedNames.size()) {
+ throw new HmsClientException("HMS delegate invoked the partition
chunk consumer for only part of "
+ + "the request: requested=" + ownedNames.size() + ",
unpublished=" + ownedByIdentity.size());
+ }
+ // Compatibility fallback for legacy/test delegates that implement the
request overload without invoking
+ // its chunk consumer. The production raw client always takes the
zero-extra-validation branch above.
+ List<HmsPartitionInfo> validated =
HmsPartitionBatchLoader.validateAndOrder(
+ ownedNames, loaded, effectiveControl);
+ publishOwnedPartitions(
+ request, validated, resultByIdentity, ownedBatch,
ownedByIdentity, effectiveControl);
+ }
+
+ private void releaseOwnedPartitionLoads(PartitionLoadBatch ownedBatch) {
+ for (PartitionKey key : ownedBatch.claimedKeys) {
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ }
+ }
+
+ private void loadAndCacheMissingPartitions(HmsPartitionRequest request,
List<String> missNames,
+ long generation, Map<List<String>, HmsPartitionInfo>
resultByIdentity) {
+ Set<List<String>> unpublishedIdentities = new LinkedHashSet<>();
+ for (String missName : missNames) {
+ unpublishedIdentities.add(HmsPartitionIdentity.fromName(missName));
+ }
+ HmsPartitionRequest missRequest = copiedPartitionRequest(request,
missNames)
+ .partitionChunkConsumer((chunkNames, chunkPartitions,
effectiveControl) ->
+ publishUncachedPartitions(request, chunkPartitions,
generation,
+ resultByIdentity, unpublishedIdentities,
effectiveControl))
+ .build();
+ List<HmsPartitionInfo> loaded = delegate.getPartitions(missRequest);
+ ConnectorOperationControl effectiveControl =
request.getEffectiveOperationControl();
+ checkOperationActive(effectiveControl);
+ if (unpublishedIdentities.isEmpty()) {
+ return;
+ }
+ if (unpublishedIdentities.size() != missNames.size()) {
+ throw new HmsClientException("HMS delegate invoked the partition
chunk consumer for only part of "
+ + "the request: requested=" + missNames.size()
+ + ", unpublished=" + unpublishedIdentities.size());
+ }
+ List<HmsPartitionInfo> validated =
HmsPartitionBatchLoader.validateAndOrder(
+ missNames, loaded, effectiveControl);
+ publishUncachedPartitions(request, validated, generation,
+ resultByIdentity, unpublishedIdentities, effectiveControl);
+ }
+
+ private void publishUncachedPartitions(HmsPartitionRequest request,
List<HmsPartitionInfo> loaded,
+ long generation, Map<List<String>, HmsPartitionInfo>
resultByIdentity,
+ Set<List<String>> unpublishedIdentities, ConnectorOperationControl
effectiveControl) {
+ for (int i = 0; i < loaded.size(); i++) {
+ checkOperationActivePeriodically(effectiveControl, i);
+ HmsPartitionInfo info = loaded.get(i);
+ if (!unpublishedIdentities.remove(info.getValues())) {
+ throw new HmsClientException(
+ "HMS chunk consumer published an unowned partition: "
+ info.getValues());
+ }
+ PartitionKey key = new PartitionKey(request.getDbName(),
request.getTableName(), info.getValues());
+ partitionsCache.putIfNotInvalidatedSince(generation, key, info);
+ resultByIdentity.put(info.getValues(), info);
+ }
+ checkOperationActive(effectiveControl);
+ }
+
+ private static HmsPartitionRequest.Builder copiedPartitionRequest(
+ HmsPartitionRequest request, List<String> partitionNames) {
+ return HmsPartitionRequest.builder()
+ .database(request.getDbName())
+ .table(request.getTableName())
+ .partitionNames(partitionNames)
+ .source(request.getSource())
+ .operationControl(request.getOperationControl())
+ .metadataAccessObserver(request.getMetadataAccessObserver())
+ .shareBatchExecutionWith(request);
+ }
+
+ private void publishOwnedPartitions(HmsPartitionRequest request,
+ List<HmsPartitionInfo> loaded, Map<List<String>, HmsPartitionInfo>
resultByIdentity,
+ PartitionLoadBatch ownedBatch, Map<List<String>,
PartitionLoadRegistration> ownedByIdentity,
+ ConnectorOperationControl effectiveControl) {
+ for (int i = 0; i < loaded.size(); i++) {
+ checkOperationActivePeriodically(effectiveControl, i);
+ HmsPartitionInfo info = loaded.get(i);
+ PartitionLoadRegistration registration =
ownedByIdentity.remove(info.getValues());
+ if (registration == null) {
+ throw new HmsClientException(
+ "HMS chunk consumer published an unowned partition: "
+ info.getValues());
+ }
+ // Every partition-cache invalidation takes the same table state
lock (flushDb/flushAll take all
+ // stripes). Therefore a relevant refresh either invalidates this
key before this critical section,
+ // making isInvalidated true, or runs after the put and removes
it. Direct publication under that lock
+ // also avoids a refresh of an unrelated table suppressing this
valid result.
+ ReentrantLock stateLock = partitionStateLock(request.getDbName(),
request.getTableName());
+ acquirePartitionStateLock(stateLock, effectiveControl);
+ try {
+ if (!ownedBatch.isInvalidated(registration.key)) {
+ partitionsCache.put(registration.key, info);
+ }
+ ownedBatch.resolvedPartitions.put(registration.key, info);
+ } finally {
+ stateLock.unlock();
+ }
+ resultByIdentity.put(info.getValues(), info);
+ }
+ checkOperationActive(effectiveControl);
+ }
+
+ private void consumeWaitingBatch(HmsPartitionRequest request,
PartitionLoadBatch batch,
+ List<PartitionLoadRegistration> registrations, Map<List<String>,
HmsPartitionInfo> resultByIdentity,
+ List<String> retryNames) {
+ long startNanos = System.nanoTime();
+ boolean success = false;
+ try {
+ boolean retrying = false;
+ for (PartitionLoadRegistration registration : registrations) {
+ awaitPartitionLoad(batch, registration.key,
request.getEffectiveOperationControl());
+ if (batch.isInvalidated(registration.key)) {
+ inFlightPartitionLoads.remove(registration.key, batch);
+ retryNames.add(registration.partitionName);
+ retrying = true;
+ continue;
+ }
+ HmsPartitionInfo partition =
batch.resolvedPartitions.get(registration.key);
+ if (partition != null) {
+ resultByIdentity.put(partition.getValues(), partition);
+ continue;
+ }
+ PartitionLoadOutcome outcome = batch.future.getNow(null);
+ checkOperationActive(request.getEffectiveOperationControl());
+ Throwable ownerFailure = Objects.requireNonNull(
+ Objects.requireNonNull(outcome,
+ "partition load is unresolved but its
completion is not available").failure,
+ "completed partition load has neither a result nor a
failure");
+ if (!isRetryableSharedFailure(ownerFailure)) {
+ rethrow(ownerFailure);
+ }
+ // Only an exception published by the OWNER reaches this
branch. Cancellation, deadline and
+ // interruption of the waiting request itself escape directly
from awaitPartitionLoad and must
+ // never remove or replace a normally-running owner's future.
+ checkOperationActive(request.getEffectiveOperationControl());
+ inFlightPartitionLoads.remove(registration.key, batch);
+ retryNames.add(registration.partitionName);
+ retrying = true;
+ }
+ success = !retrying;
+ } finally {
+ recordPartitionWait(request, registrations.size(), startNanos,
success);
+ }
+ }
+
+ private static void awaitPartitionLoad(
+ PartitionLoadBatch batch, PartitionKey key,
ConnectorOperationControl control) {
+ while (true) {
+ long remainingMillis = checkOperationActive(control);
+ if (batch.isInvalidated(key)
+ || batch.resolvedPartitions.containsKey(key) ||
batch.future.isDone()) {
+ return;
+ }
+ long waitMillis = remainingMillis == Long.MAX_VALUE
+ ? PARTITION_LOAD_WAIT_CHECK_MILLIS
+ : Math.min(remainingMillis,
PARTITION_LOAD_WAIT_CHECK_MILLIS);
+ try {
+ batch.future.get(waitMillis, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // Re-check the waiting request's cancellation and deadline at
a bounded interval.
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ConnectorOperationAbortedException(
+ ConnectorOperationAbortedException.Reason.CANCELLED,
+ "HMS in-flight partition load wait was interrupted");
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ rethrow(cause);
+ throw new AssertionError("unreachable");
+ }
+ }
+ }
+
+ private static boolean isRetryableSharedFailure(Throwable failure) {
+ return failure instanceof ConnectorOperationAbortedException
+ || failure instanceof HmsPartitionResultException;
Review Comment:
**[P2] Share terminal integrity failures across equivalent waiters.** This
predicate makes every `HmsPartitionResultException` retryable. With one owner
plus N identical cold singleton waiters and a stable missing/malformed
response, the owner fails, then each waiter discards that completed failure;
one after another becomes the next owner and repeats the same RPC. The finite
set eventually drains, but one single-flight failure becomes N+1 serialized HMS
calls and roughly N+1 times the remote tail latency (`NONE` has no deadline). A
blanket propagation is not safe because a wider owner batch may fail on an
identity the waiter did not request. Please make integrity outcomes
identity/chunk scoped, or provide one bounded, prioritized isolated retry, so
equivalent waiters share the terminal result while unrelated identities can
retry. Add identical-waiter attempt-count and wider-owner/narrower-waiter 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]
