This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 3cd652d8ab PHOENIX-7958 Reduce per-record and per-block copies in
replication log writer (#2572)
3cd652d8ab is described below
commit 3cd652d8abe637baf77d8100d684df7e18ff6892
Author: tkhurana <[email protected]>
AuthorDate: Wed Jul 15 12:51:16 2026 -0700
PHOENIX-7958 Reduce per-record and per-block copies in replication log
writer (#2572)
Switch the record encoder and block writer from
java.io.ByteArrayOutputStream
to HBase's ByteArrayOutputStream and write directly from the backing buffer
via getBuffer()/size() instead of toByteArray(). This eliminates the
per-record body copy in LogFileCodec and the per-block copy in
LogFileFormatWriter (the latter previously copied the whole ~1MB block even
in the uncompressed path). Also hoist the per-record DataOutputStream
wrapper
to a field.
Add an opt-in allocation A/B microbenchmark (testAllocationMicrobenchmark)
that drives the full append path and measures heap bytes per record via
ThreadMXBean. Measured -22.5% heap/record (3583.5 -> 2776.1 B).
---
.../phoenix/replication/log/LogFileCodec.java | 18 ++--
.../replication/log/LogFileFormatWriter.java | 17 ++--
.../phoenix/replication/log/LogFileCodecTest.java | 112 +++++++++++++++++++++
3 files changed, 133 insertions(+), 14 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java
index 0f25aa23fb..4e1f4fec93 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileCodec.java
@@ -17,7 +17,6 @@
*/
package org.apache.phoenix.replication.log;
-import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
@@ -32,6 +31,7 @@ import java.util.List;
import java.util.Map;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
import org.apache.hadoop.hbase.io.ByteBuffInputStream;
import org.apache.hadoop.hbase.nio.ByteBuff;
import org.apache.hadoop.hbase.util.Bytes;
@@ -106,11 +106,16 @@ public class LogFileCodec implements LogFile.Codec {
private static class RecordEncoder implements LogFile.Codec.Encoder {
private final DataOutput out;
+ // Scratch buffer that accumulates one record's body so we can
length-prefix it. Reused
+ // across records; HBase's ByteArrayOutputStream lets us write straight
from its backing
+ // array via getBuffer(), avoiding a per-record toByteArray() copy.
private final ByteArrayOutputStream currentRecord;
+ private final DataOutputStream recordOut;
RecordEncoder(DataOutput out) {
this.out = out;
currentRecord = new ByteArrayOutputStream();
+ recordOut = new DataOutputStream(currentRecord);
}
@Override
@@ -118,7 +123,6 @@ public class LogFileCodec implements LogFile.Codec {
if (record.getCells().isEmpty()) {
throw new IllegalArgumentException("Cannot encode a record with no
cells");
}
- DataOutput recordOut = new DataOutputStream(currentRecord);
// Header: table name + commit id
byte[] nameBytes =
record.getHBaseTableName().getBytes(StandardCharsets.UTF_8);
@@ -161,12 +165,12 @@ public class LogFileCodec implements LogFile.Codec {
}
}
- byte[] currentRecordBytes = currentRecord.toByteArray();
- WritableUtils.writeVInt(out, currentRecordBytes.length);
- out.write(currentRecordBytes);
+ int recordBodyLen = currentRecord.size();
+ WritableUtils.writeVInt(out, recordBodyLen);
+ out.write(currentRecord.getBuffer(), 0, recordBodyLen);
- ((LogFileRecord) record).setSerializedLength(
- currentRecordBytes.length +
WritableUtils.getVIntSize(currentRecordBytes.length));
+ ((LogFileRecord) record)
+ .setSerializedLength(recordBodyLen +
WritableUtils.getVIntSize(recordBodyLen));
currentRecord.reset();
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
index ead68e0ba8..942e9fa9f3 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
@@ -17,11 +17,11 @@
*/
package org.apache.phoenix.replication.log;
-import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.io.compress.Compressor;
import org.slf4j.Logger;
@@ -109,7 +109,10 @@ public class LogFileFormatWriter implements Closeable {
return; // No active block or block is empty
}
blockDataStream.flush(); // Ensure all encoded records are in the byte
array
- byte[] uncompressedBytes = currentBlockBytes.toByteArray();
+ // Read straight from the buffer's backing array to avoid a whole-block
copy. Only the first
+ // uncompressedLen bytes are valid; the array itself may be larger.
+ byte[] uncompressedBytes = currentBlockBytes.getBuffer();
+ int uncompressedLen = currentBlockBytes.size();
ByteBuffer writeBuf;
int writeLen;
Compression.Algorithm ourCompression = context.getCompression();
@@ -117,12 +120,12 @@ public class LogFileFormatWriter implements Closeable {
Compressor compressor = ourCompression.getCompressor();
try {
compressor.reset();
- compressor.setInput(uncompressedBytes, 0, uncompressedBytes.length);
+ compressor.setInput(uncompressedBytes, 0, uncompressedLen);
compressor.finish(); // We are going to one-shot this.
// Give 25% overhead for pathological cases
// We can't go below this by much because the Snappy compressor will
require more
// than 20% overhead in some cases or else it will refuse to try.
- int compressBuffNeeded = (int) (uncompressedBytes.length * 1.25f);
+ int compressBuffNeeded = (int) (uncompressedLen * 1.25f);
if (compressBuf == null || compressBuf.capacity() <
compressBuffNeeded) {
compressBuf = ByteBuffer.allocate(compressBuffNeeded);
}
@@ -137,13 +140,13 @@ public class LogFileFormatWriter implements Closeable {
context.getCompression().returnCompressor(compressor);
}
} else {
- writeBuf = ByteBuffer.wrap(uncompressedBytes);
- writeLen = uncompressedBytes.length;
+ writeBuf = ByteBuffer.wrap(uncompressedBytes, 0, uncompressedLen);
+ writeLen = uncompressedLen;
}
// Write block header
LogFile.BlockHeader blockHeader = new
LogBlockHeader().setDataCompression(ourCompression)
-
.setUncompressedDataSize(uncompressedBytes.length).setCompressedDataSize(writeLen);
+
.setUncompressedDataSize(uncompressedLen).setCompressedDataSize(writeLen);
blockHeader.write(output);
output.write(writeBuf.array(), writeBuf.arrayOffset(), writeLen);
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
index debe855a4d..98e70aad86 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellBuilderFactory;
import org.apache.hadoop.hbase.CellBuilderType;
import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Put;
@@ -306,6 +307,45 @@ public class LogFileCodecTest {
new
LogFileRecord().setHBaseTableName("TBLLVAL").setCommitId(1L).setMutation(put));
}
+ @Test
+ public void testCodecLargeRecordThenSmallReusesBufferCleanly() throws
IOException {
+ // The encoder reuses one scratch buffer across records and writes its
backing array via
+ // getBuffer() bounded by size(). A large record grows the buffer; the
following small record
+ // must not carry a stale tail from the large one. Encoding both through
the SAME encoder and
+ // asserting both round-trip (with no extra records) pins that invariant:
writing more than
+ // size() bytes would misframe the stream after the small record.
+ long ts = 12345L;
+ byte[] largeValue = new byte[128 * 1024];
+ Arrays.fill(largeValue, (byte) 0xAB);
+ Put largePut = new Put(Bytes.toBytes("largeRow"));
+ largePut.setTimestamp(ts);
+ largePut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q"), ts,
largeValue);
+ LogFile.Record large =
+ new
LogFileRecord().setHBaseTableName("TBLREUSE").setCommitId(1L).setMutation(largePut);
+
+ Put smallPut = new Put(Bytes.toBytes("smallRow"));
+ smallPut.setTimestamp(ts);
+ smallPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q"), ts,
Bytes.toBytes("v"));
+ LogFile.Record small =
+ new
LogFileRecord().setHBaseTableName("TBLREUSE").setCommitId(2L).setMutation(smallPut);
+
+ LogFileCodec codec = new LogFileCodec();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos);
+ LogFile.Codec.Encoder encoder = codec.getEncoder(dos);
+ encoder.write(large);
+ encoder.write(small);
+ dos.close();
+
+ ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
+ LogFile.Codec.Decoder decoder = codec.getDecoder(new
DataInputStream(bais));
+ assertTrue("Should decode the large record", decoder.advance());
+ LogFileTestUtil.assertRecordEquals("Large record should match", large,
decoder.current());
+ assertTrue("Should decode the small record", decoder.advance());
+ LogFileTestUtil.assertRecordEquals("Small record should match", small,
decoder.current());
+ assertFalse("Should be no trailing garbage after the small record",
decoder.advance());
+ }
+
// Cell timestamp preservation tests
// These verify that per-cell timestamps survive a codec round-trip when
they differ from the
// mutation-level timestamp. Before the fix the encoder omitted
cell.getTimestamp() entirely
@@ -580,6 +620,78 @@ public class LogFileCodecTest {
mutationCount > 1 ? (a.wireBytes - b.wireBytes) / (mutationCount - 1) :
0);
}
+ /**
+ * Allocation A/B microbenchmark. Drives the full {@link
LogFileFormatWriter#append} path (record
+ * framing in the codec plus block sealing in the format writer) and
measures heap bytes allocated
+ * by the writer thread, using the JVM's per-thread allocation counter. This
isolates the copy
+ * churn targeted by the buffer-reuse change: the per-record and per-block
buffer copies that
+ * change replaced with {@code getBuffer()} reads. Output is
bytes-allocated-per-record; run this
+ * on the baseline and the candidate (e.g. via {@code git stash}) and
compare the two numbers.
+ * <p>
+ * Opt in with {@code -Dtest.runAllocationBenchmark=true}; tune with {@code
-Dtest.mutationCount},
+ * {@code -Dtest.cellsPerMutation}, {@code -Dtest.valueSize}, {@code
-Dtest.benchmarkIterations}.
+ */
+ @Test
+ public void testAllocationMicrobenchmark() throws IOException {
+ Assume.assumeTrue("Allocation microbenchmark, opt in with
-Dtest.runAllocationBenchmark=true",
+ Boolean.getBoolean("test.runAllocationBenchmark"));
+ final com.sun.management.ThreadMXBean threadBean =
+ (com.sun.management.ThreadMXBean)
java.lang.management.ManagementFactory.getThreadMXBean();
+ Assume.assumeTrue("JVM does not support thread allocation measurement",
+ threadBean.isThreadAllocatedMemorySupported());
+
+ final String tableName = "TBLALLOC";
+ final int mutationCount = Integer.getInteger("test.mutationCount", 1000);
+ final int cellsPerMutation = Integer.getInteger("test.cellsPerMutation",
4);
+ final int valueSize = Integer.getInteger("test.valueSize", 64);
+ final int iterations = Integer.getInteger("test.benchmarkIterations", 200);
+ final int warmup = Math.max(1, iterations / 10);
+
+ // Pre-build the records once so record construction is not counted in the
measured allocation.
+ List<LogFile.Record> records = new ArrayList<>(mutationCount);
+ byte[] value = new byte[valueSize];
+ for (int m = 0; m < mutationCount; m++) {
+ Put put = new Put(Bytes.toBytes("row" + m));
+ put.setTimestamp(m);
+ for (int c = 0; c < cellsPerMutation; c++) {
+ put.addColumn(Bytes.toBytes("col" + c), Bytes.toBytes("q"), m, value);
+ }
+ List<Cell> cells = new ArrayList<>(cellsPerMutation);
+ for (List<Cell> familyCells : put.getFamilyCellMap().values()) {
+ cells.addAll(familyCells);
+ }
+ records.add(new
LogFileRecord().setHBaseTableName(tableName).setCommitId(m).setCells(cells));
+ }
+ // Large block so a full mutationCount run seals few blocks; block churn
is still exercised.
+ final long blockSize = 1024 * 1024;
+
+ long minBytes = Long.MAX_VALUE;
+ for (int i = 0; i < warmup + iterations; i++) {
+ LogFileTestUtil.SyncableByteArrayOutputStream sink =
+ new LogFileTestUtil.SyncableByteArrayOutputStream();
+ LogFileWriterContext ctx = new
LogFileWriterContext(HBaseConfiguration.create())
+ .setMaxBlockSize(blockSize).setCodec(new LogFileCodec());
+ LogFileFormatWriter w = new LogFileFormatWriter();
+ w.init(ctx, sink);
+ long threadId = Thread.currentThread().getId();
+ long before = threadBean.getThreadAllocatedBytes(threadId);
+ for (LogFile.Record r : records) {
+ w.append(r);
+ }
+ w.close();
+ long allocated = threadBean.getThreadAllocatedBytes(threadId) - before;
+ if (i >= warmup && allocated < minBytes) {
+ minBytes = allocated;
+ }
+ }
+
+ LOG.info(
+ "Allocation benchmark params: mutationCount={} cellsPerMutation={}
valueSize={} iterations={}",
+ mutationCount, cellsPerMutation, valueSize, iterations);
+ LOG.info("Allocation benchmark: totalBytes(min)={} bytesPerRecord={}",
minBytes,
+ String.format("%.1f", (double) minBytes / mutationCount));
+ }
+
/** Aggregated result of one framing mode: best-of encode time and the wire
size produced. */
private static final class BenchResult {
final long appendNs;
