This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5fef8eb1d99aff4875ee159d0443b0cd45157e52 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 12:29:50 2026 +0800 tmp --- .../db/utils/datastructure/FlatDeviceEncoder.java | 203 +++++++++++++++++++++ 1 file changed, 203 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java new file mode 100644 index 00000000000..99f4951668b --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java @@ -0,0 +1,203 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.utils.datastructure; + +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; +import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList.DeviceInfo; + +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.PriorityQueue; +import java.util.concurrent.BlockingQueue; + +/** Device-first flush merge using reusable cursors and primitive physical row indices. */ +final class FlatDeviceEncoder { + private FlatDeviceEncoder() {} + + static void encode( + List<FlatDeviceAlignedTVList> lists, + List<IMeasurementSchema> schemas, + List<Integer> columns, + EncryptParameter encryption, + BlockingQueue<Object> output, + long chunkLimit, + int pageLimit) { + PriorityQueue<Cursor> devices = + new PriorityQueue<>( + (a, b) -> FlatDeviceAlignedTVList.compareDevices(a.device.id, b.device.id)); + PriorityQueue<Cursor> rows = + new PriorityQueue<>( + (a, b) -> { + int result = a.list.compareWithinDevice(a.row(), b.list, b.row()); + // Older lists and earlier inserts are consumed first. A later non-null value wins. + return result == 0 ? Integer.compare(a.priority, b.priority) : result; + }); + for (int i = 0; i < lists.size(); i++) { + FlatDeviceAlignedTVList list = lists.get(i); + list.sort(); + Cursor cursor = new Cursor(list, i); + if (cursor.nextDevice()) { + devices.add(cursor); + } + } + FlatDeviceAlignedTVList[] selectedLists = new FlatDeviceAlignedTVList[columns.size()]; + int[] selectedRows = new int[columns.size()]; + FlatDeviceAlignedChunkWriter writer = new FlatDeviceAlignedChunkWriter(schemas, encryption); + long chunkRows = 0; + while (!devices.isEmpty()) { + // All rows in this group share TAGs; retain them even when a cursor advances devices. + DeviceInfo currentDevice = devices.peek().device; + do { + Cursor cursor = devices.remove(); + rows.add(cursor); + } while (!devices.isEmpty() + && FlatDeviceAlignedTVList.compareDevices(currentDevice.id, devices.peek().device.id) + == 0); + + while (!rows.isEmpty()) { + Cursor first = rows.peek(); + FlatDeviceAlignedTVList keyList = first.list; + int keyRow = first.row(); + long time = keyList.fieldStorage().getTime(keyRow); + Arrays.fill(selectedLists, null); + do { + Cursor cursor = rows.remove(); + int row = cursor.row(); + for (int i = 0; i < columns.size(); i++) { + int column = columns.get(i); + int tag = cursor.list.tagOrdinal(column); + boolean nil = + tag >= 0 ? cursor.device.tag(tag) == null : cursor.list.isNullValue(row, column); + if (!nil) { + selectedLists[i] = cursor.list; + selectedRows[i] = row; + } + } + if (cursor.nextRow()) { + rows.add(cursor); + } else if (cursor.nextDevice()) { + devices.add(cursor); + } + } while (!rows.isEmpty() + && keyList.compareWithinDevice(keyRow, rows.peek().list, rows.peek().row()) == 0); + + // Reuse the ordinary typed chunk writers; only the traversal and duplicate selection + // differ. No TimeValuePair, vector or per-row iterator wrapper is materialized. + for (int i = 0; i < columns.size(); i++) { + FlatDeviceAlignedTVList list = selectedLists[i]; + int row = selectedRows[i]; + int column = columns.get(i); + boolean nil = list == null; + switch (schemas.get(i).getType()) { + case BOOLEAN -> + writer.write(time, !nil && list.getBooleanByValueIndex(row, column), nil); + case INT32, DATE -> + writer.write(time, nil ? 0 : list.getIntByValueIndex(row, column), nil); + case INT64, TIMESTAMP -> + writer.write(time, nil ? 0L : list.getLongByValueIndex(row, column), nil); + case FLOAT -> + writer.write(time, nil ? 0F : list.getFloatByValueIndex(row, column), nil); + case DOUBLE -> + writer.write(time, nil ? 0D : list.getDoubleByValueIndex(row, column), nil); + case STRING, TEXT, BLOB, OBJECT -> { + Binary value = Binary.EMPTY_VALUE; + if (!nil) { + int tag = list.tagOrdinal(column); + value = tag >= 0 ? currentDevice.tag(tag) : list.getBinaryByValueIndex(row, column); + } + writer.write(time, value, nil); + } + case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); + } + } + writer.write(time); + chunkRows++; + // The chunk writer may already have sealed on its byte-size threshold. + if (writer.getTimeChunkWriter().getPageWriter().getPointNumber() >= pageLimit) { + writer.sealCurrentPage(); + } + if (chunkRows >= chunkLimit) { + if (!enqueue(writer, output)) { + return; + } + writer = new FlatDeviceAlignedChunkWriter(schemas, encryption); + chunkRows = 0; + } + } + } + if (chunkRows > 0) { + enqueue(writer, output); + } + } + + private static boolean enqueue( + FlatDeviceAlignedChunkWriter writer, BlockingQueue<Object> output) { + writer.sealCurrentPage(); + writer.clearPageWriter(); + try { + output.put(writer); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + private static final class Cursor { + private final FlatDeviceAlignedTVList list; + private final int priority; + private final Iterator<DeviceInfo> devices; + private DeviceInfo device; + private int position; + + private Cursor(FlatDeviceAlignedTVList list, int priority) { + this.list = list; + this.priority = priority; + devices = list.getDeviceInfoMap().values().iterator(); + } + + private int row() { + return device.rowIndices[position]; + } + + private boolean nextDevice() { + while (devices.hasNext()) { + device = devices.next(); + position = -1; + if (nextRow()) { + return true; + } + } + return false; + } + + private boolean nextRow() { + do { + position++; + } while (position < device.rowCount && list.isTimeDeleted(row(), false)); + return position < device.rowCount; + } + } +}
