Add bounded parallelism to sorted LSM index builds

[justinblethrow-cloud]

What does this PR do?

Adds opt-in, resource-admitted parallelism to the sorted LSM_TREE build path introduced in #5244 and extended to unique indexes in #5256.

schema.buildTypeIndex(typeName, propertyNames)
    .withType(Schema.INDEX_TYPE.LSM_TREE)
    .withBuildMode(IndexBuildMode.SORTED)
    .withBuildParallelism(4)
    .create();

The requested parallelism applies only to work that is already independent:

• materialized external-sort merge groups within one merge generation; and
• final compacted writers for independent bucket sub-indexes.

The default remains one worker. The final global ordered stream remains serial and continues to own key ordering, RID grouping, unique-key enforcement, and publication semantics.

Resource admission and diagnostics

Each stage independently bounds its admitted workers by:

• requested parallelism;
• available work groups or bucket writers;
• processor headroom;
• the sorted-build memory budget; and
• available file descriptors.

The existing sorted-build metrics now report requested, admitted, and observed merge/writer concurrency alongside the existing run, spill, and stage timing data. A request can therefore fall back to one worker without changing correctness or publication behavior.

Failure and publication behavior

Parallel merge workers own disjoint input cursors, output streams, and serialization buffers. Source runs remain intact until every group in a generation succeeds. On the first failure, the dispatcher cancels peers, waits for worker termination, preserves the first cause, and removes incomplete generation output.

Parallel bucket writers own independent staged bucket-index files. A writer failure cancels and joins its peers before the existing sorted-build cleanup removes staged components and spill artifacts. Schema publication still occurs only after all bucket output succeeds.

No new writer, on-disk layout, recovery mechanism, HA behavior, or online-build path is introduced.

Correctness and validation

Coverage includes:

• merge and writer admission against work, CPU, memory, and file-descriptor limits;
• observed overlap for admitted merge groups and bucket writers;
• unique and high-multiplicity nonunique output across buckets and RID chunks;
• preservation of the final global duplicate boundary;
• cancellation, cleanup, and no publication after worker failure;
• serial fallback for one merge group or one bucket;
• metrics, exact/range/ascending/descending reads, planner use, reopen, mutation, crash recovery, compatibility, and later compaction.

Concurrency/failure/metrics gate: 19 tests, 0 failures, 0 errors
Broad sorted-build gate:           42 tests, 0 failures, 0 errors

A full engine run completed 9,347 tests. Every PR3 and adjacent index test passed. It reported three failures outside this patch: one 256-CPU bucket-limit test defect reproduced on untouched main, plus two shared-suite failures that pass in isolation.

Performance evidence

All comparisons use the same binary, fresh fixture copies, a 4 GiB heap, full output digest checks, and reopen validation.

1M, eight source buckets

With a 64 MiB build budget, fan-in 8, and four requested workers:

Serial mean:                4.279 s
Parallel:                   3.327 s
End-to-end improvement:      22.3%
Materialized merge:  0.777 -> 0.279 s  (64.1%)
Final write:         1.077 -> 0.792 s  (26.4%)

Both merge and writer stages admitted and observed four concurrent workers.

10M, one source bucket

A serial/parallel/serial bracket isolated materialized merge work because writer admission was one:

Serial mean:               40.497 s
Parallel mean:             35.339 s
End-to-end improvement:      12.74%
Materialized merge:  7.615 -> 2.417 s  (68.26%)

Every accepted run produced identical ascending and descending digests. These results describe the tested fixtures and configurations, not a general throughput guarantee.

Scope

This remains an explicit, offline, single-server sorted build. Automatic parallelism, SQL syntax, HA/replicated construction, online snapshot-plus-delta construction, and multi-index scheduling remain out of scope.

Related: #5230.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

🟢 Metrics 146 complexity

Metric	Results
Complexity	146

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}

[gemini-code-assist]

Code Review

