gortiz commented on code in PR #13303:
URL: https://github.com/apache/pinot/pull/13303#discussion_r1734590564
##########
pinot-common/src/main/java/org/apache/pinot/common/datablock/BaseDataBlock.java:
##########
@@ -426,168 +390,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.
- * <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.
+ * Returns the list of serialized stats.
* <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.
+ * The returned list may contain nulls, which would mean that no stats were
available for that stage.
* <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) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof DataBlock)) {
+ return false;
+ }
+ DataBlock that = (DataBlock) o;
+ if (getDataBlockType() != that.getDataBlockType()) {
+ return false;
+ }
+ switch (getDataBlockType()) {
+ case ROW:
+ case COLUMNAR:
+ assert _dataSchema != null;
+ if (!_dataSchema.equals(that.getDataSchema())) {
+ return false;
+ }
+ if (_numRows != that.getNumberOfRows() || _numColumns !=
that.getNumberOfColumns()) {
+ return false;
+ }
+ DataSchema.ColumnDataType[] colTypes =
_dataSchema.getColumnDataTypes();
+ String[] colNames = _dataSchema.getColumnNames();
+ for (int colId = 0; colId < colNames.length; colId++) {
+ switch (colTypes[colId]) {
+ case INT:
+ case BOOLEAN:
+ for (int did = 0; did < _numRows; did++) {
+ if (getInt(did, colId) != that.getInt(did, colId)) {
+ return false;
+ }
+ }
+ break;
+ case LONG:
+ for (int did = 0; did < _numRows; did++) {
+ if (getLong(did, colId) != that.getLong(did, colId)) {
+ return false;
+ }
+ }
+ break;
+ case FLOAT:
+ for (int did = 0; did < _numRows; did++) {
+ if (getFloat(did, colId) != that.getFloat(did, colId)) {
+ return false;
+ }
+ }
+ break;
+ case DOUBLE:
+ for (int did = 0; did < _numRows; did++) {
+ if (getDouble(did, colId) != that.getDouble(did, colId)) {
+ return false;
+ }
+ }
+ break;
+ case TIMESTAMP:
+ break;
Review Comment:
Yes, that should be merged with LONG
--
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
