gortiz opened a new pull request, #19786:
URL: https://github.com/apache/pinot/pull/19786
## Summary
Part of #19782.
This PR adds an opt-in `ZkSerializer`, `CompactZNRecordSerializer`, in
`pinot-common`. The broker installs it on its spectator Helix manager when
`pinot.broker.routing.compact.assignment.reader.enabled` is `true` (default
`false`). The reader streams the gzip znode into Jackson with field-name
canonicalization off, interns names, and shares equal instance state maps as
immutable sorted array maps. The routing manager then wraps the record without
a copy of the outer segment map.
## Design
```
spectator ZKHelixManager (flag on: CompactAssignmentZkHelixManager)
-> ChainedPathZkSerializer (longest prefix wins)
/<cluster>/IDEALSTATES/*, /<cluster>/EXTERNALVIEW/* (brokerResource
excluded)
-> CompactZNRecordSerializer
not gzip -> ZNRecordSerializer (unchanged)
gzip -> GZIPInputStream -> JsonParser (no canonicalization)
intern names, share equal maps, read to end (CRC-32
and size check)
parse error -> ZNRecordSerializer (fallback)
-> BaseBrokerRoutingManager: HelixHelper.toIdealState / toExternalView
compact record: share the maps, no TreeMap copy
other record: new IdealState(record), as before
```
**1. Reader and map (`org.apache.pinot.common.utils.helix`).**
- `CompactZNRecordSerializer` parses gzip znodes with the compact parser and
sends plain znodes to `ZNRecordSerializer`. `serialize()` always delegates to
`ZNRecordSerializer`.
- `ImmutableSortedArrayMap` is an immutable `SortedMap` in natural `String`
order, with a sorted key array and a parallel value array. `comparator()`
returns `null`, so a `TreeMap` copy takes the linear `buildFromSorted` path.
`equals` and `hashCode` match `TreeMap`. Every mutator throws
`UnsupportedOperationException`.
- Simple fields and list fields stay mutable `LinkedHashMap`s.
- `HelixHelper.isCompact`, `toIdealState` and `toExternalView` sit next to
`cloneIdealState` (see #19738). They share the simple, list and map fields and
the raw payload. They copy the version, the creation time, the modified time
and the ephemeral owner, and set the `HelixProperty.Stat`, as
`HelixProperty(ZNRecord)` does. The two PRs are independent. The one that
merges second rebases on the other.
**2. No-copy wrap.** `BaseBrokerRoutingManager.getIdealState` and
`getExternalView` use these helpers. A record from `ZNRecordSerializer` still
goes through `new IdealState(record)` and `new ExternalView(record)`.
**3. Wiring and flag.** `BaseBrokerStarter` and
`MultiClusterHelixBrokerStarter` create their spectator managers through
`CompactAssignmentZkHelixManager.createSpectatorHelixManager`. With the flag
off, it returns `HelixManagerFactory.getZKHelixManager(...)`, so the code path
does not change. With the flag on, the `ZKHelixManager` subclass overrides
`connect()` and, after `super.connect()`, replaces the client serializer with a
`ChainedPathZkSerializer`.
## Safety
- **Default off.** The flag takes effect on broker start.
- **Fallback.** If the compact parser fails, the serializer logs a warning
and returns the result of `ZNRecordSerializer`.
- **brokerResource excluded.** It is small, and its readers are outside of
routing (the query quota manager and the broker resource state model).
- **Writes.** Writes through the spectator client use `ZNRecordSerializer`.
The participant manager has its own client with the default serializer.
**Helix internals.** `CompactAssignmentZkHelixManager` reads the protected
`_zkclient` field. Helix assigns it only in `createClient()`, which only
`connect()` calls. The serializer survives a session expiry, because Helix
keeps the client. After `disconnect()` and `connect()`, the subclass installs
it again. Tests verify both. If Helix renames the field, the build fails. If
Helix creates the client outside of `connect()`, the tests fail. The client
serializer field is not `volatile`. The broker installs it on the `connect()`
thread, before it registers any listener or starts any routing task. The thread
start and the queue handoffs publish it.
**Readers of the table ideal state and external view (IS/EV) paths through
the spectator client:**
| Reader | What it does with the record |
|---|---|
| `BaseBrokerRoutingManager` (and `RemoteClusterBrokerRoutingManager`) |
Wraps it and passes it to the routing components, which only read the maps. |
| Helix `CallbackHandler` | On INIT, FINALIZE and child changes, reads each
child to subscribe. Uses only the bucket size. Pre-fetch is off for
`ClusterChangeMediator`. |
| Custom `ClusterChangeHandler`s from the broker config | Get the spectator
manager in `init()`. A table IS read through it has immutable instance state
maps. |
| `HelixAdmin` / `HelixDataAccessor` of the spectator manager | No broker
code reads table IS/EV through them. Their `brokerResource` reads use
`ZNRecordSerializer`. |
| `ZKHelixDataAccessor` bucketized merge | Pinot does not use bucketized
IS/EV. |
No broker code mutates the map fields of a table IS/EV. A future
read-modify-write of a table ideal state through this manager fails with
`UnsupportedOperationException`. The Javadoc of
`CompactAssignmentZkHelixManager` states this.
**Parser fidelity.** The compact parser returns a `ZNRecord` equal to the
result of `ZNRecordSerializer`. It matches these cases: missing and `null`
fields, unknown fields (skipped), duplicate keys (last wins), numbers and
booleans in string maps (text kept), JSON escapes, a base64 `rawPayload`, and
content after the record (ignored, as `ObjectMapper.readValue` does). A bad
gzip trailer (CRC-32 or size) and malformed JSON fall back.
**Interned names.** Segment names, instance names and unknown state values
use `String.intern()`. The routing components already keep these names, so
interning adds no new long-lived strings. The JVM string table drops an
interned string when nothing refers to it. The per-parse dictionary for
instance names lives for one parse only.
## Gzip-only rule
Pinot writes a table ideal state with gzip when it has more than 1000
segments, or when its estimated size is large. Helix copies `enableCompression`
into the external view. So plain znodes are the small ones. A size sweep
compared the two parsers:
| Znode | Allocation (compact / Helix) | CPU (compact / Helix) |
|---|---|---|
| Gzip, as Pinot stores it (more than 1000 segments) | 0.26x to 0.34x |
0.62x to 0.76x |
| Plain, up to about 200 segments | 1x to 3.5x | 1x to 1.5x |
Thus plain znodes go to `ZNRecordSerializer`. There is no size threshold and
no other config.
## Tests
- `CompactZNRecordSerializerTest`: equality with `ZNRecordSerializer` for
plain and gzip input (empty and missing fields, escapes, content after the
record, random records), interning and map sharing, the fallback for malformed
input and a bad gzip trailer, the gzip-only rule, and the `HelixHelper` wrap
(map identity, scalar fields, stat, equality with the copy constructors).
- `ImmutableSortedArrayMapTest`: the `SortedMap` contract, views,
`equals`/`hashCode` against `TreeMap`, and the mutators.
- `CompactAssignmentInstanceSelectorTest`: the balanced, replica-group and
strict replica-group (upsert) selectors select the same servers for compact and
regular records.
- `CompactAssignmentZkHelixManagerTest` (local ZooKeeper): the flag, compact
reads for table records and default reads for `brokerResource` (also through
the data accessor and the admin), a disconnect and connect cycle, a session
expiry, and a check of the `_zkclient` field.
- `CompactAssignmentRoutingManagerTest` (local ZooKeeper): a hybrid,
partitioned table with two `BrokerRoutingManager`s, one on the compact manager
and one on the default manager. They return the same routing tables, pruning,
time boundary and partition info, before and after an assignment change.
- All tests in `org.apache.pinot.broker.routing` and
`org.apache.pinot.broker.broker.helix` pass.
## Measured effect
Replay of the ideal state and external view changes of a real 380k-segment
realtime table (JDK 25, G1, 8 MB regions, mean of 8 changes):
| Per change | Before | After |
|---|---|---|
| Parse of the ideal state and the external view, allocation | 1,807 MB |
293 MB |
| Humongous regions | about 120 | 0 |
| Ideal state and external view kept in memory during the rebuild | 311 MB |
6 MB |
For a 69k-segment table, one ideal state parse allocates 30 MB instead of
979 MB.
--
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]