This pull request introduces multi-threaded parallelism to the sorted LSM index build process in ArcadeDB, parallelizing independent sorted-build merge groups and bucket writers while dynamically admitting workers based on system resources. It also adds a comprehensive SortedIndexBuildMetrics class to collect and log detailed diagnostics, alongside extensive tests and benchmarks. The review feedback is highly constructive, identifying a potential ExecutorService leak in the BucketWriteDispatcher constructor if initialization fails, and suggesting a cleaner way to unwrap and propagate exceptions from worker threads in complete() instead of letting ExecutionException wrap the root cause.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

If an exception is thrown during the initialization of the queues, pending batches, or worker threads (for example, a RejectedExecutionException or OutOfMemoryError), the ExecutorService created at line 492 will be leaked because the constructor fails and the BucketWriteDispatcher is never assigned to a resource variable in the try-with-resources block, preventing close() from being called. Wrapping the initialization loop in a try-catch block and shutting down the executor on failure prevents this resource leak.

    private BucketWriteDispatcher(final Map<LSMTreeIndex, BucketWriter> writers, final int workerCount) {
      this.writers = writers;
      this.queues = new ArrayList<>(workerCount);
      this.pendingBatches = new ArrayList<>(workerCount);

      final List<Integer> bucketIds = new ArrayList<>(indexesByBucket.keySet());
      bucketIds.sort(Integer::compareTo);
      for (int i = 0; i < bucketIds.size(); i++)
        workerByBucket.put(bucketIds.get(i), i % workerCount);

      final AtomicInteger threadNumber = new AtomicInteger();
      executor = Executors.newFixedThreadPool(workerCount, task -> {
        final Thread thread = new Thread(task, "arcadedb-index-bucket-writer-" + threadNumber.incrementAndGet());
        thread.setDaemon(true);
        return thread;
      });
      try {
        for (int i = 0; i < workerCount; i++) {
          final ArrayBlockingQueue<WriteBatch> queue = new ArrayBlockingQueue<>(WRITER_QUEUE_CAPACITY);
          queues.add(queue);
          pendingBatches.add(new WriteBatch(false));
          futures.add(executor.submit(() -> runWorker(queue)));
        }
      } catch (final Throwable t) {
        executor.shutdownNow();
        throw t;
      }
    }

[justinblethrow-cloud]

Fixed in bc3030e. Partial worker initialization now shuts down the executor before the constructor propagates the failure. The 19-test concurrency/failure gate and 42-test broad sorted-build gate both pass.

[gemini-code-assist]

In complete(), calling future.get() directly propagates ExecutionException if a worker thread fails. This wraps the actual cause (e.g., IOException or IndexException) in an ExecutionException, which is then wrapped in another IndexException by the caller, leading to a poor developer experience. Catching ExecutionException and calling checkFailure() allows the actual unwrapped cause to be thrown directly.

      executor.shutdown();
      for (final Future<?> future : futures) {
        try {
          future.get();
        } catch (final ExecutionException error) {
          checkFailure();
          throw error;
        } catch (final InterruptedException error) {
          Thread.currentThread().interrupt();
          checkFailure();
          throw error;
        }
      }
      checkFailure();

[justinblethrow-cloud]

Fixed in bc3030e. complete() now handles ExecutionException through the recorded worker cause and restores the interrupt flag before propagating InterruptedException. Both submission gates remain green.

[lvca]

@claude please review this

[claude]

Claude finished @lvca's task in 3m 32s —— View job

---

Review

• Read trigger comment and gather PR context
• Review LSMTreeIndexBulkLoader.java changes (parallel bucket writers)
• Review LSMTreeIndexExternalSorter.java changes (parallel merge groups)
• Review SortedIndexBuildMetrics.java and TypeIndexBuilder.java
• Review new/updated tests
• Post final review feedback

I diffed 9ca99e3^..HEAD (the branch point off main), which matches the PR's stated stat exactly (9 files, +1815/-58).

Previously flagged Gemini issues

Both are correctly resolved in bc3030e:

