github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3868649352
##########
fe/fe-core/src/main/java/org/apache/doris/connector/DefaultConnectorContext.java:
##########
@@ -164,6 +166,7 @@ public DefaultConnectorContext(String catalogName, long
catalogId,
this.rawStoragePropsSupplier =
Objects.requireNonNull(rawStoragePropsSupplier,
"rawStoragePropsSupplier");
this.environment = buildEnvironment();
+ this.metadataAccessMetrics = new
ConnectorMetadataAccessMetrics(catalogName);
Review Comment:
**[P2] Give the validation metrics reference an owner.** This constructor
now acquires a shared per-catalog metrics reference even for the temporary
context created by `forCatalogCreationValidation()`. `CatalogFactory` passes
that context inline and retains only the connector, while
`PluginDrivenExternalCatalog` explicitly leaves `connectorContext` null for
this validation connector, so neither initialization nor catalog teardown can
call `DefaultConnectorContext.close()` on it. Each create/replay attempt
therefore leaves an entry in `SHARED_METRICS`; after the live catalog records
metrics, DROP also cannot unregister those catalog-labelled series because the
leaked reference keeps the count nonzero. Please make validation use a
non-acquiring metrics sink or give the temporary context an explicit owner that
closes it on every success/failure/fallback path. The same ownership rule is
also needed for live initialization: construct into a local context, publish it
only after connecto
r creation succeeds, and close it on null/throw so repeated retries cannot
overwrite and leak failed contexts.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java:
##########
@@ -282,9 +862,16 @@ public List<HmsColumnStatistics>
getTableColumnStatistics(String dbName, String
/** Drop every cached entry for one table. Backs {@code REFRESH TABLE}. */
public void flush(String dbName, String tableName) {
+ ReentrantLock stateLock = partitionStateLock(dbName, tableName);
+ stateLock.lock();
+ try {
+ invalidateInFlightPartitionLoads(key -> key.matches(dbName,
tableName), true);
+ } finally {
+ stateLock.unlock();
Review Comment:
**[P1] Keep registration fenced through the cache clear.** The state lock is
released before `partitionsCache.invalidateIf()` bumps the generation. A cold
request can therefore register after the in-flight scan, start its HMS RPC,
then let this refresh clear the cache and return; because that new batch was
never marked invalid and `publishOwnedPartitions()` uses a direct `put`, its
pre-clear load is cached afterward for the full TTL. The same gap exists in
partition/DB/catalog invalidation. Please perform the matching cache
invalidation under the same stripe(s), or capture/check a refresh epoch at
owner publication, and add the mark/register/clear/publish interleaving to the
concurrency tests.
##########
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()) {
Review Comment:
**[P1] Preserve load admission when partition caching is disabled.** This
early return skips both single-flight retention *and* the new window/slot
limiter. In the supported `hive.metastore.client.pool.size=0` configuration,
the constructor deliberately converts zero to one cold-load slot, but every
disabled-cache request now bypasses that slot and `ThriftHmsClient` creates a
fresh client per call; N concurrent scans/freshness probes can therefore open N
HMS connections. Please keep windowing and slot admission on this path while
skipping only cache coordination/publication, and cover zero-pool plus disabled
cache concurrently.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -1422,13 +1456,34 @@ public OptionalLong
getPartitionFreshnessMillis(ConnectorSession session, Connec
if (!(handle instanceof HiveTableHandle)) {
return siblingMetadata(session,
handle).getPartitionFreshnessMillis(session, handle, partitionName);
}
- HiveTableHandle hiveHandle = (HiveTableHandle) handle;
- List<HmsPartitionInfo> partitions =
hmsClient.getPartitions(hiveHandle.getDbName(),
- hiveHandle.getTableName(),
Collections.singletonList(partitionName));
- if (partitions.isEmpty()) {
+ Map<String, Long> freshness = getPartitionFreshnessMillis(
+ session, handle, Collections.singletonList(partitionName));
+ Long partitionFreshness = freshness.get(partitionName);
+ if (partitionFreshness == null) {
return OptionalLong.empty();
}
- return
OptionalLong.of(lastDdlMillis(partitions.get(0).getParameters()));
+ return OptionalLong.of(partitionFreshness);
+ }
+
+ @Override
+ public Map<String, Long> getPartitionFreshnessMillis(ConnectorSession
session, ConnectorTableHandle handle,
+ List<String> partitionNames) {
+ if (!(handle instanceof HiveTableHandle)) {
+ return siblingMetadata(session, handle)
+ .getPartitionFreshnessMillis(session, handle,
partitionNames);
+ }
+ if (partitionNames.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ HiveTableHandle hiveHandle = (HiveTableHandle) handle;
+ List<HmsPartitionInfo> partitions = hmsClient.getPartitions(
+ session, HmsPartitionAccessSource.MTMV,
Review Comment:
**[P2] Preserve the display source in freshness telemetry.** `SHOW
PARTITIONS` now builds and preloads `MTMVRefreshContext`, reaches these
freshness methods, and is always emitted as `MTMV` here; the sibling
whole-table freshness call is hard-coded the same way. There is no production
use of the newly added `HmsPartitionAccessSource.DISPLAY`, so display traffic
is indistinguishable from refresh/rewrite work in both process metrics and
Query Profile despite the per-source observability contract. Please thread the
logical access purpose into this freshness request and emit `DISPLAY` for the
proc/display path, with a production-chain test.
##########
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;
+ }
+
+ private static void rethrow(Throwable failure) {
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new HmsClientException("HMS in-flight partition load failed",
failure);
+ }
+
+ private static void recordPartitionWait(HmsPartitionRequest request, int
requestedItems,
+ long startNanos, boolean success) {
+ recordPartitionCoordinationWait(
+ request, PARTITION_INFLIGHT_WAIT_OPERATION, requestedItems,
startNanos, success);
+ }
+
+ private static void recordPartitionCoordinationWait(HmsPartitionRequest
request, String operation,
+ int requestedItems, long startNanos, boolean success) {
+ ConnectorMetadataAccessEvent event =
ConnectorMetadataAccessEvent.builder()
+ .operation(operation)
+ .source(request.getSource().name())
+ .requestedItems(requestedItems)
+
.logicalElapsedMillis(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos))
+ .success(success)
+ .build();
+ try {
+ request.getMetadataAccessObserver().record(event);
Review Comment:
**[P2] Publish coordination waits to process metrics too.** Both in-flight
and cold-load-slot wait events are recorded only through the request observer,
which is a Query Profile sink and becomes `NOOP` when profiling is disabled.
The catalog observer used by the raw loader never reaches this cache layer, so
FE metrics permanently omit both newly advertised wait operations even while
reporting their downstream HMS call. Please give the cache the catalog observer
and safely publish these events to both sinks, with process-metric coverage
when query profiling is disabled.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsPartitionBatchLoader.java:
##########
@@ -0,0 +1,458 @@
+// 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.ConnectorMetadataAccessEvent;
+import org.apache.doris.connector.spi.ConnectorMetadataAccessObserver;
+import org.apache.doris.connector.spi.ConnectorOperationAbortedException;
+import org.apache.doris.connector.spi.ConnectorOperationControl;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+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.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
+/** The single chunking, adaptive fallback, integrity and observability
implementation for HMS partitions. */
+final class HmsPartitionBatchLoader {
+
+ private static final Logger LOG =
LogManager.getLogger(HmsPartitionBatchLoader.class);
+
+ @FunctionalInterface
+ interface Fetcher {
+ List<HmsPartitionInfo> fetch(String dbName, String tableName,
List<String> partitionNames,
+ ConnectorOperationControl operationControl) throws Exception;
+ }
+
+ @FunctionalInterface
+ interface TrackedFetcher {
+ List<HmsPartitionInfo> fetch(String dbName, String tableName,
List<String> partitionNames,
+ ConnectorOperationControl operationControl, RemoteCallTracker
remoteCallTracker) throws Exception;
+ }
+
+ @FunctionalInterface
+ interface FailureClassifier {
+ boolean isDegradable(Throwable failure);
+ }
+
+ private final int maxBatchSize;
+ private final int minBatchSize;
+ private final long fallbackTimeoutMillis;
+ private final TrackedFetcher fetcher;
+ private final FailureClassifier failureClassifier;
+ private final ConnectorMetadataAccessObserver observer;
+ private final LongSupplier nanoTime;
+
+ private HmsPartitionBatchLoader(Builder builder) {
+ this.maxBatchSize = builder.maxBatchSize;
+ this.minBatchSize = builder.minBatchSize;
+ this.fallbackTimeoutMillis = builder.fallbackTimeoutMillis;
+ this.fetcher = builder.fetcher;
+ this.failureClassifier = builder.failureClassifier;
+ this.observer = builder.observer;
+ this.nanoTime = builder.nanoTime;
+ }
+
+ static Builder builder() {
+ return new Builder();
+ }
+
+ List<HmsPartitionInfo> load(HmsPartitionRequest request) {
+ List<String> names = request.getPartitionNames();
+ if (names.isEmpty()) {
+ return java.util.Collections.emptyList();
+ }
+ long startNanos = nanoTime.getAsLong();
+ long fallbackTimeoutNanos =
TimeUnit.MILLISECONDS.toNanos(fallbackTimeoutMillis);
+ long fallbackStartNanos = request.fallbackStartNanos();
+ RemoteCallTracker remoteCalls = new RemoteCallTracker(nanoTime);
+ int fallbackCount = 0;
+ boolean success = false;
+ try {
+ List<HmsPartitionInfo> result = new ArrayList<>(names.size());
+ int offset = 0;
+ int effectiveBatchSize = request.effectiveBatchSize(maxBatchSize);
+ while (offset < names.size()) {
+ checkActive(request, fallbackStartNanos, fallbackTimeoutNanos);
+ int batchSize = Math.min(effectiveBatchSize, names.size() -
offset);
+ List<String> batch = new ArrayList<>(names.subList(offset,
offset + batchSize));
+ List<HmsPartitionInfo> returned;
+ ConnectorOperationControl effectiveControl =
operationControlForAttempt(
+ request, fallbackStartNanos, fallbackTimeoutNanos);
+ try {
+ returned = fetcher.fetch(
+ request.getDbName(), request.getTableName(), batch,
+ effectiveControl, remoteCalls);
+ } catch (HmsRemoteCallException e) {
+ checkActive(request, fallbackStartNanos,
fallbackTimeoutNanos);
+ if (batchSize <= minBatchSize ||
!failureClassifier.isDegradable(e)) {
+ throw finalBatchFailure(
+ request, offset, batchSize, effectiveBatchSize,
+ remoteCalls.count, fallbackCount, e);
+ }
+ if (fallbackStartNanos ==
HmsPartitionRequest.NO_FALLBACK_START_NANOS) {
+ fallbackStartNanos =
request.startFallback(nanoTime.getAsLong());
+ }
+ checkActive(request, fallbackStartNanos,
fallbackTimeoutNanos);
+ effectiveBatchSize = Math.max(minBatchSize, batchSize / 2);
+ request.reduceEffectiveBatchSize(effectiveBatchSize);
+ fallbackCount++;
+ continue;
+ } catch (RuntimeException e) {
+ // Authorization, cancellation and local programming
failures are not transport fallback
+ // candidates. Preserve their original type and stack
instead of disguising them as a failed
+ // HMS batch.
+ throw e;
+ } catch (Exception e) {
+ throw new HmsClientException("Unexpected checked failure
fetching HMS partitions", e);
+ }
+ // Integrity validation and cache publication are deliberately
outside the remote-failure catch:
+ // a malformed response or a local write-back bug must never
trigger transport fallback or be
+ // wrapped as an HMS RPC failure.
+ checkActive(effectiveControl);
+ List<HmsPartitionInfo> ordered = validateAndOrder(batch,
returned, effectiveControl);
+ checkActive(effectiveControl);
+ request.getPartitionChunkConsumer().accept(batch, ordered,
effectiveControl);
+ checkActive(effectiveControl);
+ result.addAll(ordered);
+ checkActive(effectiveControl);
+ offset += batchSize;
+ }
+ checkActive(request, fallbackStartNanos, fallbackTimeoutNanos);
+ success = true;
+ return result;
+ } finally {
+ ConnectorMetadataAccessEvent event =
ConnectorMetadataAccessEvent.builder()
+ .operation("hms.get_partitions_by_names")
+ .source(request.getSource().name())
+ .requestedItems(names.size())
+ .rpcCount(remoteCalls.count)
+ .rpcItems(remoteCalls.items)
+ .largestBatchSize(remoteCalls.largestBatchSize)
+ .smallestBatchSize(remoteCalls.smallestBatchSize ==
Integer.MAX_VALUE
+ ? 0 : remoteCalls.smallestBatchSize)
+ .fallbackCount(fallbackCount)
+
.logicalElapsedMillis(TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() -
startNanos))
+
.rpcElapsedMillis(TimeUnit.NANOSECONDS.toMillis(remoteCalls.elapsedNanos))
+
.maxRpcElapsedMillis(TimeUnit.NANOSECONDS.toMillis(remoteCalls.maxElapsedNanos))
+ .success(success)
+ .build();
+ recordSafely("catalog metrics", observer, event);
+ recordSafely("query profile", request.getMetadataAccessObserver(),
event);
+ }
+ }
+
+ private void recordSafely(String sinkName, ConnectorMetadataAccessObserver
sink,
+ ConnectorMetadataAccessEvent event) {
+ try {
+ sink.record(event);
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to record HMS partition metadata access in {}",
sinkName, e);
+ }
+ }
+
+ private void checkActive(HmsPartitionRequest request,
+ long fallbackStartNanos, long fallbackTimeoutNanos) {
+ checkActive(request.getOperationControl());
+ if (fallbackStartNanos != HmsPartitionRequest.NO_FALLBACK_START_NANOS
+ && nanoTime.getAsLong() - fallbackStartNanos >=
fallbackTimeoutNanos) {
+ throw new ConnectorOperationAbortedException(
+
ConnectorOperationAbortedException.Reason.DEADLINE_EXCEEDED,
+ "HMS partition batch fallback deadline exceeded");
+ }
+ }
+
+ private static void checkActive(ConnectorOperationControl
operationControl) {
+ operationControl.checkActive();
+ if (operationControl.remainingTimeMillis() <= 0) {
+ throw new ConnectorOperationAbortedException(
+
ConnectorOperationAbortedException.Reason.DEADLINE_EXCEEDED,
+ "HMS partition batch request deadline exceeded");
+ }
+ }
+
+ private ConnectorOperationControl
operationControlForAttempt(HmsPartitionRequest request,
+ long fallbackStartNanos, long fallbackTimeoutNanos) {
+ if (fallbackStartNanos == HmsPartitionRequest.NO_FALLBACK_START_NANOS)
{
+
request.updateEffectiveOperationControl(request.getOperationControl());
+ return request.getOperationControl();
+ }
+ ConnectorOperationControl effectiveControl = new
FallbackOperationControl(
+ request.getOperationControl(), fallbackStartNanos,
fallbackTimeoutNanos, nanoTime);
+ request.updateEffectiveOperationControl(effectiveControl);
+ return effectiveControl;
+ }
+
+ /** Applies the fallback budget inside pool waits and
RetryingMetaStoreClient retries, not only between RPCs. */
+ private static final class FallbackOperationControl implements
ConnectorOperationControl {
+ private final ConnectorOperationControl callerControl;
+ private final long fallbackStartNanos;
+ private final long fallbackTimeoutNanos;
+ private final LongSupplier nanoTime;
+
+ private FallbackOperationControl(ConnectorOperationControl
callerControl,
+ long fallbackStartNanos, long fallbackTimeoutNanos,
LongSupplier nanoTime) {
+ this.callerControl = callerControl;
+ this.fallbackStartNanos = fallbackStartNanos;
+ this.fallbackTimeoutNanos = fallbackTimeoutNanos;
+ this.nanoTime = nanoTime;
+ }
+
+ @Override
+ public void checkActive() {
+ HmsPartitionBatchLoader.checkActive(callerControl);
+ if (fallbackRemainingNanos() <= 0) {
+ throw new ConnectorOperationAbortedException(
+
ConnectorOperationAbortedException.Reason.DEADLINE_EXCEEDED,
+ "HMS partition batch fallback deadline exceeded");
+ }
+ }
+
+ @Override
+ public long remainingTimeMillis() {
+ long callerRemainingMillis = callerControl.remainingTimeMillis();
+ long fallbackRemainingMillis = TimeUnit.NANOSECONDS.toMillis(
+ Math.max(0L, fallbackRemainingNanos()));
+ return Math.min(callerRemainingMillis, fallbackRemainingMillis);
+ }
+
+ private long fallbackRemainingNanos() {
+ return fallbackTimeoutNanos - (nanoTime.getAsLong() -
fallbackStartNanos);
+ }
+ }
+
+ static List<HmsPartitionInfo> validateAndOrder(
+ List<String> requestedNames, List<HmsPartitionInfo> returned) {
+ return validateAndOrder(requestedNames, returned,
ConnectorOperationControl.NONE);
+ }
+
+ static List<HmsPartitionInfo> validateAndOrder(List<String> requestedNames,
+ List<HmsPartitionInfo> returned, ConnectorOperationControl
operationControl) {
+ int expectedValueCount =
HmsPartitionIdentity.fromName(requestedNames.get(0)).size();
+ Map<List<String>, Integer> expected = new HashMap<>();
+ List<List<String>> requestedIdentities = new
ArrayList<>(requestedNames.size());
+ for (int i = 0; i < requestedNames.size(); i++) {
+ checkActivePeriodically(operationControl, i);
+ List<String> identity =
HmsPartitionIdentity.fromName(requestedNames.get(i));
+ requestedIdentities.add(identity);
+ Integer previous = expected.put(identity, i);
+ if (previous != null) {
+ throw new IllegalArgumentException(
+ "duplicate partition identity in request: " +
requestedNames.get(i));
+ }
+ }
+ HmsPartitionResultException.Builder failure =
HmsPartitionResultException.builder(
+ requestedNames.size(), returned == null ? 0 : returned.size());
+ List<HmsPartitionInfo> ordered = new
ArrayList<>(java.util.Collections.nCopies(requestedNames.size(), null));
+ Map<List<String>, Integer> returnedCounts = new LinkedHashMap<>();
+ if (returned == null) {
+ failure.invalid("<null response>");
+ } else {
+ for (int i = 0; i < returned.size(); i++) {
+ checkActivePeriodically(operationControl, i);
+ HmsPartitionInfo partition = returned.get(i);
+ if (partition == null) {
+ failure.invalid("<null partition>");
+ continue;
+ }
+ List<String> identity = partition.getValues();
+ if (identity.size() != expectedValueCount) {
+ failure.invalid(identity.toString());
+ continue;
+ }
+ returnedCounts.merge(identity, 1, Integer::sum);
+ Integer index = expected.get(identity);
+ if (index != null && ordered.get(index) == null) {
+ ordered.set(index, partition);
+ }
+ }
+ }
+ for (int i = 0; i < requestedIdentities.size(); i++) {
+ checkActivePeriodically(operationControl, i);
+ if (!returnedCounts.containsKey(requestedIdentities.get(i))) {
+ failure.missing(requestedNames.get(i));
+ }
+ }
+ int returnedIndex = 0;
+ for (Map.Entry<List<String>, Integer> entry :
returnedCounts.entrySet()) {
+ checkActivePeriodically(operationControl, returnedIndex++);
+ if (!expected.containsKey(entry.getKey())) {
+ failure.unexpected(entry.getKey().toString());
+ }
+ if (entry.getValue() > 1) {
+ failure.duplicate(entry.getKey().toString());
+ }
+ }
+ if (failure.hasMismatches()) {
+ throw failure.build();
+ }
+ return ordered;
+ }
+
+ private static void checkActivePeriodically(ConnectorOperationControl
operationControl, int index) {
+ if ((index & 1023) == 0) {
+ checkActive(operationControl);
+ }
+ }
+
+ private RuntimeException finalBatchFailure(HmsPartitionRequest request,
int offset,
+ int failedBatchSize, int effectiveBatchSize, int rpcCount, int
fallbackCount, Exception failure) {
+ if (failure instanceof ConnectorOperationAbortedException
+ || failure instanceof HmsPartitionResultException) {
+ return (RuntimeException) failure;
+ }
+ String message = String.format(
+ "HMS partition batch request failed: db=%s, table=%s,
requested=%d, offset=%d, "
+ + "failedBatchSize=%d, effectiveBatchSize=%d,
minBatchSize=%d, attempts=%d, "
+ + "fallbacks=%d: %s",
+ request.getDbName(), request.getTableName(),
request.getPartitionNames().size(), offset,
+ failedBatchSize, effectiveBatchSize, minBatchSize, rpcCount,
fallbackCount,
+ failure.getMessage());
+ return new HmsClientException(message, failure);
+ }
+
+ static final class RemoteCallTracker {
+ private final LongSupplier nanoTime;
+ private int count;
+ private long items;
+ private long elapsedNanos;
+ private long maxElapsedNanos;
+ private int largestBatchSize;
+ private int smallestBatchSize = Integer.MAX_VALUE;
+
+ RemoteCallTracker(LongSupplier nanoTime) {
+ this.nanoTime = nanoTime;
+ }
+
+ <T> T call(int itemCount, Callable<T> remoteCall) throws Exception {
+ count++;
+ items += itemCount;
+ largestBatchSize = Math.max(largestBatchSize, itemCount);
+ smallestBatchSize = Math.min(smallestBatchSize, itemCount);
+ long startNanos = nanoTime.getAsLong();
+ try {
+ return remoteCall.call();
+ } finally {
+ long currentElapsedNanos = nanoTime.getAsLong() - startNanos;
+ elapsedNanos += currentElapsedNanos;
+ maxElapsedNanos = Math.max(maxElapsedNanos,
currentElapsedNanos);
+ }
+ }
+ }
+
+ static boolean isDegradableRemoteFailure(Throwable failure) {
+ if (!(failure instanceof HmsRemoteCallException)) {
+ return false;
+ }
+ for (Throwable current = failure.getCause(); current != null; current
= current.getCause()) {
+ String className = current.getClass().getName();
+ if (className.endsWith(".TTransportException")) {
Review Comment:
**[P1] Do not halve batches for every transport outage.** This class-name
check makes a closed/refused/reset/EOF/timeout `TTransportException` degradable
even though reducing the payload cannot repair the connection. With the
defaults, one 5,000-name offset can be replayed 13 times down to size 1 within
the 30-second budget, and each logical call sits above Hive's own
retry/reconnect proxy and may create/taint another client. That amplifies an
HMS outage precisely while it is unhealthy. Please restrict fallback to
explicit frame/message/request/partition-limit signals (or a proven oversize
transport code), and make ordinary transport failures terminate after the
original logical attempt.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -240,9 +248,31 @@ public void run() throws JobException {
MTMVPartitionUtil.addPartition(mtmv, partitionKeyDesc);
}
}
+ boolean buildContextUnderLock = Config.isNotCloudMode()
Review Comment:
**[P1] Keep local PCT mappings atomic with their versions in mixed MVs.**
This condition moves the entire context build outside the sorted table locks
whenever *any* base table is MVCC. If the actual PCT table is a local
`OlapTable`, its mapping is copied here at T1, external preload can then block,
and the later locked `refreshLocalBaseVersions()` refreshes only versions—not
`partitionMappings`. A local partition dropped in that window remains in the
mapping and makes the locked version lookup fail; an added partition is omitted
from comparison/refresh. The base code built both together under the locks.
Please split the capture so external pins/I/O stay outside, while local PCT
mappings and versions are rebuilt together under the sorted FE locks. Cloud
local-only plans also always take this branch and the refresh helper is a no-op
there, so preserve an atomic cloud capture as well. Apply the same fix to the
analogous `PartitionsProcDir` branch and add mixed local-PCT/external-M
VCC plus cloud local-only race tests.
##########
fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java:
##########
@@ -73,8 +73,9 @@ public void connectorApiMajorTracksTheRecordedSurfaceChange()
throws IOException
Assertions.assertNotNull(in, "missing connector plugin API version
resource");
version.load(in);
}
- // Write binding gained execution-capability methods in this surface
revision. A plugin built against
- // major 5 must be refused rather than run against a contract it did
not compile against.
+ // Write binding gained execution-capability methods, while metadata
access gained operation control,
Review Comment:
**[P1] Bump the connector SPI major for this surface change.** This PR adds
methods and types to the public connector SPI, but the API is still stamped as
`6.0`. The policy beside `connector.plugin.api.version` requires a same-commit
major bump for *any* SPI surface addition, and `ApiVersionGate` checks only
major equality. As written, a plugin compiled against these new APIs is
labelled 6.0 and can be accepted by an older 6.0 FE, then fail at first use
with `NoSuchMethodError`/`NoClassDefFoundError`. Please bump the connector API
major (and this assertion) to 7.0 in this commit.
##########
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) {
Review Comment:
**[P2] Emit one logical event for the caller's request.** The cache splits
one business request into `partitionLoadWindowSize` windows and each copied
request invokes the raw loader, whose `finally` records a completed logical
event. A cold 12,000-name call therefore increments `LogicalRequests` three
times; if the third window fails, the single caller-visible request is reported
as two successes plus one failure. The cache-disabled path reports the same
call once, so success rates and requested-item metrics depend on cache
configuration rather than business semantics. All-hit and pure-waiter calls
emit no completed business event, and mixed hit/miss calls report only owner
misses. Please move event ownership to the outer cache call, aggregate the
shared business-request state, and emit exactly once while retaining
physical-attempt counters; test all-hit, mixed, pure-waiter, and multi-window
later-failure cases.
--
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]
