SAI-6231: Bucketed segment flush handling on only single doc iterator

[patsonluk]

Description

Built on top of this #40, with changes to only support delegating to a bucketed DocumentWriterPerThread (DWPT) if the input to DocumentWriter#updateDocuments is a single doc. This simplifies the logic flow

The bucket key will be the value of the input doc's field (defined by sysprop lucene.temporalField.name) mapped to the boundary of the bucket. The default boundaries are (in days) [-9, 3, 9, 32, 94, 184]. For example for -Dlucene.temporalField.name=EventStart,SessionStart,UserLastIndexedEventStart,UserTipLastEventStart, a doc with EventStart a day ago will be mapped to the bucket 3 (up to 3 days ago)

This is related to the TemporalMergePolicy work which group segments of the same bucket (using the same boundaries as defined in here) for merges.

Solution

Mostly work from @magibney from this PR

1. Logic to read sysprop lucene.temporalField.name and map input doc into bucket key
2. Changes to DocumentWriterFlushControl and DocumentWriterPerThreadPool to support creating/managing DWPT instances with a bucket key
3. PeekIterable to peek first (and only doc) to delegate to corresponding bucket DWPT
4. Change in MergePolicy to expose bucket mapping logic to higher layer (Solr TemporalMergePolicy etc)

Take note that we only support bucket mapping for DocumentWriter#updateDocuments with single doc for now. As supporting multiple document get complicated, as demonstrated in this original PR:

1. Need to delegate to multiple DWPTs while maintaining execute order across all the instances
2. Extra threads to support multiple DWPTs per updateDocuments processing

---

Note

Medium Risk
Changes the hot indexing path (DWPT acquisition and pooling) when routing is on, though behavior is opt-in via system properties and limited to single-document updates without a parent field.

Overview
Adds time-bucket routing at flush time so single-document updateDocuments paths can pin in-RAM segments to a bucket derived from a document’s temporal numeric field (primary + optional fallbacks), driven by JVM properties such as lucene.temporalField.name, lucene.temporalField.boundaries, and lucene.temporalField.adjustNow. Multi-document batches and indexes with a parent field skip routing and use the default bucket.

DocumentsWriter peeks the lone document (via PeekIterable) when routing is enabled, maps it with SegmentRoutingUtil, and calls obtainAndLock(bucket). DocumentsWriterPerThreadPool keeps separate free queues per bucket; each DocumentsWriterPerThread carries its bucket for checkout/return. MergePolicy.mapToBucket re-exports the boundary logic for Solr-style temporal merge policies.

A concurrent testUpdateDocumentRouting checks that live docs land in one temporal bucket per leaf after adds/updates; merge-policy reflection tests ignore the new static helper.

^{Reviewed by Cursor Bugbot for commit 5d28831. Bugbot is set up for automated code reviews on this repo. Configure here.}

[magibney]

Looking good I think!

I wonder if it'd be easy/worth pulling out as much as possible into its own class, as opposed to static fields/methods on DWPT (and PeekIterable from DocumentsWriter as well). As it stands it's kinda obvious where the changes are, but putting in a dedicated file would make it a bit clearer/cleaner?

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 39a3420. Configure here.}

[patsonluk]

Looking good I think!

I wonder if it'd be easy/worth pulling out as much as possible into its own class, as opposed to static fields/methods on DWPT (and PeekIterable from DocumentsWriter as well). As it stands it's kinda obvious where the changes are, but putting in a dedicated file would make it a bit clearer/cleaner?

Thanks! Did a quick refactor at here . For now keeping PeekIterable in where it is, cause the coupling is quite strong? 😁 Lemme know!

[magibney]

we shouldn't really need to repeateedly invoke System.currentTimeMillis(). Ideally we could set a TEMPORAL_ADJUST_NANOS based on diff between System.nanoTime() and configured sysprop lucene.temporalField.adjustNow?

[patsonluk]

are u thinking about optimization of calling System.nanoTime() instead of System.currentTimeMillis() all the time? like calculate the base diff at init time (vs System.currentTimeMillis() or TEMPORAL_ADJUST_NOW). Then later on we can just use base diff + System.nanoTime() (ie cheaper?)

[magibney]

exactly that, yes.

[patsonluk]

I have made the changes here, can u please 👀 ? e9809bc

It's slightly different for the handling of lucene.temporalField.adjustNow. In such case I think we don't even want the time to advance at all, since having a static now time will be the most consistent for testing.

[patsonluk]

ping @hiteshk25 before we merge (think it should be close to approval from @magibney), please let us know if there are any concerns. A TL;DR of the changes:

1. The new routing logic are all in new class SegmentRoutingUtil
2. Changes to existing classes (DocumentsWriter...) are mainly:
   1. Allow lookups of DocumentWriterPerThread using a "bucket key" which is mapped from the hard-coded day range boundaries
   2. The routing logic only get triggered if sysprop lucene.temporalField.name is defined. If undefined, it should run the old logic, using the "default" bucket key for everything and w/o any routing.

[hiteshk25]

👁️

[magibney]

I think this is not formally correct. System.nanoTime() can rollover and be negative. I think if we're going to do this, we have to convert fully to nanos -- e.g.

NOW_BASE_MILLIS = ADJUST_NOW == null ? System.currentTimeMillis() : ADJUST_NOW;
NOW_BASE_NANOS = System.nanoTime();

getNow() {
  return NOW_BASE_MILLIS + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - NOW_BASE_NANOS);
}

[patsonluk]

I was trying to save one arithmetic but good point on the wrap-around issue!

Im not going to care about ADJUST_NOW, since if it's defined, we will NOT do any computation and use the value directly

[magibney]

LGTM