924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3910118889
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -133,7 +144,32 @@ IcebergSnapshotEntryKey.class,
IcebergSnapshotCacheValue.class, defaultEntryCach
public Table getIcebergTable(ExternalTable dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease != null) {
+ return lease.getIcebergTable();
+ }
+ // Background callers have no deterministic statement boundary. Use a
live catalog load
+ // instead of returning a cache generation that can be evicted
immediately after lookup.
+ return getWritableIcebergTable(dorisTable);
+ }
+
+ ThreadPoolExecutor getIcebergTableExecutor(ExternalTable dorisTable) {
+ IcebergTableCacheValue.Lease lease =
statementLease(dorisTable.getOrBuildNameMapping());
+ if (lease == null || lease.getPlanningExecutor() == null) {
+ return dorisTable.getCatalog().getThreadPoolWithPreAuth();
+ }
+ return lease.getPlanningExecutor();
Review Comment:
Fixed in b0611e134dd. The HMS cache-loader branch now captures
context.getOps() and applies the same post-load
generation/authenticator/mapping-options fence before promote() as the Iceberg
catalog branch. A deterministic reset-during-HMS-load test verifies the stale
generation is rejected, never promoted, and the load context is closed.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -56,6 +60,9 @@
Review Comment:
Fixed in b0611e134dd. Iceberg schema and partition-column projection now
execute entirely inside withSnapshotCacheValue, so the snapshot, retained
table, schema and partition spec stay on one statement/operation lease. Added a
regression that asserts schema/spec lookup occurs while the snapshot lease is
active and does not fall back to the unretained snapshot path.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -494,22 +544,52 @@ private List<HudiSplit> planPartitionSplits(HivePartition
partition) throws IOEx
private void getPartitionsSplits(List<HivePartition> partitions,
List<Split> splits) {
Executor executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
- CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
+ ListingFsViewOwner createdOwner = new ListingFsViewOwner(fsViewLease,
executor);
+ ListingFsViewOwner owner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ owner =
statementContext.getOrRegisterStatementResource(listingFsViewResourceKey, () ->
createdOwner);
+ if (owner != createdOwner) {
+ throw new IllegalStateException("Hudi listing owner was
registered twice");
+ }
+ } catch (RuntimeException e) {
+ createdOwner.discardBeforeSubmission();
+ throw e;
+ }
+ }
+ // The owner now releases the exact fs-view generation after every
accepted task terminates.
+ if (!fsViewReleased.compareAndSet(false, true)) {
+ owner.discardBeforeSubmission();
+ throw new IllegalStateException("Hudi filesystem-view lease has
already been released");
+ }
AtomicReference<Throwable> throwable = new AtomicReference<>();
+ RuntimeException submissionFailure = null;
long startTime = System.currentTimeMillis();
- partitions.forEach(partition -> executor.execute(() -> {
+ for (HivePartition partition : partitions) {
+ TerminalTask task = terminalTask(() -> {
+ try {
+ ensureHmsRuntimeGeneration();
+ getPartitionSplits(partition, splits);
+ ensureHmsRuntimeGeneration();
+ } catch (Throwable t) {
+ throwable.compareAndSet(null, t);
+ }
+ }, () -> { });
try {
- getPartitionSplits(partition, splits);
- } catch (Throwable t) {
- throwable.set(t);
Review Comment:
Fixed in b0611e134dd. BatchFsViewOwner now records each task with its exact
executor, checks cancellation before and after submission, and removes
cancelled not-started FutureTasks from the corresponding ThreadPoolExecutor
queue. The deterministic test blocks both workers, queues producer and
partition tasks, closes the owner, and verifies both queues are drained.
--
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]
