gortiz opened a new pull request, #19783: URL: https://github.com/apache/pinot/pull/19783
## 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: | Change | `TablePartitionInfo` | `TablePartitionReplicatedServersInfo` | |---|---|---| | A segment is added or removed, or a partition id or creation time changes | recomputed | recomputed | | Only the online servers of a segment change | kept | recomputed | | A new segment stops being new (see below) | kept | recomputed | | Nothing relevant changes | kept | kept | - 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. -- 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]
