github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3964679666
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1918,51 +2005,81 @@ public void startSplit(int numBackends) {
final List<String> allPartitions =
new
ArrayList<>(selectedPartitions.selectedPartitions.keySet());
final int batchSize = sessionVariable.getNumPartitionsInBatchMode();
+ SummaryProfile batchSummaryProfile =
SummaryProfile.getSummaryProfile(ConnectContext.get());
+ final RuntimeProfile batchExecutionSummary = batchSummaryProfile ==
null
+ ? null : batchSummaryProfile.getExecutionSummary();
Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
AtomicReference<UserException> batchException = new
AtomicReference<>(null);
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
-
- CompletableFuture.runAsync(() -> {
- for (int begin = 0; begin < allPartitions.size(); begin +=
batchSize) {
- int end = Math.min(begin + batchSize, allPartitions.size());
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
+ SubmittedTaskFinalizer profileFinalizer = new
SubmittedTaskFinalizer(() -> {
+ try {
+ List<ConnectorScanProfile> profiles =
onPluginClassLoader(scanProvider,
+ () ->
scanProvider.collectScanProfiles(connectorSession));
+ writeScanProfilesInto(batchExecutionSummary, profiles);
+ } catch (Exception e) {
+ UserException profileFailure = new
UserException(e.getMessage(), e);
+ UserException primaryFailure = batchException.get();
+ if (primaryFailure == null) {
+ batchException.compareAndSet(null, profileFailure);
+ } else {
+ primaryFailure.addSuppressed(profileFailure);
}
- List<String> batch = allPartitions.subList(begin, end);
- int curBatchSize = end - begin;
- try {
- CompletableFuture.runAsync(() -> {
- try {
- List<ConnectorScanRange> ranges =
onPluginClassLoader(scanProvider,
- () ->
scanProvider.planScanForPartitionBatch(
- connectorSession, batchRequest,
batch));
- List<Split> batchSplits = new
ArrayList<>(ranges.size());
- for (ConnectorScanRange range : ranges) {
- batchSplits.add(new PluginDrivenSplit(range));
- }
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(batchSplits);
- }
- } catch (Exception e) {
- batchException.set(new
UserException(e.getMessage(), e));
- } finally {
- if (batchException.get() != null) {
-
splitAssignment.setException(batchException.get());
- }
- if (numFinishedPartitions.addAndGet(curBatchSize)
== allPartitions.size()) {
- splitAssignment.finishSchedule();
+ splitAssignment.setException(batchException.get());
+ } finally {
+ splitAssignment.finishSchedule();
+ }
+ });
+
+ Runnable dispatch = () -> {
+ try {
+ for (int begin = 0; begin < allPartitions.size(); begin +=
batchSize) {
+ int end = Math.min(begin + batchSize,
allPartitions.size());
+ if (batchException.get() != null ||
splitAssignment.isStop()) {
+ break;
+ }
+ List<String> batch = allPartitions.subList(begin, end);
+ profileFinalizer.taskSubmitted();
+ try {
+ CompletableFuture.runAsync(() -> {
+ try {
+ List<ConnectorScanRange> ranges =
onPluginClassLoader(scanProvider,
+ () ->
scanProvider.planScanForPartitionBatch(
+ connectorSession,
batchRequest, batch));
+ List<Split> batchSplits = new
ArrayList<>(ranges.size());
+ for (ConnectorScanRange range : ranges) {
+ batchSplits.add(new
PluginDrivenSplit(range));
+ }
+ if (splitAssignment.needMoreSplit()) {
+ splitAssignment.addToQueue(batchSplits);
+ }
+ } catch (Exception e) {
+ batchException.compareAndSet(null, new
UserException(e.getMessage(), e));
+ } finally {
+ if (batchException.get() != null) {
Review Comment:
**[P1] Do not re-publish the same batch exception from every worker.** After
one worker stores `E` in both `batchException` and `SplitAssignment`, the next
admitted worker reaches this `finally` and calls `setException(E)` again.
`SplitAssignment.addUserException` then executes `E.addSuppressed(E)`, which
throws `IllegalArgumentException: Self-suppression not permitted`; because
`taskFinished()` is the following statement, the outstanding count never
reaches zero and the finalizer never drains profiles or calls
`finishSchedule()`. Publish the winning failure once (or make publication
idempotent), guarantee `taskFinished()` in a nested `finally`, and cover two
admitted tasks where one fails before the other completes.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java:
##########
@@ -203,76 +210,293 @@ public List<String> listPartitionNamesFresh(String
dbName, String tableName, int
@Override
public List<HmsPartitionInfo> getPartitions(String dbName, String
tableName, List<String> partNames) {
- if (partNames == null || partNames.isEmpty()) {
- return Collections.emptyList();
- }
- // Per-partition assembly (Trino CachingHiveMetastore / legacy
HiveExternalMetaCache shape): serve each
- // requested partition from its own entry and fetch only the misses in
ONE delegate round-trip, so
- // overlapping requests share partition objects and the capacity
bounds partition OBJECTS, not
- // request-lists. Correctness is independent of name-parse fidelity: a
LOOKUP is keyed by the requested
- // name parsed to values, but a STORE is always keyed by the
partition's OWN values, so a name whose
- // parse diverges (a rare escaped value) simply misses and is
re-fetched — never a wrong or dropped
- // partition. Callers consume the result as a SET (they never rely on
order or 1:1 name↔result
- // correspondence — the delegate get_partitions_by_names never
guaranteed either).
- List<HmsPartitionInfo> result = new ArrayList<>(partNames.size());
- List<String> missNames = null;
- for (String name : partNames) {
- HmsPartitionInfo hit =
- partitionsCache.getIfPresent(new PartitionKey(dbName,
tableName, toPartitionValues(name)));
+ return getPartitionsWithStats(dbName, tableName,
partNames).getPartitions();
+ }
+
+ @Override
+ public HmsPartitionBatchResult getPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames) {
+ return getPartitionsWithStats(dbName, tableName, partNames, false);
+ }
+
+ @Override
+ public List<HmsPartitionInfo> getExistingPartitions(
+ String dbName, String tableName, List<String> partNames) {
+ return getExistingPartitionsWithStats(dbName, tableName,
partNames).getPartitions();
+ }
+
+ @Override
+ public HmsPartitionBatchResult getExistingPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames) {
+ return getPartitionsWithStats(dbName, tableName, partNames, true);
+ }
+
+ private HmsPartitionBatchResult getPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames, boolean
allowMissing) {
+ long logicalStartNanos = System.nanoTime();
+ if (partNames.isEmpty()) {
+ HmsPartitionBatchStats stats = HmsPartitionBatchStats.builder()
+ .logicalElapsedNanos(System.nanoTime() - logicalStartNanos)
+ .build();
+ return new HmsPartitionBatchResult(Collections.emptyList(), stats);
+ }
+ HmsPartitionRequest request = new HmsPartitionRequest(dbName,
tableName, partNames);
+ // Keep the existing cache policy: aggregate every miss into one
logical delegate request and publish
+ // only after that request succeeds. Reassemble from partition
identities afterwards because a mixed
+ // hit/miss request must preserve the caller's exact order even when
HMS returns a different order.
+ List<List<String>> requestedValues = new ArrayList<>(partNames.size());
+ Map<List<String>, HmsPartitionInfo> resultByIdentity = new HashMap<>();
+ List<HmsPartitionIdentity.ParsedPartitionName> misses = new
ArrayList<>();
+ for (HmsPartitionIdentity.ParsedPartitionName partition :
request.getPartitions()) {
+ List<String> values = partition.getValues();
+ requestedValues.add(values);
+ HmsPartitionInfo hit = partitionsCache.getIfPresent(new
PartitionKey(dbName, tableName, values));
if (hit != null) {
- result.add(hit);
+ resultByIdentity.put(values, hit);
} else {
- if (missNames == null) {
- missNames = new ArrayList<>();
+ misses.add(partition);
+ }
+ }
+ PartitionStatsAccumulator physicalStats = new
PartitionStatsAccumulator();
+ if (!misses.isEmpty()) {
+ try {
+ loadMissingPartitions(dbName, tableName, allowMissing, misses,
+ resultByIdentity, physicalStats);
+ } catch (HmsClientException e) {
+ HmsPartitionBatchStats failedStats =
e.getPartitionBatchStats();
+ if (failedStats != null) {
+ physicalStats.add(failedStats);
+ e.withPartitionBatchStats(physicalStats.build(
+ partNames.size(), System.nanoTime() -
logicalStartNanos));
}
- missNames.add(name);
+ throw e;
+ }
+ }
+ List<HmsPartitionInfo> result = new ArrayList<>(partNames.size());
+ for (int i = 0; i < requestedValues.size(); i++) {
+ List<String> values = requestedValues.get(i);
+ if (allowMissing && !resultByIdentity.containsKey(values)) {
+ continue;
+ }
+ HmsPartitionInfo partition = resultByIdentity.get(values);
+ if (partition == null) {
+ throw HmsPartitionResultException.builder(partNames.size(),
resultByIdentity.size())
+ .missing(request.getPartitions().get(i).getName())
+ .build();
}
+ result.add(partition);
}
- 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).
- try (MetaCache.BulkLoad<PartitionKey, HmsPartitionInfo> load =
- partitionsCache.beginBulkLoad(ScopePath.table(dbName,
tableName))) {
- for (HmsPartitionInfo info : delegate.getPartitions(dbName,
tableName, missNames)) {
- load.publish(new PartitionKey(dbName, tableName,
info.getValues()), info);
- result.add(info);
+ HmsPartitionBatchStats stats = physicalStats.build(
+ partNames.size(), System.nanoTime() - logicalStartNanos);
+ return new HmsPartitionBatchResult(result, stats);
+ }
+
+ private void loadMissingPartitions(String dbName, String tableName,
boolean allowMissing,
+ List<HmsPartitionIdentity.ParsedPartitionName> initialMisses,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ PartitionStatsAccumulator physicalStats) {
+ List<HmsPartitionIdentity.ParsedPartitionName> pending = initialMisses;
+ while (!pending.isEmpty()) {
+ // Elect one owner per missing identity. One caller can own a
batch and wait on identities owned by
+ // another caller, so partially overlapping requests still issue
one transport load per identity.
+ PartitionLoadBatch ownedBatch = new
PartitionLoadBatch(allowMissing);
+ List<PartitionLoadRegistration> owned = new ArrayList<>();
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting =
new IdentityHashMap<>();
+ try {
+ for (HmsPartitionIdentity.ParsedPartitionName partition :
pending) {
+ registerPartitionLoad(dbName, tableName, partition,
ownedBatch,
+ resultByIdentity, owned, waiting);
}
+ afterPartitionLoadRegistrationForTest();
+ if (!owned.isEmpty()) {
+ loadOwnedPartitions(dbName, tableName, allowMissing,
registrationsToPartitions(owned),
+ resultByIdentity, physicalStats, ownedBatch);
+ }
+ ownedBatch.complete(null);
+ } catch (RuntimeException | Error failure) {
+ ownedBatch.complete(failure);
+ releaseWaitingBatches(waiting.keySet());
+ throw failure;
+ } finally {
+ releaseOwnedPartitionLoads(ownedBatch);
}
+ List<HmsPartitionIdentity.ParsedPartitionName> retries = new
ArrayList<>();
+ consumeWaitingBatches(waiting, resultByIdentity, retries,
allowMissing);
+ pending = retries;
}
- 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;
+ void afterPartitionLoadRegistrationForTest() {
+ }
+
+ void beforePartitionLoadElectionForTest() {
+ }
+
+ void afterPartitionLoadOwnershipForTest() {
+ }
+
+ int inFlightPartitionLoadCountForTest() {
+ return inFlightPartitionLoads.size();
+ }
+
+ private void registerPartitionLoad(String dbName, String tableName,
+ HmsPartitionIdentity.ParsedPartitionName partition,
PartitionLoadBatch ownedBatch,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<PartitionLoadRegistration> owned,
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting) {
+ PartitionKey key = new PartitionKey(dbName, tableName,
partition.getValues());
+ HmsPartitionInfo hit = partitionsCache.getIfPresent(key);
+ if (hit != null) {
+ resultByIdentity.put(partition.getValues(), hit);
+ return;
+ }
+ beforePartitionLoadElectionForTest();
while (true) {
- while (start < partitionName.length() &&
partitionName.charAt(start) != '=') {
- start++;
+ PartitionLoadBatch existing =
inFlightPartitionLoads.putIfAbsent(key, ownedBatch);
+ if (existing == null) {
+ ownedBatch.claimedKeys.add(key);
+ afterPartitionLoadOwnershipForTest();
+ hit = partitionsCache.getIfPresent(key);
+ if (hit == null) {
+ owned.add(new PartitionLoadRegistration(partition, key));
+ } else {
+ resultByIdentity.put(partition.getValues(), hit);
+ ownedBatch.claimedKeys.remove(key);
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ }
+ return;
}
- start++;
- int end = start;
- while (end < partitionName.length() && partitionName.charAt(end)
!= '/') {
- end++;
+ List<PartitionLoadRegistration> registrations =
waiting.get(existing);
+ if (registrations != null) {
+ registrations.add(new PartitionLoadRegistration(partition,
key));
+ return;
}
- if (start > partitionName.length()) {
- break;
+ if (existing.tryRegisterWaiter()) {
+ waiting.computeIfAbsent(existing, ignored -> new ArrayList<>())
+ .add(new PartitionLoadRegistration(partition, key));
+ return;
+ }
+ inFlightPartitionLoads.remove(key, existing);
+ }
+ }
+
+ private void loadOwnedPartitions(String dbName, String tableName, boolean
allowMissing,
+ List<HmsPartitionIdentity.ParsedPartitionName> owned,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ PartitionStatsAccumulator physicalStats, PartitionLoadBatch
ownedBatch) {
+ List<String> names = new ArrayList<>(owned.size());
+ for (HmsPartitionIdentity.ParsedPartitionName partition : owned) {
+ names.add(partition.getName());
+ }
+ MetaCache.BulkLoad<PartitionKey, HmsPartitionInfo> load =
+ partitionsCache.beginBulkLoad(ScopePath.table(dbName,
tableName));
+ ownedBatch.setLoad(load);
+ HmsPartitionBatchResult loadedResult = allowMissing
+ ? delegate.getExistingPartitionsWithStats(dbName, tableName,
names)
+ : delegate.getPartitionsWithStats(dbName, tableName, names);
+ physicalStats.add(loadedResult.getStats());
+ List<HmsPartitionInfo> loaded = loadedResult.getPartitions();
+ for (HmsPartitionInfo info : loaded) {
+ PartitionKey key = new PartitionKey(dbName, tableName,
info.getValues());
+ load.publish(key, info);
+ resultByIdentity.put(info.getValues(), info);
+ ownedBatch.resolvedPartitions.put(key, info);
+ }
+ }
+
+ private static List<HmsPartitionIdentity.ParsedPartitionName>
registrationsToPartitions(
+ List<PartitionLoadRegistration> registrations) {
+ List<HmsPartitionIdentity.ParsedPartitionName> partitions = new
ArrayList<>(registrations.size());
+ for (PartitionLoadRegistration registration : registrations) {
+ partitions.add(registration.partition);
+ }
+ return partitions;
+ }
+
+ private void releaseOwnedPartitionLoads(PartitionLoadBatch ownedBatch) {
+ for (PartitionKey key : ownedBatch.claimedKeys) {
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ }
+ ownedBatch.releaseOwner();
+ }
+
+ private static void releaseWaitingBatches(Set<PartitionLoadBatch> batches)
{
+ for (PartitionLoadBatch batch : batches) {
+ batch.releaseWaiter();
+ }
+ }
+
+ private void consumeWaitingBatches(
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<HmsPartitionIdentity.ParsedPartitionName> retries, boolean
allowMissing) {
+ List<Map.Entry<PartitionLoadBatch, List<PartitionLoadRegistration>>>
entries =
+ new ArrayList<>(waiting.entrySet());
+ for (int i = 0; i < entries.size(); i++) {
+ try {
+ Map.Entry<PartitionLoadBatch, List<PartitionLoadRegistration>>
entry = entries.get(i);
+ consumeWaitingBatch(entry.getKey(), entry.getValue(),
resultByIdentity, retries, allowMissing);
+ } catch (RuntimeException | Error failure) {
+ for (int remaining = i + 1; remaining < entries.size();
remaining++) {
+ entries.get(remaining).getKey().releaseWaiter();
+ }
+ throw failure;
}
-
values.add(FileUtils.unescapePathName(partitionName.substring(start, end)));
- start = end + 1;
}
- return values;
+ }
+
+ private void consumeWaitingBatch(PartitionLoadBatch batch,
+ List<PartitionLoadRegistration> registrations,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<HmsPartitionIdentity.ParsedPartitionName> retries, boolean
allowMissing) {
+ try {
+ Throwable failure = batch.await();
+ if (failure != null) {
+ if (failure instanceof HmsPartitionResultException
+ && (batch.allowMissing != allowMissing
+ || batch.claimedKeys.size() != registrations.size()
+ || registrations.stream().anyMatch(
+ registration ->
!batch.claimedKeys.contains(registration.key)))) {
+ for (PartitionLoadRegistration registration :
registrations) {
+ retries.add(registration.partition);
+ }
+ return;
+ }
+ rethrow(failure, batch.getFailureStats());
+ }
+ for (PartitionLoadRegistration registration : registrations) {
+ HmsPartitionInfo hit =
partitionsCache.getIfPresent(registration.key);
+ if (hit != null) {
+ resultByIdentity.put(hit.getValues(), hit);
+ } else if (failure == null &&
batch.canReuse(registration.key)) {
Review Comment:
**[P1] Retry a handed-off key that this batch did not resolve.**
`canReuse(key)` only says this batch's table-scoped bulk fence is current; it
does not prove this batch loaded that key. A caller can claim `p1`, acquire a
`p1` waiter, then hit `p1` in the post-claim cache check and remove it from
`claimedKeys`. If the same batch owns `p2`, publishing `p2` at capacity 1 can
evict `p1`; the waiter then enters this branch because the `p2` bulk handle is
current, finds no `resolvedPartitions[p1]`, and falls through without retrying.
Exact callers fail spuriously and omission-tolerant scans silently drop an
existing partition. Retry when the resolved value is absent (or track per-key
provenance), and cover this handoff with `[p1,p2]` and capacity 1.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1909,6 +1995,7 @@ public void startSplit(int numBackends) {
pinRewriteFileScope();
final ConnectorTableHandle handle = currentHandle;
final ConnectorScanPlanProvider scanProvider = resolveScanProvider();
+ registerQueryFinishCallbacks(connectorSession, scanProvider);
Review Comment:
**[P1] Register statement-scope cleanup before fallible batch setup.** The
statement has already memoized its `ConnectorMetadata` when this path starts,
but both batch flavors can return earlier from the sys-table check,
column/projection/filter setup, or snapshot pinning. For Arrow Flight,
`returnResultFromLocal` is already false, so `StatementContext.close()`
deliberately skips `closeAll()` and later query finalization has no registered
scope callback to run; those early failures retain the metadata and any
statement-owned resources. Bind scope cleanup before the first fallible setup
step (provider transaction release can remain provider-dependent), and cover an
Arrow Flight early-setup failure.
##########
fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java:
##########
@@ -143,6 +150,208 @@ public void
planScanForPartitionBatchResolvesOnlyTheBatch() {
Assertions.assertNull(lister.callsPerLocation.get("year=2024/month=02"));
}
+ @Test
+ public void fullScanOmitsPartitionDroppedAfterNameListing() {
+ CountingLister lister = new CountingLister();
+ List<String> listed = Arrays.asList("year=2024/month=01",
"year=2024/month=02");
+ HiveScanPlanProvider provider = provider(
+ new FakeHmsClient(listed, "year=2024/month=01"), lister);
+ HiveTableHandle handle = new HiveTableHandle.Builder("db", "t",
HiveTableType.HIVE)
+ .inputFormat(PARQUET_INPUT_FORMAT)
+ .serializationLib(PARQUET_SERDE)
+ .partitionKeyNames(PART_KEYS)
+ .build();
+
+ List<ConnectorScanRange> ranges = provider.planScan(new FakeSession(),
+ ConnectorScanRequest.builder(handle,
Collections.<ConnectorColumnHandle>emptyList()).build());
+
+ Assertions.assertEquals(1, ranges.size());
+
Assertions.assertNull(lister.callsPerLocation.get("year=2024/month=01"));
+ Assertions.assertEquals(1, (int)
lister.callsPerLocation.get("year=2024/month=02"));
+ }
+
+ @Test
+ public void partitionBatchStatsAreExposedAsOneScanProfile() {
+ HiveScanPlanProvider provider = provider(new FakeHmsClient(), new
CountingLister());
+ HiveTableHandle handle = new HiveTableHandle.Builder("db", "t",
HiveTableType.HIVE)
+ .inputFormat(PARQUET_INPUT_FORMAT)
+ .serializationLib(PARQUET_SERDE)
+ .partitionKeyNames(PART_KEYS)
+ .build();
+ FakeSession session = new FakeSession();
+ ConnectorScanRequest request = ConnectorScanRequest.builder(
+ handle,
Collections.<ConnectorColumnHandle>emptyList()).build();
+
+ provider.planScanForPartitionBatch(session, request,
+ Arrays.asList("year=2024/month=01", "year=2024/month=02"));
+ provider.planScanForPartitionBatch(session, request,
+ Collections.singletonList("year=2024/month=03"));
+
+ List<ConnectorScanProfile> profiles =
provider.collectScanProfiles(session);
+ Assertions.assertEquals(1, profiles.size());
+ ConnectorScanProfile profile = profiles.get(0);
+ Assertions.assertEquals("Connector Metadata Access",
profile.getGroupName());
+ Assertions.assertTrue(profile.getScanLabel().contains("db.t"));
+ Assertions.assertEquals("2",
profile.getMetrics().get("LogicalRequests"));
+ Assertions.assertEquals("0",
profile.getMetrics().get("FailedRequests"));
+ Assertions.assertEquals("3",
profile.getMetrics().get("RequestedItems"));
+ Assertions.assertEquals("2",
profile.getMetrics().get("TransportInvocations"));
+ Assertions.assertEquals("3",
profile.getMetrics().get("TransportItems"));
+ Assertions.assertEquals("2",
profile.getMetrics().get("LargestBatchSize"));
+ Assertions.assertEquals("1",
profile.getMetrics().get("SmallestBatchSize"));
+ Assertions.assertTrue(provider.collectScanProfiles(session).isEmpty());
+ }
+
+ @Test
+ public void
firstFailedPartitionRequestIsExposedForSynchronousAndBatchPlanning() {
+ assertFailedPartitionProfile(false, HmsPartitionBatchStats.builder()
+ .requestedItems(2)
+ .transportInvocations(1)
+ .transportItems(2)
+ .largestBatchSize(2)
+ .smallestBatchSize(2)
+ .build());
+ assertFailedPartitionProfile(true, HmsPartitionBatchStats.builder()
+ .requestedItems(2)
+ .transportInvocations(1)
+ .transportItems(2)
+ .largestBatchSize(2)
+ .smallestBatchSize(2)
+ .build());
+ }
+
+ @Test
+ public void exhaustedFallbackIsExposedForSynchronousAndBatchPlanning() {
+ HmsPartitionBatchStats stats = HmsPartitionBatchStats.builder()
+ .requestedItems(2)
+ .transportInvocations(2)
+ .transportItems(3)
+ .largestBatchSize(2)
+ .smallestBatchSize(1)
+ .fallbackCount(1)
+ .build();
+ assertFailedPartitionProfile(false, stats);
+ assertFailedPartitionProfile(true, stats);
+ }
+
+ private static void assertFailedPartitionProfile(boolean batchPlanning,
HmsPartitionBatchStats stats) {
+ List<String> names = Arrays.asList("year=2024/month=01",
"year=2024/month=02");
+ HmsClientException partitionFailure = new HmsClientException("failed",
stats);
Review Comment:
**[P1] Attach stats to the injected failure.** Because
`HmsPartitionBatchStats` is not a `Throwable`, this call selects
`HmsClientException(String, Object...)`; `String.format("failed", stats)`
ignores the extra argument and leaves `getPartitionBatchStats()` null. The
provider records no failed-request profile, so the following
`collectScanProfiles(session).get(0)` fails in both new test cases instead of
validating the metrics. Please construct a genuinely stats-bearing exception
(or add an appropriate test helper/API) so these tests exercise the intended
failure path.
--
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]
