HDDS-14945. Implement Iceberg position delete file rewrite for path migration

[sreejasahithi]

What changes were proposed in this pull request?

Position delete files of an Iceberg table contain the absolute path to the data files which contains the rows deleted.
We need to rewrite these position delete files as part of path migration. For each selected position delete file, we can take help of Iceberg's RewriteTablePathUtil to change sourcePrefix to targetPrefix for each data file absolute path mentioned in it, and add the rewritten position delete file to a staging location.

Introduce OzonePositionDeleteReaderWriter, which implements Iceberg’s PositionDeleteReaderWriter, to perform format-specific reads and writes for Avro, Parquet, and ORC.

Also added test coverage wrt position delete files and manifest file.

What is the link to the Apache JIRA

HDDS-14945

How was this patch tested?

Updated testcases
Green CI : https://github.com/sreejasahithi/ozone/actions/runs/26036208134

[sreejasahithi]

@ashishkumar50 could you please review this patch.

[peterxcli]

LGTM +1

[peterxcli]

I think there must be a cleaner way to do this.

[peterxcli]

The ‎executorService is already a bounded thread pool with ‎parallelism threads

Suggested change

	Semaphore semaphore = new Semaphore(maxInFlight);
	ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(executorService);
	int submittedTasks = 0;
	int completedTasks = 0;
	try {
	for (DeleteFile deleteFile : toRewrite) {
	semaphore.acquire();
	boolean taskSubmitted = false;
	try {
	completionService.submit(() -> {
	try {
	rewritePositionDelete(deleteFile, table, sourcePrefix, targetPrefix, stagingDir, posDeleteReaderWriter);
	return null;
	} finally {
	semaphore.release();
	}
	});
	taskSubmitted = true;
	submittedTasks++;
	} finally {
	if (!taskSubmitted) {
	semaphore.release();
	}
	}
	List<Future<Void>> futures = new ArrayList<>();
	for (DeleteFile deleteFile : toRewrite) {
	futures.add(executorService.submit(() -> {
	rewritePositionDelete(deleteFile, table, sourcePrefix,
	targetPrefix, stagingDir, posDeleteReaderWriter);
	return null;
	}));
	}
	for (Future<Void> f : futures) {
	try {
	f.get();
	} catch (ExecutionException e) {
	executorService.shutdownNow();
	throw new RuntimeException(
	"Failed to rewrite position delete file", e.getCause());
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	executorService.shutdownNow();
	throw new RuntimeException(
	"Interrupted while rewriting position delete files", e);
	}
	}

[sreejasahithi]

The Semaphore + ExecutorCompletionService pattern here is intentional, and here is the reasoning:

Executors.newFixedThreadPool uses an unbounded LinkedBlockingQueue internally. While the thread count limits concurrent execution, all submitted tasks are still queued immediately. For a table with a very large number of position delete files, submitting everything upfront would create a task wrapper per file and can lead to memory pressure proportional to the number of files.

The semaphore provides explicit backpressure and keeps the number of tasks admitted into the executor bounded at any point in time.

The ExecutorCompletionService polling inside the submission loop allows us to opportunistically drain already completed tasks during submission. This helps surface task failures earlier during execution compared to waiting for a final drain phase after all submissions are completed.

A bounded ThreadPoolExecutor with an ArrayBlockingQueue is an alternative, but when the queue fills, behaviour depends on the rejection policy:

• CallerRunsPolicy changes execution semantics by running tasks on the submitting thread, which can stall submission progress, this naturally slows down the rate of new task submissions.
• custom RejectedExecutionHandler provides similar backpressure behavior to Semaphore.acquire(), but moves the control flow into executor internals rather than keeping it explicit at the call site.

In both cases, ExecutorCompletionService is still required for result collection and error propagation. The semaphore keeps the backpressure and concurrency limits explicit and visible in the code, rather than implicit in executor configuration.

[peterxcli]

Got it, make sense. Thanks for the explanation!

[peterxcli]

I also noticed the Semaphore + ExecutorCompletionService pattern is repeated across every rewriteXXX method. We could extract it into a shared utility as a follow up.

[sreejasahithi]

Yes, that cleanup is under my consideration. I think it would be better to take it up in a separate PR since it would change the scope/context of this PR.

[peterxcli]

@sreejasahithi curious if you have any followup for this specifically? I just encounter the similar issue, would appreciate the cleanup solution from your side, thanks!

[sreejasahithi]

@peterxcli I haven't had time to open the followup yet. It is on my list, will pick it up once I have the bandwidth.
Thanks.

[adoroszlai]

Thanks @sreejasahithi for the patch. I would like to know how ozone-iceberg.jar is used in practice.

• Are you supposed to copy all 14 new dependency jars to make it work? If so, its usage is getting cumbersome, and we should provide a fat jar, like ozone-filesystem-hadoop3 for Hadoop environment.
• Or are these already available in Iceberg environment? If so, then the dependencies should be added with provided scope, and shouldn't be part of the Ozone binary distribution.

[adoroszlai]

Thanks @sreejasahithi for updating the patch. Can you please split build changes into a separate PR, to be merged before this one? I will review it.

[sreejasahithi]

Reverted the fat jar commit as RewriteTablePathOzoneAction will be invoked via CLI command RewriteTablePathCommand(will add in future patch) in Ozone so we only need the fat jar as it is not used in other environment.

[sreejasahithi]

@ashishkumar50 could you please review this patch.

[ashishkumar50]

@sreejasahithi thanks for the patch, LGTM with some minor nits.

[ashishkumar50]

LGTM

[peterxcli]

Thanks @sreejasahithi for the patch, @adoroszlai, @ashishkumar50 for the review!