You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/04/15 11:31:20 UTC
(pinot) 02/25: Fix binary key stream decoder (#17927)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch release/1.5.0-STP-2.134.x
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 205b54a3caba6b686b4bc340233f79c599a61102
Author: NOOB <43...@users.noreply.github.com>
AuthorDate: Sat Mar 21 04:41:07 2026 +0530
Fix binary key stream decoder (#17927)
---
.../realtime/RealtimeSegmentDataManager.java | 3 +-
.../writer/StatelessRealtimeSegmentWriter.java | 3 +-
.../pinot/spi/stream/StreamDataDecoderImpl.java | 20 ++++++++++-
.../spi/stream/StreamDataDecoderImplTest.java | 41 ++++++++++++++++++++++
4 files changed, 64 insertions(+), 3 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index 37c9511d2bb..0660cdfb9d8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -1800,7 +1800,8 @@ public class RealtimeSegmentDataManager extends SegmentDataManager {
retryPolicy.attempt(() -> {
try {
StreamMessageDecoder streamMessageDecoder = createMessageDecoder(fieldsToRead);
- localStreamDataDecoder.set(new StreamDataDecoderImpl(streamMessageDecoder));
+ boolean isKeyBytesType = StreamDataDecoderImpl.isKeyBytesType(_schema);
+ localStreamDataDecoder.set(new StreamDataDecoderImpl(streamMessageDecoder, isKeyBytesType));
return true;
} catch (Exception e) {
_segmentLogger.warn("Failed to initialize the StreamMessageDecoder: ", e);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
index 10d37a945a3..cc0c9e92887 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
@@ -221,7 +221,8 @@ public class StatelessRealtimeSegmentWriter implements Closeable {
retryPolicy.attempt(() -> {
try {
StreamMessageDecoder streamMessageDecoder = createMessageDecoder(fieldsToRead);
- localStreamDataDecoder.set(new StreamDataDecoderImpl(streamMessageDecoder));
+ boolean isKeyBytesType = StreamDataDecoderImpl.isKeyBytesType(_schema);
+ localStreamDataDecoder.set(new StreamDataDecoderImpl(streamMessageDecoder, isKeyBytesType));
return true;
} catch (Exception e) {
_logger.warn("Failed to create StreamMessageDecoder. Retrying...", e);
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataDecoderImpl.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataDecoderImpl.java
index 96073ace839..62a51c3c1d9 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataDecoderImpl.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataDecoderImpl.java
@@ -20,6 +20,8 @@ package org.apache.pinot.spi.stream;
import java.nio.charset.StandardCharsets;
import java.util.Map;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,6 +37,7 @@ public class StreamDataDecoderImpl implements StreamDataDecoder {
public static final String RECORD_SERIALIZED_VALUE_SIZE_KEY = METADATA_KEY_PREFIX + "recordSerializedValueSize";
private final StreamMessageDecoder _valueDecoder;
+ private final boolean _isKeyBytesType;
private final GenericRow _reuse = new GenericRow();
/**
@@ -44,8 +47,21 @@ public class StreamDataDecoderImpl implements StreamDataDecoder {
return key.equals(KEY) || key.startsWith(HEADER_KEY_PREFIX) || key.startsWith(METADATA_KEY_PREFIX);
}
+ /**
+ * @return Whether the __key column is defined as BYTES type in the schema.
+ */
+ public static boolean isKeyBytesType(Schema schema) {
+ FieldSpec fieldSpec = schema.getFieldSpecFor(KEY);
+ return fieldSpec != null && fieldSpec.getDataType() == FieldSpec.DataType.BYTES;
+ }
+
public StreamDataDecoderImpl(StreamMessageDecoder valueDecoder) {
+ this(valueDecoder, false);
+ }
+
+ public StreamDataDecoderImpl(StreamMessageDecoder valueDecoder, boolean isKeyBytesType) {
_valueDecoder = valueDecoder;
+ _isKeyBytesType = isKeyBytesType;
}
@Override
@@ -58,7 +74,9 @@ public class StreamDataDecoderImpl implements StreamDataDecoder {
GenericRow row = _valueDecoder.decode(value, 0, length, _reuse);
if (row != null) {
if (message.getKey() != null) {
- row.putValue(KEY, new String(message.getKey(), StandardCharsets.UTF_8));
+ row.putValue(KEY, _isKeyBytesType
+ ? message.getKey()
+ : new String(message.getKey(), StandardCharsets.UTF_8));
}
StreamMessageMetadata metadata = message.getMetadata();
GenericRow headers = metadata.getHeaders();
diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/stream/StreamDataDecoderImplTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/stream/StreamDataDecoderImplTest.java
index 0096691cc13..1819a57b4ec 100644
--- a/pinot-spi/src/test/java/org/apache/pinot/spi/stream/StreamDataDecoderImplTest.java
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/stream/StreamDataDecoderImplTest.java
@@ -82,6 +82,47 @@ public class StreamDataDecoderImplTest {
Assert.assertEquals(row.getValue(StreamDataDecoderImpl.RECORD_SERIALIZED_VALUE_SIZE_KEY), value.length());
}
+ /**
+ * Demonstrates that binary keys (e.g. Confluent Avro wire format with 0x00 magic byte)
+ * were previously corrupted when decoded as UTF-8 string, and now work correctly when
+ * the __key column is defined as BYTES type.
+ */
+ @Test
+ public void testBinaryKeyCorruptedAsString() {
+ TestDecoder messageDecoder = new TestDecoder();
+ messageDecoder.init(Map.of(), Set.of(NAME_FIELD), "");
+ String value = "Alice";
+ // Simulate binary key with bytes that are invalid UTF-8 sequences.
+ // 0xFE and 0xFF are never valid in UTF-8; 0x80 is an invalid start byte.
+ // A UTF-8 round-trip will replace these with the replacement character U+FFFD,
+ // corrupting the original bytes.
+ byte[] avroKey = new byte[]{(byte) 0xFE, (byte) 0xFF, (byte) 0x80, 0x01, 0x02};
+ StreamMessageMetadata metadata = new StreamMessageMetadata.Builder().setRecordIngestionTimeMs(1234L)
+ .setOffset(new LongMsgOffset(0), new LongMsgOffset(1))
+ .build();
+ BytesStreamMessage message =
+ new BytesStreamMessage(avroKey, value.getBytes(StandardCharsets.UTF_8), metadata);
+
+ // Old behavior (isKeyBytesType=false): binary key is decoded as UTF-8 string.
+ // The leading 0x00 byte and other non-printable bytes produce a corrupted string
+ // that does NOT round-trip back to the original bytes.
+ StreamDataDecoderResult stringResult = new StreamDataDecoderImpl(messageDecoder).decode(message);
+ Assert.assertNotNull(stringResult.getResult());
+ Object stringKey = stringResult.getResult().getValue(StreamDataDecoderImpl.KEY);
+ Assert.assertTrue(stringKey instanceof String);
+ // The UTF-8 round-trip corrupts the binary data
+ Assert.assertNotEquals(((String) stringKey).getBytes(StandardCharsets.UTF_8), avroKey,
+ "Binary key should be corrupted when decoded as UTF-8 string");
+
+ // New behavior (isKeyBytesType=true): binary key bytes are preserved as-is
+ StreamDataDecoderResult bytesResult = new StreamDataDecoderImpl(messageDecoder, true).decode(message);
+ Assert.assertNotNull(bytesResult.getResult());
+ Object bytesKey = bytesResult.getResult().getValue(StreamDataDecoderImpl.KEY);
+ Assert.assertTrue(bytesKey instanceof byte[]);
+ Assert.assertEquals((byte[]) bytesKey, avroKey,
+ "Binary key should be preserved losslessly when __key is BYTES type");
+ }
+
@Test
public void testNoExceptionIsThrown() {
ThrowingDecoder messageDecoder = new ThrowingDecoder();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org