github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3878132093
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTableCacheValue.java:
##########
@@ -27,35 +27,189 @@
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.SupportsStorageCredentials;
+import java.io.Closeable;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
public class IcebergTableCacheValue {
private volatile Table icebergTable;
+ @Nullable
+ private final ThreadPoolExecutor planningExecutor;
+ private TableCleanupOwner tableCleanupOwner;
+ private final Runnable cleanup;
+ private final AtomicInteger references;
+ private final AtomicBoolean cacheReferenceReleased = new AtomicBoolean();
+ private final AtomicBoolean loaderReferenceReleased;
// The execution authenticator active when this generation was loaded; see
the Paimon
// counterpart for the concurrent catalog-reset rationale.
@Nullable
private volatile
org.apache.doris.common.security.authentication.ExecutionAuthenticator
authenticator;
+ private volatile boolean enableMappingVarbinary;
+ private volatile boolean enableMappingTimestampTz;
private String retainedCurrentSnapshotJson;
private volatile boolean queryIsolationPrepared;
private long retainedTablePayloadBytes;
private MetaCacheSizeEstimate sizeEstimate;
public IcebergTableCacheValue(Table icebergTable) {
+ this(icebergTable, null, () -> null, () -> { }, false);
+ }
+
+ IcebergTableCacheValue(Table icebergTable,
Supplier<IcebergSnapshotCacheValue> ignoredSnapshotSupplier,
+ Runnable cleanup) {
+ this(icebergTable, null, ignoredSnapshotSupplier, cleanup, true);
+ }
+
+ IcebergTableCacheValue(Table icebergTable, ThreadPoolExecutor
planningExecutor,
+ Supplier<IcebergSnapshotCacheValue> ignoredSnapshotSupplier,
Runnable cleanup) {
+ this(icebergTable, planningExecutor, ignoredSnapshotSupplier, cleanup,
true);
+ }
+
+ private IcebergTableCacheValue(Table icebergTable, @Nullable
ThreadPoolExecutor planningExecutor,
+ Supplier<IcebergSnapshotCacheValue> ignoredSnapshotSupplier,
Runnable cleanup, boolean loading) {
+ this(icebergTable, planningExecutor, ignoredSnapshotSupplier, () -> {
}, cleanup, loading);
+ }
+
+ IcebergTableCacheValue(Table icebergTable, @Nullable ThreadPoolExecutor
planningExecutor,
+ Supplier<IcebergSnapshotCacheValue> ignoredSnapshotSupplier,
+ Runnable tableCleanup, Runnable cleanup) {
+ this(icebergTable, planningExecutor, ignoredSnapshotSupplier,
tableCleanup, cleanup, true);
+ }
+
+ private IcebergTableCacheValue(Table icebergTable, @Nullable
ThreadPoolExecutor planningExecutor,
+ Supplier<IcebergSnapshotCacheValue> ignoredSnapshotSupplier,
+ Runnable tableCleanup, Runnable cleanup, boolean loading) {
this.icebergTable =
IcebergSnapshotCacheValue.retainTableGeneration(icebergTable);
+ this.planningExecutor = planningExecutor;
+ this.tableCleanupOwner = new TableCleanupOwner(
Review Comment:
[P1] Coordinate this owner with Iceberg's tracked operations
Unlike the earlier Doris-invalidation case, the closer here is Iceberg's own
tracker. Iceberg 1.10.1 tracks REST vended-IO and every Glue `TableOperations`
in a weak-key
[`FileIOTracker`](https://github.com/apache/iceberg/blob/apache-iceberg-1.10.1/core/src/main/java/org/apache/iceberg/io/FileIOTracker.java#L31-L47)
whose removal listener closes `ops.io()`. This new Doris cleanup owner retains
only the copied `FileIO`: `retainTableGeneration()` replaces the SDK table with
`FrozenTableOperations`, which does not retain the original operations key.
GC/maintenance can therefore close the exact IO while a Doris cache/statement
lease is still using it, and this owner later closes it independently again.
Preserve the SDK tracker key for the whole frozen generation and establish one
coordinated close owner, with a REST/Glue regression that runs weak-key cleanup
while a Doris lease remains.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -552,59 +634,308 @@ private void initPrunedPartitions() throws UserException
{
throw new UserException(ExceptionUtils.getRootCauseMessage(e), e);
}
partitionInit = true;
+ ensureHmsRuntimeGeneration();
}
@Override
public void startSplit(int numBackends) {
+ ensureHmsRuntimeGeneration();
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
+ releaseFsViewOnce();
return;
}
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
+ acquireFsView();
ExecutorService scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
+ Executor producerExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- } catch (InterruptedException e) {
- batchException.set(new UserException(e.getMessage(), e));
- break;
+ BatchFsViewOwner createdOwner = new BatchFsViewOwner(splitAssignment,
fsViewLease);
+ BatchFsViewOwner batchOwner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ batchOwner = statementContext.getOrRegisterStatementResource(
+ batchFsViewResourceKey, () -> createdOwner);
+ if (batchOwner != createdOwner) {
+ createdOwner.finish();
+ throw new IllegalStateException("Hudi batch split owner
was registered twice");
}
- CompletableFuture.runAsync(() -> {
+ } catch (RuntimeException e) {
+ createdOwner.finish();
+ throw e;
+ }
+ }
+
+ BatchFsViewOwner finalBatchOwner = batchOwner;
+ splitAssignment.addCloseable(finalBatchOwner);
+ AtomicInteger pendingTasks = new AtomicInteger(1); // producer
reference
+ Runnable taskFinished = () -> {
+ if (pendingTasks.decrementAndGet() == 0) {
+ finishBatchSplit(finalBatchOwner, startTime);
+ }
+ };
+ TerminalTask producerTask = terminalTask(() -> {
+ try {
+ ensureHmsRuntimeGeneration();
+ for (HivePartition partition : prunedPartitions) {
+ if (batchException.get() != null ||
splitAssignment.isStop()) {
+ break;
+ }
try {
- List<Split> allFiles = Lists.newArrayList();
- getPartitionSplits(partition, allFiles, false);
- if (allFiles.size() > numSplitsPerPartition.get()) {
- numSplitsPerPartition.set(allFiles.size());
- }
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(allFiles);
- }
- } catch (Exception e) {
- batchException.set(new UserException(e.getMessage(),
e));
- } finally {
+ splittersOnFlight.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ recordBatchException(e);
+ break;
+ }
+ if (batchException.get() != null ||
splitAssignment.isStop()) {
splittersOnFlight.release();
- if (batchException.get() != null) {
- splitAssignment.setException(batchException.get());
- }
- if (numFinishedPartitions.incrementAndGet() ==
prunedPartitions.size()) {
- if (getSummaryProfile() != null) {
-
getSummaryProfile().addExternalTableGetFileScanTasksTime(
- System.currentTimeMillis() -
startTime);
+ break;
+ }
+ pendingTasks.incrementAndGet();
+ TerminalTask partitionTask = terminalTask(() -> {
+ try {
+ ensureHmsRuntimeGeneration();
+ List<Split> allFiles = Lists.newArrayList();
+ getPartitionSplits(partition, allFiles, false);
+ if (allFiles.size() > numSplitsPerPartition.get())
{
+ numSplitsPerPartition.set(allFiles.size());
+ }
+ if (splitAssignment.needMoreSplit()) {
+ ensureHmsRuntimeGeneration();
+ splitAssignment.addToQueue(allFiles);
}
- splitAssignment.finishSchedule();
+ } catch (Throwable t) {
+ recordBatchException(t);
}
+ }, () -> {
+ splittersOnFlight.release();
+ taskFinished.run();
+ });
+ finalBatchOwner.track(partitionTask);
+ try {
+ scheduleExecutor.execute(partitionTask);
+ } catch (RuntimeException e) {
+ recordBatchException(e);
+ partitionTask.cancelBeforeStart();
+ break;
}
- }, scheduleExecutor);
+ }
+ } catch (Throwable t) {
+ recordBatchException(t);
+ }
+ }, taskFinished);
+ finalBatchOwner.track(producerTask);
+ try {
+ producerExecutor.execute(producerTask);
+ } catch (RuntimeException e) {
+ recordBatchException(e);
+ producerTask.cancelBeforeStart();
+ }
+ }
+
+ private TerminalTask terminalTask(Runnable task, Runnable taskFinished) {
+ return new TerminalTask(task, taskFinished);
+ }
+
+ @VisibleForTesting
+ static class TerminalTask extends FutureTask<Void> {
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final AtomicBoolean interruptRequested = new AtomicBoolean();
+ private final Runnable taskFinished;
+ private volatile Thread runner;
+ private volatile Runnable ownerDone = () -> { };
+
+ TerminalTask(Runnable task, Runnable taskFinished) {
+ super(task, null);
+ this.taskFinished = taskFinished;
+ }
+
+ @Override
+ public void run() {
+ if (started.compareAndSet(false, true)) {
+ runner = Thread.currentThread();
+ if (interruptRequested.get()) {
+ runner.interrupt();
+ }
+ try {
+ super.run();
+ } finally {
+ runner = null;
+ }
+ }
+ }
+
+ boolean cancelBeforeStart() {
+ return started.compareAndSet(false, true) && cancel(false);
+ }
+
+ void requestStop() {
+ if (cancelBeforeStart()) {
+ return;
+ }
+ interruptRequested.set(true);
+ Thread runningThread = runner;
+ if (runningThread != null) {
+ runningThread.interrupt();
+ }
+ }
+
+ void setOwnerDone(Runnable ownerDone) {
+ this.ownerDone = ownerDone;
+ }
+
+ @Override
+ protected void done() {
+ try {
+ taskFinished.run();
+ } finally {
+ ownerDone.run();
+ }
+ }
+ }
+
+ private void recordBatchException(Throwable t) {
+ UserException failure = new UserException(t.getMessage(), t);
+ if (batchException.compareAndSet(null, failure)) {
+ // Consumers should observe the known failure immediately; sibling
terminal accounting
+ // remains independent and still owns the filesystem-view lease
until every task exits.
+ splitAssignment.setException(failure);
+ }
+ }
+
+ private void finishBatchSplit(BatchFsViewOwner batchOwner, long startTime)
{
+ try {
+ if (getSummaryProfile() != null) {
+
getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis()
- startTime);
+ }
+ splitAssignment.finishSchedule();
+ } finally {
+ batchOwner.finish();
+ }
+ }
+
+ @VisibleForTesting
+ static class BatchFsViewOwner implements Closeable {
+ private final SplitAssignment splitAssignment;
+ private final HudiFsViewCacheValue.Lease lease;
+ private final AtomicBoolean finished = new AtomicBoolean();
+ private final ConcurrentLinkedQueue<TerminalTask> tasks = new
ConcurrentLinkedQueue<>();
+ private final AtomicBoolean stopping = new AtomicBoolean();
+
+ BatchFsViewOwner(SplitAssignment splitAssignment,
HudiFsViewCacheValue.Lease lease) {
+ this.splitAssignment = splitAssignment;
+ this.lease = lease;
+ }
+
+ void finish() {
+ if (finished.compareAndSet(false, true)) {
+ try {
+ lease.close();
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to release Hudi fs-view lease after batch
tasks terminated", e);
+ }
+ }
+ }
+
+ void track(TerminalTask task) {
+ task.setOwnerDone(() -> tasks.remove(task));
+ tasks.add(task);
+ if (stopping.get()) {
+ task.requestStop();
+ }
+ }
+
+ @Override
+ public void close() {
+ if (finished.get() || !stopping.compareAndSet(false, true)) {
+ return;
}
- if (batchException.get() != null) {
- splitAssignment.setException(batchException.get());
+ try {
+ splitAssignment.stop();
+ } catch (RuntimeException e) {
+ tasks.forEach(TerminalTask::requestStop);
+ throw e;
}
- }, scheduleExecutor);
+ tasks.forEach(TerminalTask::requestStop);
+ // Already-started filesystem calls may be blocked in storage code
that does not respond to
+ // interruption. Their TerminalTask.done callbacks retain exact
task accounting and eventually call
+ // finish(), which releases the fs-view lease only after the last
task exits. Cancellation must return
+ // promptly instead of waiting here and wedging statement/Arrow
cleanup behind remote storage.
+ }
+ }
+
+ @VisibleForTesting
+ static class ListingFsViewOwner implements Closeable {
+ private final HudiFsViewCacheValue.Lease lease;
+ private final AtomicInteger pendingTasks = new AtomicInteger(1);
+ private final AtomicBoolean submissionFinished = new AtomicBoolean();
+ private final AtomicBoolean stopping = new AtomicBoolean();
+ private final ConcurrentLinkedQueue<TerminalTask> tasks = new
ConcurrentLinkedQueue<>();
+ private final CompletableFuture<Void> tasksFinished = new
CompletableFuture<>();
+ private final CompletableFuture<Void> cancelled = new
CompletableFuture<>();
+
+ ListingFsViewOwner(HudiFsViewCacheValue.Lease lease) {
+ this.lease = lease;
+ }
+
+ void track(TerminalTask task) {
+ pendingTasks.incrementAndGet();
+ task.setOwnerDone(() -> {
+ tasks.remove(task);
+ taskDone();
+ });
+ tasks.add(task);
+ if (stopping.get()) {
+ task.requestStop();
+ }
+ }
+
+ void submissionDone() {
+ if (submissionFinished.compareAndSet(false, true)) {
+ taskDone();
+ }
+ }
+
+ void discardBeforeSubmission() {
+ close();
+ submissionDone();
+ }
+
+ private void taskDone() {
+ if (pendingTasks.decrementAndGet() == 0) {
+ try {
+ lease.close();
+ tasksFinished.complete(null);
+ } catch (RuntimeException e) {
+ tasksFinished.completeExceptionally(e);
+ }
+ }
+ }
+
+ void awaitCompletion() {
+ try {
+ CompletableFuture.anyOf(tasksFinished, cancelled).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ close();
+ throw new CancellationException("Hudi split listing was
interrupted");
+ } catch (java.util.concurrent.ExecutionException e) {
+ throw new
RuntimeException(ExceptionUtils.getRootCauseMessage(e), e);
+ }
+ if (cancelled.isDone() && !tasksFinished.isDone()) {
+ throw new CancellationException("Hudi split listing was
cancelled");
+ }
+ }
+
+ @Override
+ public void close() {
+ if (stopping.compareAndSet(false, true)) {
+ tasks.forEach(TerminalTask::requestStop);
Review Comment:
[P1] Publish cancellation before terminal task callbacks
This is distinct from the earlier uninterruptible-wait issue: the
cancellation-aware owner can now report cancellation as successful completion.
`close()` requests task stops before completing `cancelled`. Cancelling a
not-yet-started `FutureTask` invokes `done()` synchronously, so the final
callback can complete `tasksFinished` before cancellation is visible; if both
futures are complete, line 927 also treats that state as success.
`getPartitionsSplits()` can then return only the partitions that finished
before statement cancellation. Publish a cancellation outcome before stopping
tasks and make it win deterministically in `awaitCompletion()`; please cover
one completed partition plus one accepted task cancelled before start.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java:
##########
@@ -73,8 +86,22 @@ public class HMSExternalCatalog extends ExternalCatalog {
//for "type" = "hms" , but is iceberg table.
private IcebergMetadataOps icebergMetadataOps;
+ private IcebergCatalogResourceTracker icebergResourceTracker = new
IcebergCatalogResourceTracker();
private volatile AbstractHiveProperties hmsProperties;
+ private AtomicLong runtimeGeneration = new AtomicLong();
+
+ public long getRuntimeGeneration() {
+ return runtimeGeneration.get();
+ }
+
+ @Override
+ public synchronized void modifyCatalogProps(Map<String, String> props) {
Review Comment:
[P1] Include replay's pre-write in the runtime fence
This is distinct from the earlier failed-validation/rollback window: even
after detached validation, `isReplay=true` still always pre-publishes. The
changed override only fences `modifyCatalogProps()`, but
`alterExternalCatalogPropsFenced(..., isReplay=true)` first calls inherited
`tryModifyCatalogProps()`, which mutates `CatalogProperty` without this monitor
or a `runtimeGeneration` increment. Concurrent queries do not take the
catalog-manager write lock, so an HMS Iceberg load can capture new
mapping/storage properties with the old ops, authenticator, executor, and
tracker (native Iceberg has the same window), while Hudi's old generation still
passes. The later synchronized reset cannot repair an already-borrowed mixed
generation. Make replay publish/reset atomically under the same catalog fence
instead of pre-publishing, and cover a load paused between the replay pre-write
and final reset.
--
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]
