@Override
  public void writeDecodeValuePage(long[] times, TsPrimitiveType[] values, TSDataType dataType)
      throws IOException {
    pageNumbers.set(pageNumbers.size() - 1, pageNumbers.get(pageNumbers.size() - 1) + 1);
    final long startTime = timePartitionSlot.getStartTime();
    // beware of overflow
    long endTime = startTime + TimePartitionUtils.getTimePartitionInterval() - 1;
    if (endTime <= startTime) {
      endTime = Long.MAX_VALUE;
    }
    final int satisfiedLength = satisfiedLengthQueue.poll();
    // serialize needDecode==true
    dataSize += ReadWriteIOUtils.write(true, stream);
    dataSize += ReadWriteIOUtils.write(satisfiedLength, stream);
    satisfiedLengthQueue.offer(satisfiedLength);

    // The time column data is not included after serialization, so each value page needs to record
    // its start time and end time.
    long pageStartTime = Long.MAX_VALUE, pageEndTime = Long.MIN_VALUE;
    for (int i = 0; i < times.length; i++) {
      if (times[i] > endTime) {
        break;
      }
      if (times[i] >= startTime) {
        if (values.length == 0 || values[i] == null) {
          dataSize += ReadWriteIOUtils.write(true, stream);
        } else {
          pageStartTime = Math.min(pageStartTime, times[i]);
          pageEndTime = Math.max(pageEndTime, times[i]);
          dataSize += ReadWriteIOUtils.write(false, stream);
          switch (dataType) {
            case INT32:
            case DATE:
              dataSize += ReadWriteIOUtils.write(values[i].getInt(), stream);
              break;
            case INT64:
            case TIMESTAMP:
              dataSize += ReadWriteIOUtils.write(values[i].getLong(), stream);
              break;
            case FLOAT:
              dataSize += ReadWriteIOUtils.write(values[i].getFloat(), stream);
              break;
            case DOUBLE:
              dataSize += ReadWriteIOUtils.write(values[i].getDouble(), stream);
              break;
            case BOOLEAN:
              dataSize += ReadWriteIOUtils.write(values[i].getBoolean(), stream);
              break;
            case TEXT:
            case BLOB:
            case STRING:
              dataSize += ReadWriteIOUtils.write(values[i].getBinary(), stream);
              break;
            default:
              throw new UnSupportedDataTypeException(
                  String.format("Data type %s is not supported.", dataType));
          }
        }
      }
    }
    dataSize += ReadWriteIOUtils.write(pageStartTime, stream);
    dataSize += ReadWriteIOUtils.write(pageEndTime, stream);
  }
This class is used to be compatible with the new distribution of aligned series in chunk group.
In past versions, a time chunk and all its corresponding value chunks were continuous within the
chunk group. In order to solve the problem of excessive memory usage during compaction, the value
column is grouped and merged, which will cause the above rules to be broken.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/splitter/BatchedAlignedValueChunkData.java