Make broker routing components update incrementally on assignment changes - #19783

PR flow

Broker routing components now update incrementally, recomputing only when inputs change and sharing partition functions/sets.

flowchart TD
  N0["SegmentPartitionMetadataManager#46;onAssignmentChange #40;F1#41;"]:::stModified
  N1["Detect changes #40;partition id#47;creation time#44; online servers#44; segment removal#41; #40;F1#41;"]:::stModified
  N2["ComputeTablePartitionInfo #40;F1#41;"]:::stModified
  N3["ComputeTablePartitionReplicatedServersInfo #40;F1#41;"]:::stModified
  N4["TimeBoundaryManager#46;onAssignmentChange #40;F3#41;"]:::stModified
  N5["Batch read ZK metadata for new segments #40;F3#41;"]:::stModified
  N6["Update end time map and max end time incrementally #40;F3#41;"]:::stModified
  N7["UpdateTimeBoundaryInfo #40;F3#41;"]:::stModified
  N8["SegmentPartitionUtils#46;extractPartitionInfo #40;F2#41;"]:::stModified
  N9["getPartitionFunction #40;cache lookup#47;creation#41; #40;F2#41;"]:::stModified
  N10["getPartitions #40;shared singleton#41; #40;F2#41;"]:::stModified
  N0 -->|"triggers detection"| N1
  N1 -->|"if partitions changed"| N2
  N1 -->|"if online servers changed or expiration"| N3
  N4 -->|"initiates batch ZK read"| N5
  N5 -->|"updates end times and max"| N6
  N6 -->|"updates time boundary info"| N7
  N8 -->|"gets partition function"| N9
  N8 -->|"gets partition set"| N10
  classDef stAdded fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
  classDef stModified fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
  classDef stRemoved fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
  classDef stUnchanged fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px

AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing

• Regenerate PR flow

Summary

Part of #19782.

This PR makes three broker routing components do less work on each ideal state or external view (IS/EV) change. SegmentPartitionMetadataManager recomputes its published partition info only when its inputs change. TimeBoundaryManager reads the ZK metadata of new segments in one batch and keeps the max end time up to date incrementally. SegmentPartitionUtils shares one PartitionFunction and one immutable one-element partition set across segments with equal metadata.

Each component compares the new inputs with its writer-private state, and builds a new published object only when the result can be different. Most of the cost in #19782 comes from the IS/EV parse and the instance selector. These three components are a small part of it.

Changes

1. SegmentPartitionMetadataManager: recompute only when inputs change

Before, onAssignmentChange() did two full passes over all segments on each change. Now it finds what changed and recomputes only what depends on it:

• A pulled segment counts as a change only when it is new, or its partition id or creation time is different. Segments without ZK metadata are pulled on each change, so without this check every change counts.
• hasSameOnlineServers() compares without allocation. Equal lists of online servers are shared through a per-call cache.

New-segment expiry. TablePartitionReplicatedServersInfo treats a segment as "new" for a configured time after its creation. Before, every change recomputed the object and so evaluated the clock. Now the compute step stores the earliest expiration time of the new segments in _nextNewSegmentExpirationTimeMs (Long.MAX_VALUE if none). If the current time is past it, the next change recomputes the object. Thus a new segment still becomes visible on the first change after it expires.

2. TimeBoundaryManager: batched ZK reads and incremental max end time

• One ZkHelixPropertyStore.get(List, ...) call reads the ZK metadata of all new ONLINE segments, as init() and SegmentZkMetadataFetcher already do. It returns null for a missing node and for a failed read. Then the manager does not record the segment, and the next change reads it again.
• Trade-off: a segment whose ZK metadata is missing for good is read again on every change, with a WARN log each time. Before this PR, the exception path also read it again on every change.
• The writer-private _maxEndTimeMs holds the max of _endTimeMsMap. An added segment updates it with Math.max(). A removed segment causes a full scan only when it held the max. refreshSegment() causes a full scan only when the segment held the max and its end time decreased.
• updateTimeBoundaryInfo() still runs on each change. Thus a change of the explicit time boundary in the ideal state still takes effect.

3. SegmentPartitionUtils: share partition functions and partition sets

• One PartitionFunction is shared per (function name, number of partitions, function config) through a static ConcurrentHashMap. The cache holds at most 1024 entries. Past that limit, functions are created and not cached.
• A one-element partition set becomes a shared immutable IntSets.singleton() for partition ids less than 1024. All callers only read the partition sets.
• The PartitionFunction contract already requires stateless implementations that are safe for concurrent use. The query path already calls one instance from many threads.
• SinglePartitionColumnSegmentPruner compares functions with equals() and keys a HashMap by function. Shared instances only help there through the usual this == o fast path.

Concurrency

This PR keeps the concurrency rule of #19782.

• Published: TablePartitionInfo, TablePartitionReplicatedServersInfo and TimeBoundaryInfo, swapped with one volatile write as before. When nothing changes, the old object stays published.
• Writer-private: _segmentInfoMap, _nextNewSegmentExpirationTimeMs, _endTimeMsMap and _maxEndTimeMs. synchronized methods serialize the writers, as before.
• Shared: the lists of online servers are unmodifiable, and the recompute code copies them before any change. Partition functions and sets are stateless or immutable.

Tests

• SegmentPartitionMetadataManagerTest.testRecomputeOnlyOnChange: equal inputs keep both objects, an online server change replaces only TablePartitionReplicatedServersInfo, and a removed segment replaces both. ONLINE to CONSUMING is not a change. A segment without ZK metadata does not cause a recompute on each change.
• SegmentPartitionMetadataManagerTest.testNewSegmentExpiresOnNoOpChange: a new segment that misses a replica is excluded. After it expires, a no-op change includes it. After that, a no-op change keeps the published object.
• TimeBoundaryManagerTest.testBatchedReadsAndIncrementalMaxEndTime: two added segments cause one batched read. A change with no new segment does no ZK read. A removal or a refresh moves the time boundary back only when it affects the max end time. A new segment without ZK metadata is read again on the next change.
• SegmentPartitionUtilsTest (new): sharing for equal metadata, distinct functions for a different name, number of partitions or config, and multi-partition sets kept as they are.

All tests under org.apache.pinot.broker.routing pass (19 classes, 257 tests, 0 failures, 0 skipped).

Measured effect

TimeBoundaryManager runs only for the offline side of a hybrid table. We replayed the segments of a real 380k-segment table as the OFFLINE side of a hybrid table. Per change, TimeBoundaryManager went from 22 MB and about 440 ms to 14 MB and about 40 ms.

Compatibility

There is no change to configs, wire formats, or public APIs. All changed code is broker-internal.