gortiz commented on code in PR #13303:
URL: https://github.com/apache/pinot/pull/13303#discussion_r1684456687
##########
pinot-common/src/main/java/org/apache/pinot/common/datablock/BaseDataBlock.java:
##########
@@ -426,168 +401,202 @@ public Map<Integer, String> getExceptions() {
return _errCodeToExceptionMap;
}
- /**
- * Serialize this data block to a byte array.
- * <p>
- * In order to deserialize it, {@link
DataBlockUtils#getDataBlock(ByteBuffer)} should be used.
- */
@Override
- public byte[] toBytes()
+ public List<ByteBuffer> serialize()
throws IOException {
- ThreadResourceUsageProvider threadResourceUsageProvider = new
ThreadResourceUsageProvider();
-
- UnsynchronizedByteArrayOutputStream byteArrayOutputStream = new
UnsynchronizedByteArrayOutputStream(8192);
- DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream);
- writeLeadingSections(dataOutputStream);
-
- // Write metadata: length followed by actual metadata bytes.
- // NOTE: We ignore metadata serialization time in
"responseSerializationCpuTimeNs" as it's negligible while
- // considering it will bring a lot code complexity.
- serializeMetadata(dataOutputStream);
-
- return byteArrayOutputStream.toByteArray();
- }
-
- private void writeLeadingSections(DataOutputStream dataOutputStream)
- throws IOException {
- dataOutputStream.writeInt(getDataBlockVersionType());
- dataOutputStream.writeInt(_numRows);
- dataOutputStream.writeInt(_numColumns);
- int dataOffset = HEADER_SIZE;
-
- // Write exceptions section offset(START|SIZE).
- dataOutputStream.writeInt(dataOffset);
- byte[] exceptionsBytes;
- exceptionsBytes = serializeExceptions();
- dataOutputStream.writeInt(exceptionsBytes.length);
- dataOffset += exceptionsBytes.length;
-
- // Write dictionary map section offset(START|SIZE).
- dataOutputStream.writeInt(dataOffset);
- byte[] dictionaryBytes = null;
- if (_stringDictionary != null) {
- dictionaryBytes = serializeStringDictionary();
- dataOutputStream.writeInt(dictionaryBytes.length);
- dataOffset += dictionaryBytes.length;
- } else {
- dataOutputStream.writeInt(0);
- }
-
- // Write data schema section offset(START|SIZE).
- dataOutputStream.writeInt(dataOffset);
- byte[] dataSchemaBytes = null;
- if (_dataSchema != null) {
- dataSchemaBytes = _dataSchema.toBytes();
- dataOutputStream.writeInt(dataSchemaBytes.length);
- dataOffset += dataSchemaBytes.length;
- } else {
- dataOutputStream.writeInt(0);
+ if (_serialized == null) {
+ _serialized = DataBlockUtils.serialize(this);
}
+ return _serialized;
+ }
- // Write fixed size data section offset(START|SIZE).
- dataOutputStream.writeInt(dataOffset);
- if (_fixedSizeDataBytes != null) {
- dataOutputStream.writeInt(_fixedSizeDataBytes.length);
- dataOffset += _fixedSizeDataBytes.length;
+ @Override
+ public String toString() {
+ if (_dataSchema == null) {
+ return "{}";
} else {
- dataOutputStream.writeInt(0);
+ return "resultSchema:" + '\n' + _dataSchema + '\n' + "numRows: " +
_numRows + '\n';
}
+ }
- // Write variable size data section offset(START|SIZE).
- dataOutputStream.writeInt(dataOffset);
- if (_variableSizeDataBytes != null) {
- dataOutputStream.writeInt(_variableSizeDataBytes.length);
- } else {
- dataOutputStream.writeInt(0);
- }
+ @Nullable
+ @Override
+ public String[] getStringDictionary() {
+ return _stringDictionary;
+ }
- // Write actual data.
- // Write exceptions bytes.
- dataOutputStream.write(exceptionsBytes);
- // Write dictionary map bytes.
- if (dictionaryBytes != null) {
- dataOutputStream.write(dictionaryBytes);
- }
- // Write data schema bytes.
- if (dataSchemaBytes != null) {
- dataOutputStream.write(dataSchemaBytes);
- }
- // Write fixed size data bytes.
- if (_fixedSizeDataBytes != null) {
- dataOutputStream.write(_fixedSizeDataBytes);
- }
- // Write variable size data bytes.
- if (_variableSizeDataBytes != null) {
- dataOutputStream.write(_variableSizeDataBytes);
- }
+ @Nullable
+ @Override
+ public DataBuffer getFixedData() {
+ return _fixedSizeData;
}
- /**
- * Writes the metadata section to the given data output stream.
- */
- protected void serializeMetadata(DataOutputStream dataOutputStream)
- throws IOException {
- dataOutputStream.writeInt(0);
+ @Nullable
+ @Override
+ public DataBuffer getVarSizeData() {
+ return _variableSizeData;
}
/**
- * Deserializes the metadata section from the given byte buffer.
+ * Returns the list of serialized stats.
* <p>
- * This is the counterpart of {@link #serializeMetadata(DataOutputStream)}
and it is guaranteed that the buffer will
- * be positioned at the start of the metadata section when this method is
called.
+ * The returned list may contain nulls, which would mean that no stats were
available for that stage.
* <p>
- * <strong>Important:</strong> It is mandatory for implementations to leave
the cursor at the end of the metadata, in
- * the exact same position as it was when {@link
#serializeMetadata(DataOutputStream)} was called.
- * <p>
- * <strong>Important:</strong> This method will be called at the end of the
BaseDataConstructor constructor to read
- * the metadata section. This means that it will be called
<strong>before</strong> the subclass have been constructor
- * have been called. Therefore it is not possible to use any subclass fields
in this method.
+ * The list itself may also be null.
*/
- protected void deserializeMetadata(ByteBuffer buffer)
- throws IOException {
- buffer.getInt();
+ @Nullable
+ @Override
+ public List<DataBuffer> getStatsByStage() {
+ return Collections.emptyList();
}
- private byte[] serializeExceptions()
- throws IOException {
- if (_errCodeToExceptionMap.isEmpty()) {
- return new byte[4];
- }
- UnsynchronizedByteArrayOutputStream byteArrayOutputStream = new
UnsynchronizedByteArrayOutputStream(1024);
- DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream);
-
- dataOutputStream.writeInt(_errCodeToExceptionMap.size());
-
- for (Map.Entry<Integer, String> entry : _errCodeToExceptionMap.entrySet())
{
- int key = entry.getKey();
- String value = entry.getValue();
- byte[] valueBytes = value.getBytes(UTF_8);
- dataOutputStream.writeInt(key);
- dataOutputStream.writeInt(valueBytes.length);
- dataOutputStream.write(valueBytes);
- }
-
- return byteArrayOutputStream.toByteArray();
+ @Override
+ public Raw asRaw() {
+ return this;
}
- private Map<Integer, String> deserializeExceptions(ByteBuffer buffer)
- throws IOException {
- int numExceptions = buffer.getInt();
- Map<Integer, String> exceptions = new
HashMap<>(HashUtil.getHashMapCapacity(numExceptions));
- for (int i = 0; i < numExceptions; i++) {
- int errCode = buffer.getInt();
- String errMessage = DataTableUtils.decodeString(buffer);
- exceptions.put(errCode, errMessage);
+ @Override
+ public final boolean equals(Object o) {
Review Comment:
That is what ByteBuffer does under the hood. Also, if we end up comparing
data blocks in production right now, the difference shouldn't be that large:
1. If the byte buffers are pointing to different data, we will fail fast (as
fast as the first different byte is compared). We would end up returning not
equal as before.
2. If we are comparing the same instances, it will success fast (as we
include the `==` shortcut). We would end up returning equal as before
3. If we are comparing different instances pointing to memory that contains
the same bytes, it will have a linear cost and we will be returning equal
instead of not equal as before.
Only in the third case we would have problems. But probably the issue there
would have been that we wanted to compare references instead of content, in
which case we should not be using equals. And I it is clear we shouldn't be
doing that.
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