• Executor leak in BucketWriteDispatcher constructor (LSMTreeIndexBulkLoader.java:498-509): the worker-init loop is now wrapped in try { ... } catch (RuntimeException | Error error) { executor.shutdownNow(); throw error; }. Since none of ArrayBlockingQueue's constructor, WriteBatch's constructor, or executor.submit(...) declare checked exceptions, RuntimeException | Error is a complete catch here — no narrower Throwable catch is needed.
• ExecutionException unwrapping in complete() (LSMTreeIndexBulkLoader.java:526-552): on ExecutionException, the code now records the cause via failure.compareAndSet(null, error.getCause()) and calls checkFailure(), which rethrows the real root cause (Exception/Error/wrapped IndexException) instead of the ExecutionException wrapper.

Minor nit (not blocking)

LSMTreeIndexBulkLoader.java:538-541:

} catch (final ExecutionException error) {
  failure.compareAndSet(null, error.getCause());
  checkFailure();
  throw error;
}

Once failure.compareAndSet(...) runs, failure.get() is guaranteed non-null (either just set, or already set by a peer worker/failure), so checkFailure() always throws here and the trailing throw error; is unreachable dead code. Harmless, but worth deleting for clarity — e.g. checkFailure(); alone, since it always exits via exception in this branch.
Fix this →

Merge-group parallelism (LSMTreeIndexExternalSorter.executeMergePlans)

• Failure handling looks correct: on any worker exception, all Futures are cancelled, the executor is force-shut-down, and the first cause is preserved and rethrown unwrapped as IOException/RuntimeException/Error. Interrupt status is restored via Thread.currentThread().interrupt() before rethrow.
• Source runs for a failed generation are untouched (deletion of sources only happens after all MergeResults in results succeed in consolidateRuns()), matching the PR description.
• Output files for successfully completed groups within a failed generation aren't explicitly deleted at throw time — only the failing worker's own partial output is removed in its finally. This is fine because the whole spill directory is wiped by LSMTreeIndexExternalSorter.close() regardless of success/failure, so nothing leaks.
• File-descriptor/CPU/memory admission math (selectMergeParallelism, selectWriterParallelism) is self-consistent with how many concurrent file handles/buffers each worker actually holds.

Tests

• New parallelism suites (LSMTreeSortedBucketParallelismTest, LSMTreeSortedMergeParallelismTest, SortedIndexBuildMetricsTest) look thorough per the PR's stated 19+42 test gates.
• TestSortedIndexParallelismBenchmark is correctly tagged @Tag("benchmark") so it's excluded from regular CI, consistent with the existing performance/TestSortedIndexBuildBenchmark naming precedent.
• One gap: there's no test exercising the constructor's partial-init failure path directly (the leak Gemini flagged) — understandably hard to trigger without mocking ExecutorService/injecting RejectedExecutionException. Not blocking given the fix is straightforward and mechanically verifiable by inspection.

Overall

The design (bounded per-stage admission, serial global stream owning ordering/uniqueness/publication, cancel-and-cleanup on first failure) is sound and matches the described scope. I didn't find new correctness issues beyond the dead-code nit above. Nice, well-scoped piece of work.

[justinblethrow-cloud]

Addressed the non-blocking exception-path nit in afce2494: a null ExecutionException cause now has an explicit fallback, and the normal non-null cause is recorded and rethrown without a redundant trailing throw. Re-ran the 19-test concurrency/failure/metrics gate and the 42-test broad sorted-build gate; both pass.

[codecov]

Codecov Report

❌ Patch coverage is 83.22440% with 77 lines in your changes missing coverage. Please review.
✅ Project coverage is 66.27%. Comparing base (e15c88a) to head (afce249).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...arcadedb/index/lsm/LSMTreeIndexExternalSorter.java	75.90%	29 Missing and 11 partials ⚠️
...com/arcadedb/index/lsm/LSMTreeIndexBulkLoader.java	81.21%	27 Missing and 10 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5259      +/-   ##
============================================
+ Coverage     65.45%   66.27%   +0.81%     
============================================
  Files          1703     1660      -43     
  Lines        139116   133277    -5839     
  Branches      29816    28876     -940     
============================================
- Hits          91064    88327    -2737     
+ Misses        35663    32851    -2812     
+ Partials      12389    12099     -290     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lvca]

Another great PR, thanks @justinblethrow-cloud !