Copilot commented on code in PR #17941:
URL: https://github.com/apache/iotdb/pull/17941#discussion_r3412869400
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java:
##########
@@ -303,29 +424,50 @@ protected void flushAlignedPageToChunkWriter(
// flush new time page to chunk writer directly
alignedChunkWriter.writePageHeaderAndDataIntoTimeBuff(compressedTimePageData,
timePageHeader);
+ long writtenValuePageSize = 0;
// flush new value pages to chunk writer directly
for (int i = 0; i < valuePageHeaders.size(); i++) {
- if (valuePageHeaders.get(i) == null) {
+ PageHeader valuePageHeader = valuePageHeaders.get(i);
+ if (valuePageHeader == null) {
// sub sensor does not exist in current file or value page has been
deleted completely
alignedChunkWriter.getValueChunkWriterByIndex(i).writeEmptyPageToPageBuffer();
continue;
}
alignedChunkWriter.writePageHeaderAndDataIntoValueBuff(
- compressedValuePageDatas.get(i), valuePageHeaders.get(i), i);
+ compressedValuePageDatas.get(i), valuePageHeader, i);
+ if (hasVariableLengthTypeArray[subTaskId]) {
+ writtenValuePageSize += valuePageHeader.getSerializedPageSize();
+ }
}
chunkPointNumArray[subTaskId] += timePageHeader.getStatistics().getCount();
+ if (hasVariableLengthTypeArray[subTaskId]) {
+ // Direct-flushed pages are already serialized, so use value page size
as checkpoint estimate.
+ writtenPointTotalSizeArray[subTaskId] += writtenValuePageSize;
+ }
Review Comment:
In the direct aligned-page flush path, writtenPointTotalSizeArray only
accounts for value page sizes. The time page is also buffered in the chunk
writer, so excluding its serialized size can delay the size checkpoint and
chunk-size checks for aligned chunks with variable-length columns.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java:
##########
@@ -180,17 +212,103 @@ protected void writeDataPoint(long timestamp,
TsPrimitiveType value, IChunkWrite
} else {
AlignedChunkWriterImpl alignedChunkWriter = (AlignedChunkWriterImpl)
chunkWriter;
alignedChunkWriter.write(timestamp, value.getVector());
+ if (hasVariableLengthTypeArray[subTaskId]) {
+ writtenPointTotalSize = estimateWrittenPointTotalSize(value);
+ }
+ }
+ chunkPointNumArray[subTaskId]++;
+ if (hasVariableLengthTypeArray[subTaskId]) {
+ writtenPointTotalSizeArray[subTaskId] += writtenPointTotalSize;
}
}
+ private long estimateWrittenPointTotalSize(TsPrimitiveType value) {
+ long size = Long.BYTES;
+ TsPrimitiveType[] vector = value.getVector();
+ for (TsPrimitiveType tsPrimitiveType : vector) {
+ if (tsPrimitiveType == null) {
+ continue;
+ }
+ TSDataType dataType = tsPrimitiveType.getDataType();
+ switch (dataType) {
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ size += tsPrimitiveType.getBinary().getLength();
+ break;
+ case DOUBLE:
+ case INT64:
+ case TIMESTAMP:
+ size += Long.BYTES;
+ break;
+ case INT32:
+ case DATE:
+ case FLOAT:
+ size += Integer.BYTES;
+ break;
+ case BOOLEAN:
+ size += 1;
+ break;
+ default:
+ break;
+ }
+ }
+ return size;
+ }
+
+ protected long estimateWrittenPointTotalSize(TsBlock tsBlock) {
+ int pointCount = tsBlock.getPositionCount();
+ long size = 0;
Review Comment:
estimateWrittenPointTotalSize(TsBlock) currently ignores the time column
size. Since this value is used to trigger size-based chunk-size checks for
variable-length data, under-counting by ~8 bytes/point can delay the checkpoint
and allow larger-than-expected in-memory buffers.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
