PHOENIX-7931 Coalesce per-batch mutations into a single record#2540
Merged
tkhurana merged 3 commits intoJul 2, 2026
Merged
Conversation
Switch the replication log framing to a cell-oriented record (List<Cell> + record-level attributes) and coalesce IndexRegionObserver.replicateMutations to emit one record per (data table, batch) and one record per index target table per batch, instead of one record per Mutation. The consumer reconstructs Put/Delete mutations on the row+type boundary via MutationCellGrouper, so coprocessor-merged cells (local index, conditional TTL, ON DUPLICATE KEY UPDATE) survive the framing change. Drops the Mutation overload from LogFile.Writer (production no longer needs it) and rewrites the test layer to assert against cell lists.
tkhurana
changed the title
apurtell
approved these changes
Jun 30, 2026
| * Default Codec for encoding and decoding ReplicationLog Records within a block buffer. This | ||
| * implementation uses standard Java DataInput/DataOutput for serialization. Record Format within a | ||
| * block: | ||
| * Default Codec for encoding and decoding ReplicationLog Records within a block buffer. The on-disk |
Contributor
There was a problem hiding this comment.
This is the most interesting change (to me). I suppose it's of a piece with the evolution of HBase WAL with WALEdit to group cells and on the RPC side "cell blocks" instead of individually described cells in the IDL.
It was probably inevitable that this grouping was necessary to bring down overheads but I wanted to go with simple and atomic first if we could get away with it.
We may want to restructure the ascii art here further
RECORD LENGTH (vint)
RECORD HEADER
Table name length (vint)
Table name bytes
Commit id (vlong)
ATTRIBUTES COUNT (vint) -- NEW: record-level attribute map
PER-ATTRIBUTE (repeated):
Key length (vint)
Key bytes (UTF-8)
Value length (vint)
Value bytes
CELL COUNT (vint)
PER-CELL (repeated):
Row length (vint) -- row written on EVERY cell
Row bytes
Family length (vint) -- family written on EVERY cell
Family bytes
Qualifier length (vint)
Qualifier bytes
Cell timestamp (long, 8 bytes)
Cell type byte (1 byte)
Value length (vint)
Value bytes
| currentBatchSize++; | ||
| currentBatchSizeBytes += mutation.heapSize(); | ||
|
|
||
| // Process when we reach either the batch count or size limit |
Contributor
There was a problem hiding this comment.
Before this change a record was always a single Mutation, so batches always split between records. Now a single record's mutations can be split across two processReplicationLogBatch invocations. I think it's fine but future maintainers cannot assume a certain atomicity here.
| * before the action — the broken writer is never touched. No replay needed (empty batch). | ||
| */ | ||
| @Test | ||
| public void testIdleLeaseRecoveryDrainsStagedWriter() throws Exception { |
Contributor
There was a problem hiding this comment.
I think this test is a real concurrency concern. After a sync clears currentBatch, the system goes idle, a rotation tick stages a pending writer, invalidating the old writer. When events resume we need to ensure apply() drains to the new and healthy writer before the action and should ensure the dead writer is never touched.
The body of this test only needs some simple updates (any(Mutation.class) → any(List.class), eq(put) → eq(LogFileTestUtil.cellsOf(put))) Let's restore and migrate it.
There was a problem hiding this comment.
Pull request overview
This PR (PHOENIX-7931) updates Phoenix replication logging to coalesce a server-side batch’s per-table mutations into a single log record carrying a flat cell stream, and adds replay-side logic to reconstruct Put/Delete mutations by grouping on row + cell-type boundaries. This reduces replication ring-buffer pressure and updates the on-disk record format and related APIs to be cell-oriented.
Changes:
- Switch replication log record format and writer APIs from
Mutation-oriented toList<Cell>-oriented, including codec encode/decode updates. - Introduce
MutationCellGrouperto regroup a flat cell stream into mutations during replay, and wire it into the replication reader path. - Update unit/integration tests and add new assertions to pin record-count and regrouping behavior.
Reviewed changes
Copilot reviewed 19 out of 19 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java | Updates mocks/verifications for cell-list appends; adds deterministic sync-metrics unit test and per-batch framing simulator support. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/MutationCellGrouperTest.java | New unit tests pinning the row+type regrouping behavior used during replay. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterTest.java | Updates writer tests to append records by cell stream instead of mutation. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileWriterSyncTest.java | Updates sync tests to use flattened cell lists for append. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileTestUtil.java | Adds mutation→cells flattener and updates equality helpers to compare cell streams/attributes. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCompressionTest.java | Updates compression tests to append cell streams. |
| phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java | Expands round-trip coverage for the new format; rejects empty records; adds opt-in framing microbenchmark. |
| phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java | Adds record-count contract IT and shifts metrics assertions out of flaky IT path. |
| phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java | Updates ITs to append using cell streams. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/tool/LogFileAnalyzer.java | Adds per-table record counting and adapts analysis to multi-mutation records. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java | Adds cell-list append API and changes ring-buffer payload to carry flattened cells. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java | Updates replay/append plumbing to pass cell streams into the writer and persist them in batches. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java | Replays multi-mutation records by iterating reconstructed mutations per record. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java | New regrouping utility implementing row+cell-type boundary reconstruction. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileWriter.java | Updates writer append signature to accept List<Cell> and build cell-oriented records. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileRecord.java | Changes record body to store cells + attributes, and reconstruct mutations via MutationCellGrouper. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java | Implements new cell-oriented on-disk format with per-record attributes and per-cell row/family/qualifier fields. |
| phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFile.java | Updates Record/Writer interfaces for cell-stream bodies and multi-mutation reconstruction. |
| phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java | Coalesces per-(table,batch) writes into a single append using flat cell streams; removes per-mutation splitting on write path. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+576
to
+581
| public void append(String tableName, long commitId, List<Cell> cells) throws IOException { | ||
| if (LOG.isTraceEnabled()) { | ||
| LOG.trace("Append: table={}, commitId={}, cells={}", tableName, commitId, cells.size()); | ||
| } | ||
| publishDataEvent(new Record(tableName, commitId, cells)); | ||
| } |
Comment on lines
555
to
+564
| public void append(String tableName, long commitId, Mutation mutation) throws IOException { | ||
| if (LOG.isTraceEnabled()) { | ||
| LOG.trace("Append: table={}, commitId={}, mutation={}", tableName, commitId, mutation); | ||
| } | ||
| List<Cell> cells = new ArrayList<>(); | ||
| for (List<Cell> familyCells : mutation.getFamilyCellMap().values()) { | ||
| cells.addAll(familyCells); | ||
| } | ||
| publishDataEvent(new Record(tableName, commitId, cells)); | ||
| } |
Comment on lines
+30
to
+34
| * Groups a flat cell stream into Put/Delete mutations, mirroring the algorithm HBase's | ||
| * ReplicationSink uses to reconstruct mutations from a WALEdit. A new mutation is started whenever | ||
| * the row key or the put-vs-delete disposition differs from the immediately preceding cell; | ||
| * consecutive cells sharing both are collected into one mutation. There is no precondition on the | ||
| * input ordering: any cell stream produces valid mutations. Ordering only affects how the cells are |
…re lease-recovery test Review feedback on PR apache#2540: - Fail fast on empty/null cell lists in both ReplicationLogGroup.append overloads so the failure is synchronous rather than poisoning the consumer thread later at codec-encode time. - Correct MutationCellGrouper Javadoc: the boundary is row + full cell type (delete subtypes split too), not just put-vs-delete. - Note in ReplicationLogProcessor that a record's reconstructed mutations may span batch boundaries; no per-record atomicity. - Restore testIdleLeaseRecoveryDrainsStagedWriter, migrated to the cell-list append API and made deterministic against the swap event by installing broken-stream stubs while idle and awaiting the old writer's async close before verifying its invocation counts.
testReadAfterMultipleRotations and its VaryingBatchSizes sibling captured the writer path via getWriter() BEFORE writing the batch. Since the swap event (PHOENIX-7864) drives writer swaps asynchronously on the consumer thread, and checkAndReplaceWriter's pendingWriter.getAndSet(null) and the currentWriter assignment are not atomic, getWriter() could observe the pending writer already taken by the consumer while currentWriter still pointed at the previous file. That mis-recorded logPaths -- one file's path duplicated, its successor's skipped -- so read-back compared records against the wrong file (e.g. index 700 read commitId 600). Capture the path AFTER sync(), once the batch's apply() has settled currentWriter to the file the records actually landed on and before this iteration's forceRotation stages the next writer. Verified: both tests 5/5 green.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PHOENIX-7931: Coalesce per-batch replication appends into a single record
Targets the
PHOENIX-7562-feature-newfeature branch (not master).What changed
On the write path, a server-side batch's mutations for a given table are now coalesced into a single replication log record carrying a flat cell stream, rather than one record per mutation. On the replay path,
MutationCellGrouperreconstructs Put/Delete mutations from that cell stream on the row + put-vs-delete boundary, mirroring the algorithm HBase'sReplicationSinkuses for WALEdits.This cuts ring-buffer pressure substantially (≈5× fewer Disruptor events at high batch sizes) without changing the fsync floor — coalescing already minimized inner syncs. The on-disk record format changes accordingly (per-cell row/family/qualifier/timestamp/type/value, with a per-record table name + commitId + attributes header).
Key files
IndexRegionObserver— per-(table, batch) cell aggregation before a singleappendMutationCellGrouper(new) — replay-side cell-stream → mutation regroupingLogFileCodec/LogFileRecord/LogFile— new coalesced record formatReplicationLogGroup/ReplicationLog— cell-oriented append APITesting
MutationCellGrouperTest(10 cases): empty input, single Put/Delete, contiguous same-row/type coalescing, row-change splits, Put→Delete same-row type-boundary split, three-row Put/Delete/Put, adjacent DeleteColumn/DeleteFamily subtype split, non-consecutive same-row partitioning, and cell preservation.testAppendAndSyncSingleBatchRecordCountIT pins the coalesced record-count contract with cross-cluster cell-level equality.testSyncMetricsEmittedunit test — deterministic replacement for the flaky IT metrics assertion (injects an fsync delay to clear ms-truncation; reads metrics only after a graceful close that joins the consumer).LogFileCodecTestround-trip coverage for the new format (multi-family ordering, empty-record rejection, all cell types).testReplicationSyncPathSimulator(opt-in) for sync-coalescing measurement under contention.Verification
phoenix-core-servercompiles clean;MutationCellGrouperTest,LogFileCodecTest, and the replication unit tests pass.mvn spotless:checkpasses on bothphoenix-core-serverandphoenix-core.