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 f17367d5400fa211663f072aba1331f4153b708d Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 12:09:26 2026 +0800 tmp --- .../common/FlatDeviceDescPriorityMergeReader.java | 48 + .../datastructure/FlatDeviceAlignedTVList.java | 1111 ++++++++++++++++++++ .../datastructure/FlatDeviceAlignedTVListTest.java | 563 ++++++++++ 3 files changed, 1722 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDeviceDescPriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDeviceDescPriorityMergeReader.java new file mode 100644 index 00000000000..eec6d53aa3f --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/FlatDeviceDescPriorityMergeReader.java @@ -0,0 +1,48 @@ +/* + * 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.storageengine.dataregion.read.reader.common; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TimeValuePair; + +import java.util.ArrayList; +import java.util.List; + +/** Descending priority merge reader for TAG-first flat-device pages. */ +public class FlatDeviceDescPriorityMergeReader extends AlignedDescPriorityMergeReader { + + private List<TSDataType> dataTypes = List.of(); + private List<Integer> rowKeyColumnIndices = List.of(); + + public FlatDeviceDescPriorityMergeReader() {} + + @Override + public void configureRowKeyColumns( + List<TSDataType> dataTypes, List<Integer> rowKeyColumnIndices) { + this.dataTypes = new ArrayList<>(dataTypes); + this.rowKeyColumnIndices = new ArrayList<>(rowKeyColumnIndices); + super.configureRowKeyColumns(dataTypes, rowKeyColumnIndices); + } + + @Override + protected int compareRowKey(TimeValuePair left, TimeValuePair right) { + return AlignedPriorityMergeReader.compareFlatDeviceRowKey( + left, right, dataTypes, rowKeyColumnIndices); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java new file mode 100644 index 00000000000..9bb277fa90d --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java @@ -0,0 +1,1111 @@ +/* + * 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.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.i18n.DataNodeMiscMessages; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.io.DataInputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.function.IntPredicate; + +import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; + +/** + * FIELD values have compact column storage; TAGs exist only in the ordered device map. Logical + * column indices remain stable for query/schema/WAL callers and are mapped to FIELDs or TAGs here. + * Sorting changes each device's row indices, never its FIELD arrays or physical timestamps. + */ +public class FlatDeviceAlignedTVList extends AlignedTVList { + static final byte WAL_DATA_TYPE_MARKER = -3; + + private final String tableName; + private final List<Integer> tagColumns; + private final List<Integer> sortColumns = new ArrayList<>(); + private Map<IDeviceID, DeviceInfo> deviceInfoMap = + new TreeMap<>(FlatDeviceAlignedTVList::compareDevices); + private AlignedTVList fields; + private int[] fieldColumns; + private int[] tagOrdinals; + private Object[] fieldInput; + private BitMap[] fieldBitMaps; + private Binary[] lookupTags; + private DeviceInfo lastLookup; + + // The inverse permutation locates a physical row in the sorted device ranges. It avoids both + // a per-row DeviceID dictionary and binary-searching time-sorted rowIndices by physical row ID. + private int[] valuePositions = new int[0]; + private List<DeviceInfo> devicesInOrder = new ArrayList<>(); + private boolean deviceSorted = true; + + FlatDeviceAlignedTVList( + List<TSDataType> types, + List<Integer> tags, + boolean initializeValueColumns, + String tableName) { + super(new ArrayList<>(), false); + this.tableName = Objects.requireNonNull(tableName); + this.dataTypes = new ArrayList<>(types); + this.tagColumns = new ArrayList<>(tags); + Set<Integer> unique = new HashSet<>(); + for (int tag : tags) { + if (tag < 0 + || tag >= types.size() + || !unique.add(tag) + || types.get(tag) != TSDataType.STRING) { + throw new IllegalArgumentException(); + } + } + List<TSDataType> fieldTypes = new ArrayList<>(); + for (int column = 0; column < types.size(); column++) { + if (!unique.contains(column)) { + fieldTypes.add(types.get(column)); + } + } + fields = AlignedTVListFactory.createAlignedList(fieldTypes, initializeValueColumns); + updateColumnMapping(); + timestamps = fields.timestamps; + } + + private void updateColumnMapping() { + fieldColumns = new int[dataTypes.size()]; + tagOrdinals = new int[dataTypes.size()]; + Arrays.fill(fieldColumns, -1); + Arrays.fill(tagOrdinals, -1); + for (int ordinal = 0; ordinal < tagColumns.size(); ordinal++) { + tagOrdinals[tagColumns.get(ordinal)] = ordinal; + } + int field = 0; + for (int column = 0; column < dataTypes.size(); column++) { + if (tagOrdinals[column] < 0) { + fieldColumns[column] = field++; + } + } + fieldInput = new Object[field]; + fieldBitMaps = new BitMap[field]; + lookupTags = new Binary[tagColumns.size()]; + } + + /** Adds a TAG to the schema without allocating a value-column container. */ + public synchronized void addTagColumn(int column, TSDataType type) { + if (type != TSDataType.STRING) { + throw new IllegalArgumentException(); + } + insertLogicalColumn(column, type); + tagColumns.add(column); + updateColumnMapping(); + deviceSorted = false; + seqRowCount = 0; + } + + @Override + public synchronized void insertColumn(int column, TSDataType type) { + if (column < 0 || column > dataTypes.size()) { + throw new IndexOutOfBoundsException(column); + } + int field = 0; + for (int i = 0; i < column; i++) { + if (fieldColumns[i] >= 0) { + field++; + } + } + insertLogicalColumn(column, type); + fields.insertColumn(field, type); + updateColumnMapping(); + } + + private void insertLogicalColumn(int column, TSDataType type) { + if (column < 0 || column > dataTypes.size()) { + throw new IndexOutOfBoundsException(column); + } + tagColumns.replaceAll(index -> index >= column ? index + 1 : index); + sortColumns.replaceAll(index -> index >= column ? index + 1 : index); + dataTypes.add(column, type); + } + + @Override + public void extendColumn(TSDataType type) { + insertColumn(dataTypes.size(), type); + } + + /** FIELD sort columns within each device, followed by time; TAG order comes from the map. */ + public synchronized void setSortColumnIndices(List<Integer> columns) { + Set<Integer> unique = new HashSet<>(); + for (int column : columns) { + if (column < 0 + || column >= dataTypes.size() + || fieldColumns[column] < 0 + || !unique.add(column) + || dataTypes.get(column) == TSDataType.VECTOR + || dataTypes.get(column) == TSDataType.UNKNOWN) { + throw new IllegalArgumentException(); + } + } + sortColumns.clear(); + sortColumns.addAll(columns); + deviceSorted = false; + seqRowCount = 0; + } + + public List<Integer> getTagColumnIndices() { + return new ArrayList<>(tagColumns); + } + + public List<Integer> getSortColumnIndices() { + List<Integer> key = new ArrayList<>(tagColumns); + key.addAll(sortColumns); + return key; + } + + public int getSortColumnCount() { + return tagColumns.size() + sortColumns.size(); + } + + @Override + protected int getRowKeyColumnCount() { + return getSortColumnCount(); + } + + @Override + public synchronized boolean isSorted() { + return deviceSorted; + } + + @Override + public synchronized void putAlignedValue(long time, Object[] input) { + for (int tag = 0; tag < tagColumns.size(); tag++) { + lookupTags[tag] = (Binary) input[tagColumns.get(tag)]; + } + appendDevice(rowCount, time); + for (int column = 0; column < fieldColumns.length; column++) { + if (fieldColumns[column] >= 0) { + fieldInput[fieldColumns[column]] = input[column]; + } + } + fields.putAlignedValue(time, fieldInput); + Arrays.fill(fieldInput, null); + finishAppend(); + } + + @Override + public synchronized void putAlignedValues( + long[] times, Object[] input, BitMap[] nulls, int start, int end, TSStatus[] results) { + if (start >= end) { + return; + } + for (int row = start; row < end; row++) { + boolean failed = + results != null + && results[row] != null + && results[row].code != TSStatusCode.SUCCESS_STATUS.getStatusCode(); + for (int tag = 0; tag < tagColumns.size(); tag++) { + int column = tagColumns.get(tag); + lookupTags[tag] = + failed + || input[column] == null + || (nulls != null && nulls[column] != null && nulls[column].isMarked(row)) + ? null + : ((Binary[]) input[column])[row]; + } + appendDevice(rowCount + row - start, times[row]); + } + for (int column = 0; column < fieldColumns.length; column++) { + int field = fieldColumns[column]; + if (field >= 0) { + fieldInput[field] = input[column]; + fieldBitMaps[field] = nulls == null ? null : nulls[column]; + } + } + fields.putAlignedValues(times, fieldInput, fieldBitMaps, start, end, results); + Arrays.fill(fieldInput, null); + Arrays.fill(fieldBitMaps, null); + finishAppend(); + } + + private void finishAppend() { + int previousCount = rowCount; + rowCount = fields.rowCount(); + timestamps = fields.timestamps; + // FIELD storage has no row-deletion bitmap. Its historical bounds may include deleted + // rows, so only incorporate newly appended timestamps into this list's live bounds. + for (int row = previousCount; row < rowCount; row++) { + long time = fields.getTime(row); + minTime = Math.min(minTime, time); + maxTime = Math.max(maxTime, time); + } + if (indices != null) { + while (indices.size() < timestamps.size()) { + indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32)); + } + for (int row = previousCount; row < rowCount; row++) { + indices.get(row / ARRAY_SIZE)[row % ARRAY_SIZE] = row; + } + } + // Appending after a query sort keeps the published prefix unchanged until the next sort. + deviceSorted = rowCount <= 1; + sorted = false; + seqRowCount = deviceSorted ? rowCount : 0; + } + + private void appendDevice(int row, long time) { + DeviceInfo info = lastLookup; + if (info == null || !info.matches(lookupTags)) { + String[] segments = new String[lookupTags.length + 1]; + segments[0] = tableName; + for (int i = 0; i < lookupTags.length; i++) { + segments[i + 1] = + lookupTags[i] == null + ? null + : lookupTags[i].getStringValue(TSFileConfig.STRING_CHARSET); + } + IDeviceID id = IDeviceID.Factory.DEFAULT_FACTORY.create(segments); + info = deviceInfoMap.computeIfAbsent(id, DeviceInfo::new); + } + info.addRow(row, time); + lastLookup = info; + Arrays.fill(lookupTags, null); + } + + private DeviceInfo deviceAt(int valueIndex) { + DeviceInfo cached = lastLookup; + if (valueIndex < valuePositions.length) { + int position = valuePositions[valueIndex]; + if (cached != null && position >= cached.sortedStart && position < cached.sortedEnd) { + return cached; + } + int low = 0; + int high = devicesInOrder.size() - 1; + while (low <= high) { + int mid = (low + high) >>> 1; + DeviceInfo info = devicesInOrder.get(mid); + if (position < info.sortedStart) { + high = mid - 1; + } else if (position >= info.sortedEnd) { + low = mid + 1; + } else { + return lastLookup = info; + } + } + } else { + if (cached != null && cached.containsUnsortedRow(valueIndex)) { + return cached; + } + for (DeviceInfo info : deviceInfoMap.values()) { + if (info.containsUnsortedRow(valueIndex)) { + return lastLookup = info; + } + } + } + throw new IndexOutOfBoundsException(valueIndex); + } + + static int compareDevices(IDeviceID left, IDeviceID right) { + int count = Math.max(left.segmentNum(), right.segmentNum()); + for (int i = 0; i < count; i++) { + String a = i < left.segmentNum() ? (String) left.segment(i) : null; + String b = i < right.segmentNum() ? (String) right.segment(i) : null; + if (Objects.equals(a, b)) { + continue; + } + if (a == null || b == null) { + return a == null ? -1 : 1; + } + // UTF-8 Binary order is code-point order. String.compareTo uses UTF-16 units and would + // disagree with disk/query TAG ordering for supplementary characters. + int ai = 0; + int bi = 0; + while (ai < a.length() && bi < b.length()) { + int ac = a.codePointAt(ai); + int bc = b.codePointAt(bi); + if (ac != bc) { + return Integer.compare(ac, bc); + } + ai += Character.charCount(ac); + bi += Character.charCount(bc); + } + int result = Integer.compare(a.length() - ai, b.length() - bi); + if (result != 0) { + return result; + } + } + return 0; + } + + @Override + public long getTime(int index) { + return fields.getTime(getValueIndex(index)); + } + + @Override + public synchronized int sort() { + if (deviceSorted) { + return rowCount; + } + if (indices == null) { + indices = new ArrayList<>(); + } + while (indices.size() < timestamps.size()) { + indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32)); + } + valuePositions = new int[rowCount]; + devicesInOrder = new ArrayList<>(deviceInfoMap.values()); + int position = 0; + for (DeviceInfo info : devicesInOrder) { + sortRows(info); + info.sortedStart = position; + for (int i = 0; i < info.rowCount; i++) { + int row = info.rowIndices[i]; + indices.get(position / ARRAY_SIZE)[position % ARRAY_SIZE] = row; + valuePositions[row] = position++; + } + info.sortedEnd = position; + info.sortedRowCount = info.rowCount; + } + lastLookup = null; + deviceSorted = true; + sorted = isTimeOrdered(); + seqRowCount = rowCount; + return rowCount; + } + + private void sortRows(DeviceInfo info) { + // In-place heap sort uses primitive row IDs only. The row-ID tie-break preserves insertion + // order for duplicate keys, which is required for latest-non-null resolution. + for (int root = info.rowCount / 2 - 1; root >= 0; root--) { + siftDown(info.rowIndices, root, info.rowCount); + } + for (int end = info.rowCount - 1; end > 0; end--) { + int value = info.rowIndices[0]; + info.rowIndices[0] = info.rowIndices[end]; + info.rowIndices[end] = value; + siftDown(info.rowIndices, 0, end); + } + } + + private void siftDown(int[] rows, int root, int size) { + int value = rows[root]; + while (root < size / 2) { + int child = root * 2 + 1; + if (child + 1 < size && comparePhysicalRows(rows[child], rows[child + 1]) < 0) { + child++; + } + if (comparePhysicalRows(value, rows[child]) >= 0) { + break; + } + rows[root] = rows[child]; + root = child; + } + rows[root] = value; + } + + private int comparePhysicalRows(int left, int right) { + int result = compareWithinDevice(left, this, right); + return result == 0 ? Integer.compare(left, right) : result; + } + + int compareWithinDevice(int left, FlatDeviceAlignedTVList rightList, int right) { + int count = Math.max(sortColumns.size(), rightList.sortColumns.size()); + for (int i = 0; i < count; i++) { + int a = i < sortColumns.size() ? sortColumns.get(i) : -1; + int b = i < rightList.sortColumns.size() ? rightList.sortColumns.get(i) : -1; + int result = compareField(left, rightList, right, a, b); + if (result != 0) { + return result; + } + } + return Long.compare(fields.getTime(left), rightList.fields.getTime(right)); + } + + private int compareField(int left, FlatDeviceAlignedTVList rightList, int right, int a, int b) { + boolean an = isNullValue(left, a); + boolean bn = rightList.isNullValue(right, b); + if (an || bn) { + return an == bn ? 0 : an ? -1 : 1; + } + return switch (dataTypes.get(a)) { + case BOOLEAN -> + Boolean.compare( + getBooleanByValueIndex(left, a), rightList.getBooleanByValueIndex(right, b)); + case INT32, DATE -> + Integer.compare(getIntByValueIndex(left, a), rightList.getIntByValueIndex(right, b)); + case INT64, TIMESTAMP -> + Long.compare(getLongByValueIndex(left, a), rightList.getLongByValueIndex(right, b)); + case FLOAT -> + Float.compare(getFloatByValueIndex(left, a), rightList.getFloatByValueIndex(right, b)); + case DOUBLE -> + Double.compare(getDoubleByValueIndex(left, a), rightList.getDoubleByValueIndex(right, b)); + case STRING, TEXT, BLOB, OBJECT -> + Integer.signum( + getBinaryByValueIndex(left, a).compareTo(rightList.getBinaryByValueIndex(right, b))); + case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); + }; + } + + @Override + public int compareRowKey(int leftRow, AlignedTVList rightList, int rightRow) { + if (!(rightList instanceof FlatDeviceAlignedTVList flat)) { + return super.compareRowKey(leftRow, rightList, rightRow); + } + int left = getValueIndex(leftRow); + int right = flat.getValueIndex(rightRow); + int result = compareDevices(deviceAt(left).id, flat.deviceAt(right).id); + return result == 0 ? compareWithinDevice(left, flat, right) : result; + } + + @Override + public boolean isNullValue(int row, int column) { + if (column < 0 || column >= dataTypes.size()) { + return true; + } + if (tagOrdinals[column] >= 0) { + return deviceAt(row).tag(tagOrdinals[column]) == null; + } + return fields.isNullValue(row, fieldColumns[column]); + } + + @Override + public Binary getBinaryByValueIndex(int row, int column) { + int tag = tagOrdinals[column]; + return tag >= 0 + ? deviceAt(row).tag(tag) + : fields.getBinaryByValueIndex(row, fieldColumns[column]); + } + + @Override + public int getIntByValueIndex(int row, int column) { + return fields.getIntByValueIndex(row, fieldColumns[column]); + } + + @Override + public long getLongByValueIndex(int row, int column) { + return fields.getLongByValueIndex(row, fieldColumns[column]); + } + + @Override + public float getFloatByValueIndex(int row, int column) { + return fields.getFloatByValueIndex(row, fieldColumns[column]); + } + + @Override + public double getDoubleByValueIndex(int row, int column) { + return fields.getDoubleByValueIndex(row, fieldColumns[column]); + } + + @Override + public boolean getBooleanByValueIndex(int row, int column) { + return fields.getBooleanByValueIndex(row, fieldColumns[column]); + } + + /** Returns physical FIELD storage only, with no TAG placeholders. */ + @Override + public List<List<Object>> getValues() { + return fields.getValues(); + } + + @Override + public List<List<BitMap>> getBitMaps() { + return fields.getBitMaps(); + } + + public boolean isValueArrayMaterialized(int logicalColumn, int block) { + int field = fieldColumns[logicalColumn]; + if (field < 0) { + return false; + } + List<Object> arrays = fields.getValues().get(field); + return arrays != null && block < arrays.size() && arrays.get(block) != null; + } + + @Override + public BitMap getAllValueColDeletedMap(int count) { + if (count == 0) { + return null; + } + BitMap result = new BitMap(count); + for (int row = 0; row < count; row++) { + boolean empty = true; + for (int column = 0; column < dataTypes.size(); column++) { + if (!isNullValue(row, column)) { + empty = false; + break; + } + } + if (empty) { + result.mark(row); + } + } + return result; + } + + @Override + protected boolean markNullValue(int column, int block, int offset) { + if (column < 0 || column >= dataTypes.size()) { + return false; + } + checkFieldDeletion(column); + return fields.markNullValue(fieldColumns[column], block, offset); + } + + private void checkFieldDeletion(int column) { + // TAGs are immutable parts of the logical device identity, including null TAGs. + if (column >= 0 && column < tagOrdinals.length && tagOrdinals[column] >= 0) { + throw new UnsupportedOperationException( + String.format( + DataNodeMiscMessages + .EXCEPTION_DELETING_TAG_COLUMN_ARG_IS_NOT_SUPPORTED_FOR_FLAT_DEVICE_TABLE_ARG_9B440C82, + column, + tableName)); + } + } + + @Override + public synchronized int delete(long lower, long upper) { + // The inherited deleteTime() clears values before marking rows deleted. Keep device keys + // intact in that path as well as when deleting all FIELD values in a time range. + int deleted = 0; + for (int column = 0; column < fieldColumns.length; column++) { + if (fieldColumns[column] >= 0) { + deleted += delete(lower, upper, column).left; + } + } + return deleted; + } + + @Override + public synchronized Pair<Integer, Boolean> delete(long lower, long upper, int column) { + checkFieldDeletion(column); + int deleted = 0; + boolean all = true; + for (int row = 0; row < rowCount; row++) { + long time = fields.getTime(row); + if (time < lower || time > upper) { + all = false; + } else if (markNullValue(column, row / ARRAY_SIZE, row % ARRAY_SIZE)) { + deleted++; + } + } + if (sortColumns.contains(column) && deleted > 0) { + deviceSorted = false; + seqRowCount = 0; + } + return new Pair<>(deleted, all); + } + + @Override + public synchronized int deleteMatchingRows( + long lower, long upper, IntPredicate matches, List<Integer> columns) { + if (columns != null) { + // Validate before mutating any FIELDs in a mixed-column deletion request. + columns.forEach(this::checkFieldDeletion); + } + int deleted = super.deleteMatchingRows(lower, upper, matches, columns); + if (deleted > 0 && columns != null && columns.stream().anyMatch(sortColumns::contains)) { + deviceSorted = false; + seqRowCount = 0; + } + return deleted; + } + + @Override + public synchronized void deleteColumn(int column) { + delete(Long.MIN_VALUE, Long.MAX_VALUE, column); + } + + @Override + public synchronized AlignedTVList clone() { + return copy(fields.clone()); + } + + private FlatDeviceAlignedTVList copy(AlignedTVList fieldCopy) { + FlatDeviceAlignedTVList copy = + new FlatDeviceAlignedTVList(dataTypes, tagColumns, false, tableName); + copy.fields = fieldCopy; + copy.timestamps = fieldCopy.timestamps; + copy.sortColumns.addAll(sortColumns); + copy.rowCount = rowCount; + copy.seqRowCount = seqRowCount; + copy.sorted = sorted; + copy.deviceSorted = deviceSorted; + copy.minTime = minTime; + copy.maxTime = maxTime; + copy.timeDeletedCnt = timeDeletedCnt; + copy.timeColDeletedMap = timeColDeletedMap == null ? null : timeColDeletedMap.clone(); + if (indices != null) { + copy.indices = new ArrayList<>(indices.size()); + for (int[] block : indices) { + copy.indices.add(cloneIndex(block)); + } + } + for (DeviceInfo info : deviceInfoMap.values()) { + copy.deviceInfoMap.put(info.id, info.copy()); + } + copy.valuePositions = valuePositions.clone(); + for (DeviceInfo info : devicesInOrder) { + copy.devicesInOrder.add(copy.deviceInfoMap.get(info.id)); + } + return copy; + } + + @Override + public synchronized AlignedTVList cloneForFlushSort() { + // Row indices are mutable sort state; they must not be shared with an active query snapshot. + return copy(fields.cloneForFlushSort()); + } + + private Set<Integer> fieldSet(Set<Integer> columns) { + if (columns == null) { + return null; + } + Set<Integer> result = new HashSet<>(); + // FIELD sort keys remain necessary even for a time-only query: they define duplicates. + for (int column : sortColumns) { + result.add(fieldColumns[column]); + } + for (int column : columns) { + if (column >= 0 && column < fieldColumns.length && fieldColumns[column] >= 0) { + result.add(fieldColumns[column]); + } + } + return result; + } + + @Override + public synchronized PartialClonePlan preparePartialClone(Set<Integer> columns) { + PartialClonePlan fieldPlan = + fields.preparePartialClone(fieldSet(Objects.requireNonNull(columns))); + FlatDeviceAlignedTVList copy = copy(fieldPlan.getCloneList()); + // Even a FIELD-only query needs device keys to keep equal timestamps on different devices + // distinct. Only FIELD arrays can be released independently. + return new PartialClonePlan(copy, fieldPlan::commit); + } + + @Override + public synchronized void releaseNonQueryColumns(Set<Integer> columns) { + fields.releaseNonQueryColumns(fieldSet(columns)); + } + + @Override + public TVList getTvListByColumnIndex( + List<Integer> columns, List<TSDataType> types, boolean ignoreNull) { + FlatDeviceAlignedTVList source = this; + AlignedTVList projection = + new TimAlignedTVList(new ArrayList<>(types)) { + @Override + public long getTime(int index) { + return source.getTime(index); + } + + @Override + public boolean isTimeOrdered() { + return source.isTimeOrdered(); + } + + @Override + public boolean isSameRowKey(int a, int b) { + return source.isSameRowKey(a, b); + } + + @Override + public boolean isNullValue(int row, int column) { + return source.isNullValue(row, columns.get(column)); + } + + @Override + public Binary getBinaryByValueIndex(int row, int column) { + return source.getBinaryByValueIndex(row, columns.get(column)); + } + + @Override + public int getIntByValueIndex(int row, int column) { + return source.getIntByValueIndex(row, columns.get(column)); + } + + @Override + public long getLongByValueIndex(int row, int column) { + return source.getLongByValueIndex(row, columns.get(column)); + } + + @Override + public float getFloatByValueIndex(int row, int column) { + return source.getFloatByValueIndex(row, columns.get(column)); + } + + @Override + public double getDoubleByValueIndex(int row, int column) { + return source.getDoubleByValueIndex(row, columns.get(column)); + } + + @Override + public boolean getBooleanByValueIndex(int row, int column) { + return source.getBooleanByValueIndex(row, columns.get(column)); + } + }; + projection.rowCount = rowCount; + projection.timestamps = timestamps; + projection.indices = indices; + projection.timeColDeletedMap = timeColDeletedMap; + projection.timeDeletedCnt = timeDeletedCnt; + projection.allValueColDeletedMap = ignoreNull ? getAllValueColDeletedMap() : null; + return projection; + } + + @Override + protected void clearTime() { + // FIELD storage owns the physical timestamps and releases them exactly once. + } + + @Override + protected void clearValue() { + fields.clear(); + deviceInfoMap = new TreeMap<>(FlatDeviceAlignedTVList::compareDevices); + valuePositions = new int[0]; + devicesInOrder = new ArrayList<>(); + lastLookup = null; + timeColDeletedMap = null; + timeDeletedCnt = 0; + deviceSorted = true; + Arrays.fill(fieldInput, null); + Arrays.fill(fieldBitMaps, null); + Arrays.fill(lookupTags, null); + } + + @Override + protected void clearBitMap() {} + + public int getLogicalDeviceCount() { + return deviceInfoMap.size(); + } + + public synchronized Map<IDeviceID, DeviceInfo> getDeviceInfoMap() { + return Collections.unmodifiableMap(deviceInfoMap); + } + + @Override + public synchronized long getRamSize() { + return getRamSize(null); + } + + @Override + public synchronized long getRamSize(Set<Integer> columns) { + long size = + fields.getRamSize(fieldSet(columns)) + + RamUsageEstimator.shallowSizeOf(this) + + RamUsageEstimator.shallowSizeOf(deviceInfoMap) + + RamUsageEstimator.sizeOfIntArray(valuePositions.length) + + RamUsageEstimator.sizeOfIntArray(fieldColumns.length) + + RamUsageEstimator.sizeOfIntArray(tagOrdinals.length) + + RamUsageEstimator.sizeOfObjectArray(devicesInOrder.size()) + + RamUsageEstimator.sizeOfObjectArray(lookupTags.length) + + RamUsageEstimator.sizeOfObjectArray(dataTypes.size()) + + RamUsageEstimator.sizeOfObjectArray(fieldInput.length) * 2; + if (indices != null) { + size += (long) indices.size() * RamUsageEstimator.sizeOfIntArray(ARRAY_SIZE); + } + if (timeColDeletedMap != null) { + size += timeColDeletedMap.ramBytesUsed(); + } + for (DeviceInfo info : deviceInfoMap.values()) { + size += 40 + info.ramBytesUsed(); + } + return size; + } + + @Override + public synchronized RamInfo calculateRamSize() { + return calculateRamSize(null); + } + + @Override + public synchronized RamInfo calculateRamSize(Set<Integer> columns) { + return new RamInfo( + timestamps.size(), + alignedTvListArrayMemCost(columns), + getRamSize(columns), + rowCount, + new ArrayList<>(dataTypes)); + } + + @Override + public int getAvgPointSizeOfLargestColumn() { + int size = fields.getAvgPointSizeOfLargestColumn(); + for (DeviceInfo info : deviceInfoMap.values()) { + for (Binary tag : info.tags) { + if (tag != null) { + size = Math.max(size, tag.getLength()); + } + } + } + return size; + } + + @Override + public int serializedSize() { + int size = + 1 + + Integer.BYTES + + dataTypes.size() + + Integer.BYTES * (tagColumns.size() + sortColumns.size() + 3) + + Long.BYTES * rowCount + + dataTypes.size() * rowCount + + 1; + for (int column = 0; column < dataTypes.size(); column++) { + switch (dataTypes.get(column)) { + case STRING, TEXT, BLOB, OBJECT -> { + for (int row = 0; row < rowCount; row++) { + Binary value = + isNullValue(row, column) ? Binary.EMPTY_VALUE : getBinaryByValueIndex(row, column); + size += ReadWriteIOUtils.sizeToWrite(value); + } + } + case BOOLEAN -> size += rowCount; + case INT32, DATE, FLOAT -> size += Integer.BYTES * rowCount; + case INT64, TIMESTAMP, DOUBLE -> size += Long.BYTES * rowCount; + case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); + } + } + if (timeColDeletedMap != null) { + size += Integer.BYTES + BitMap.getSizeOfBytes(rowCount); + } + return size; + } + + @Override + public void serializeToWAL(IWALByteBufferView buffer) { + buffer.put(WAL_DATA_TYPE_MARKER); + buffer.putInt(dataTypes.size()); + for (TSDataType type : dataTypes) { + buffer.put(type.serialize()); + } + buffer.putInt(tagColumns.size()); + for (int tag : tagColumns) { + buffer.putInt(tag); + } + buffer.putInt(sortColumns.size()); + for (int column : sortColumns) { + buffer.putInt(column); + } + buffer.putInt(rowCount); + // Keep the established column-oriented row layout on WAL. TAGs are projected only here. + for (int row = 0; row < rowCount; row++) { + buffer.putLong(getTime(row)); + } + for (int column = 0; column < dataTypes.size(); column++) { + for (int position = 0; position < rowCount; position++) { + int row = getValueIndex(position); + boolean nil = isNullValue(row, column); + switch (dataTypes.get(column)) { + case STRING, TEXT, BLOB, OBJECT -> + WALWriteUtils.write( + nil ? Binary.EMPTY_VALUE : getBinaryByValueIndex(row, column), buffer); + case BOOLEAN -> WALWriteUtils.write(!nil && getBooleanByValueIndex(row, column), buffer); + case INT32, DATE -> buffer.putInt(nil ? 0 : getIntByValueIndex(row, column)); + case INT64, TIMESTAMP -> buffer.putLong(nil ? 0 : getLongByValueIndex(row, column)); + case FLOAT -> buffer.putFloat(nil ? 0 : getFloatByValueIndex(row, column)); + case DOUBLE -> buffer.putDouble(nil ? 0 : getDoubleByValueIndex(row, column)); + case VECTOR, UNKNOWN -> throw new IllegalArgumentException(); + } + WALWriteUtils.write(nil, buffer); + } + } + buffer.put((byte) (timeColDeletedMap == null ? 0 : 1)); + if (timeColDeletedMap != null) { + // Appended rows can move a deleted row beyond the original bitmap's capacity when + // devices are sorted. The WAL bitmap uses the complete serialized row order. + BitMap deleted = new BitMap(rowCount); + for (int row = 0; row < rowCount; row++) { + if (isTimeDeleted(row)) { + deleted.mark(row); + } + } + buffer.putInt(deleted.getByteArray().length); + buffer.put(deleted.getByteArray()); + } + } + + static FlatDeviceAlignedTVList deserialize(DataInputStream stream, int count, String tableName) + throws IOException { + List<TSDataType> types = new ArrayList<>(); + for (int i = 0; i < count; i++) { + types.add(ReadWriteIOUtils.readDataType(stream)); + } + int tagCount = stream.readInt(); + List<Integer> tags = new ArrayList<>(); + for (int i = 0; i < tagCount; i++) { + tags.add(stream.readInt()); + } + FlatDeviceAlignedTVList list = new FlatDeviceAlignedTVList(types, tags, true, tableName); + int sortCount = stream.readInt(); + List<Integer> columns = new ArrayList<>(); + for (int i = 0; i < sortCount; i++) { + columns.add(stream.readInt()); + } + list.setSortColumnIndices(columns); + AlignedTVList.deserializeRows(stream, list, types); + return list; + } + + /** Encodes lists in device order, merging only row IDs within the current device. */ + public static void encode( + List<FlatDeviceAlignedTVList> lists, + List<IMeasurementSchema> schemas, + List<Integer> columns, + EncryptParameter encryption, + BlockingQueue<Object> output, + long chunkLimit, + int pageLimit) { + FlatDeviceEncoder.encode(lists, schemas, columns, encryption, output, chunkLimit, pageLimit); + } + + AlignedTVList fieldStorage() { + return fields; + } + + int tagOrdinal(int column) { + return column < 0 || column >= tagOrdinals.length ? -1 : tagOrdinals[column]; + } + + @Override + public long alignedTvListArrayMemCost(Set<Integer> columns) { + // The superclass constructor calls this before FIELD storage has been initialized. + return fields == null + ? super.alignedTvListArrayMemCost(columns) + : fields.alignedTvListArrayMemCost(fieldSet(columns)) + + (indices == null ? 0 : (long) ARRAY_SIZE * Integer.BYTES); + } + + @Override + public long alignedTvListArrayMemCostWithoutPrimitiveArrays() { + return fields.alignedTvListArrayMemCostWithoutPrimitiveArrays() + + (indices == null ? 0 : (long) ARRAY_SIZE * Integer.BYTES); + } + + public static final class DeviceInfo { + final IDeviceID id; + final Binary[] tags; + int[] rowIndices = new int[8]; + int rowCount; + int sortedRowCount; + int sortedStart; + int sortedEnd; + private long startTime = Long.MAX_VALUE; + private long endTime = Long.MIN_VALUE; + + DeviceInfo(IDeviceID id) { + this.id = id; + tags = new Binary[Math.max(0, id.segmentNum() - 1)]; + for (int i = 0; i < tags.length; i++) { + Object tag = id.segment(i + 1); + tags[i] = tag == null ? null : new Binary((String) tag, TSFileConfig.STRING_CHARSET); + } + } + + Binary tag(int ordinal) { + return ordinal < tags.length ? tags[ordinal] : null; + } + + boolean matches(Binary[] input) { + int count = Math.max(input.length, tags.length); + for (int i = 0; i < count; i++) { + if (!Objects.equals(i < input.length ? input[i] : null, tag(i))) { + return false; + } + } + return true; + } + + void addRow(int row, long time) { + if (rowCount == rowIndices.length) { + rowIndices = Arrays.copyOf(rowIndices, rowIndices.length * 2); + } + rowIndices[rowCount++] = row; + startTime = Math.min(startTime, time); + endTime = Math.max(endTime, time); + } + + boolean containsUnsortedRow(int row) { + return Arrays.binarySearch(rowIndices, sortedRowCount, rowCount, row) >= 0; + } + + DeviceInfo copy() { + DeviceInfo copy = new DeviceInfo(id); + copy.rowIndices = rowIndices.clone(); + copy.rowCount = rowCount; + copy.sortedRowCount = sortedRowCount; + copy.sortedStart = sortedStart; + copy.sortedEnd = sortedEnd; + copy.startTime = startTime; + copy.endTime = endTime; + return copy; + } + + long ramBytesUsed() { + long size = + RamUsageEstimator.shallowSizeOf(this) + + id.ramBytesUsed() + + RamUsageEstimator.sizeOfIntArray(rowIndices.length) + + RamUsageEstimator.sizeOfObjectArray(tags.length); + for (Binary tag : tags) { + if (tag != null) { + size += RamUsageEstimator.shallowSizeOf(tag) + RamUsageEstimator.sizeOf(tag.getValues()); + } + } + return size; + } + + public int[] getRowIndices() { + return Arrays.copyOf(rowIndices, rowCount); + } + + public long getStartTime() { + return startTime; + } + + public long getEndTime() { + return endTime; + } + + public int getRowCount() { + return rowCount; + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java new file mode 100644 index 00000000000..d9c8c12aadf --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java @@ -0,0 +1,563 @@ +/* + * 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.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.i18n.DataNodeMiscMessages; +import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.junit.Assert; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.IntStream; + +public class FlatDeviceAlignedTVListTest { + + private static final IDeviceID PHYSICAL_DEVICE = + IDeviceID.Factory.DEFAULT_FACTORY.create("table"); + + private static FlatDeviceAlignedTVList list() { + return (FlatDeviceAlignedTVList) + AlignedTVListFactory.createFlatDeviceAlignedList( + new ArrayList<>(List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING)), + List.of(0, 2), + PHYSICAL_DEVICE); + } + + private static Binary binary(String text) { + return new Binary(text, TSFileConfig.STRING_CHARSET); + } + + @Test + public void testDeviceOrderAndFieldThenTimeOrder() { + FlatDeviceAlignedTVList list = list(); + list.setSortColumnIndices(List.of(1)); + list.putAlignedValue(1, new Object[] {binary("b"), 0L, null}); + list.putAlignedValue(1, new Object[] {binary("a"), 2L, null}); + list.putAlignedValue(9, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(3, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(3, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(99, new Object[] {binary("a"), null, null}); + list.sort(); + + List<IDeviceID> devices = new ArrayList<>(list.getDeviceInfoMap().keySet()); + Assert.assertEquals("a", devices.get(0).segment(1)); + Assert.assertEquals("b", devices.get(1).segment(1)); + Assert.assertArrayEquals( + new int[] {5, 3, 4, 2, 1}, list.getDeviceInfoMap().get(devices.get(0)).getRowIndices()); + Assert.assertArrayEquals( + new int[] {0}, list.getDeviceInfoMap().get(devices.get(1)).getRowIndices()); + Assert.assertTrue(list.isSameRowKey(1, 2)); + Assert.assertEquals(99, list.getTime(0)); + Assert.assertEquals(1, list.getTime(4)); + // FIELD/time storage remains in insertion order; only row IDs were permuted. + Assert.assertEquals(1, list.getTimestamps().get(0)[0]); + Assert.assertEquals(99, list.getTimestamps().get(0)[5]); + Assert.assertEquals(1, list.getValues().size()); + } + + @Test + public void testAppendDoesNotRestoreDeletedTimeBounds() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(1, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(100, new Object[] {binary("b"), 100L, null}); + list.putAlignedValue(20, new Object[] {binary("a"), 20L, null}); + list.sort(); + list.deleteMatchingRows(1, 1, row -> true, null); + list.deleteMatchingRows(100, 100, row -> true, null); + list.putAlignedValue(30, new Object[] {binary("c"), 30L, null}); + Assert.assertEquals(20, list.getMinTime()); + Assert.assertEquals(30, list.getMaxTime()); + } + + @Test + public void testWalDeletionAfterAppendMovesRowBeyondOriginalBitmap() throws Exception { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(1, new Object[] {binary("z"), 1L, null}); + list.deleteMatchingRows(1, 1, row -> true, null); + for (int i = 0; i < 1025; i++) { + list.putAlignedValue(i + 2, new Object[] {binary("a"), (long) i, null}); + } + list.sort(); + ByteBuffer buffer = ByteBuffer.allocate(list.serializedSize()); + list.serializeToWAL(new WALByteBufferForTest(buffer)); + Assert.assertEquals(0, buffer.remaining()); + AlignedTVList restored = + AlignedTVListFactory.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), PHYSICAL_DEVICE); + restored.sort(); + Assert.assertFalse(restored.isTimeDeleted(0)); + Assert.assertTrue(restored.isTimeDeleted(1025)); + Assert.assertEquals(2, restored.getMinTime()); + Assert.assertEquals(1026, restored.getMaxTime()); + } + + @Test + public void testInsertFieldKeepsLogicalMappingAndSortKeys() { + FlatDeviceAlignedTVList list = list(); + list.setSortColumnIndices(List.of(1)); + list.putAlignedValue(1, new Object[] {binary("a"), 2L, binary("x")}); + list.insertColumn(1, TSDataType.DOUBLE); + list.putAlignedValue(9, new Object[] {binary("a"), 3.5D, 1L, binary("x")}); + list.sort(); + Assert.assertEquals(List.of(0, 3, 2), list.getSortColumnIndices()); + Assert.assertEquals(2, list.getValues().size()); + Assert.assertEquals(9, list.getTime(0)); + Assert.assertEquals(3.5D, list.getDoubleByValueIndex(list.getValueIndex(0), 1), 0D); + Assert.assertTrue(list.isNullValue(list.getValueIndex(1), 1)); + AlignedTVList.PartialClonePlan plan = list.preparePartialClone(Set.of()); + plan.commit(); + Assert.assertFalse(list.isNullValue(0, 2)); + Assert.assertTrue(list.isNullValue(1, 1)); + Assert.assertEquals(3.5D, plan.getCloneList().getDoubleByValueIndex(1, 1), 0D); + } + + @Test + public void testAppendAfterSortAndIndependentFlushSnapshot() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(8, new Object[] {binary("b"), 8L, null}); + list.putAlignedValue(6, new Object[] {binary("a"), 6L, null}); + list.putAlignedValue(2, new Object[] {binary("b"), 2L, null}); + FlatDeviceAlignedTVList snapshot = (FlatDeviceAlignedTVList) list.cloneForFlushSort(); + snapshot.sort(); + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table", "b"}); + Assert.assertArrayEquals(new int[] {0, 2}, list.getDeviceInfoMap().get(device).getRowIndices()); + Assert.assertArrayEquals( + new int[] {2, 0}, snapshot.getDeviceInfoMap().get(device).getRowIndices()); + list.sort(); + list.putAlignedValue(1, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(4, new Object[] {binary("c"), 4L, null}); + Assert.assertEquals(binary("b"), list.getBinaryByValueIndex(0, 0)); + Assert.assertEquals(binary("a"), list.getBinaryByValueIndex(3, 0)); + Assert.assertEquals(binary("c"), list.getBinaryByValueIndex(4, 0)); + list.sort(); + Assert.assertArrayEquals( + new long[] {1, 6, 2, 8, 4}, + IntStream.range(0, list.rowCount()).mapToLong(list::getTime).toArray()); + Assert.assertEquals(3, snapshot.rowCount()); + } + + @Test + public void testUtf8DeviceOrder() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(1, new Object[] {binary("\uD800\uDC00"), 1L, null}); + list.putAlignedValue(2, new Object[] {binary("\uE000"), 2L, null}); + list.sort(); + Assert.assertEquals(binary("\uE000"), list.getBinaryByValueIndex(list.getValueIndex(0), 0)); + Assert.assertTrue(list.compareRowKey(0, list, 1) < 0); + } + + @Test + public void testWalPreservesFieldSortAndDeletion() throws Exception { + FlatDeviceAlignedTVList list = list(); + list.setSortColumnIndices(List.of(1)); + list.putAlignedValue(1, new Object[] {binary("a"), 2L, null}); + list.putAlignedValue(9, new Object[] {binary("a"), 1L, null}); + list.putAlignedValue(2, new Object[] {binary("b"), 3L, null}); + list.deleteMatchingRows(2, 2, row -> true, null); + list.sort(); + ByteBuffer buffer = ByteBuffer.allocate(list.serializedSize()); + list.serializeToWAL(new WALByteBufferForTest(buffer)); + Assert.assertEquals(0, buffer.remaining()); + FlatDeviceAlignedTVList restored = + (FlatDeviceAlignedTVList) + AlignedTVListFactory.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), PHYSICAL_DEVICE); + Assert.assertEquals(List.of(0, 2, 1), restored.getSortColumnIndices()); + restored.sort(); + Assert.assertEquals(9, restored.getTime(0)); + Assert.assertEquals(1, restored.getTime(1)); + Assert.assertTrue(restored.isTimeDeleted(2)); + + Assert.assertEquals(1, restored.getValues().size()); + } + + @Test + public void testTimeOnlyOwnershipTransferRetainsDeviceKeys() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(1, new Object[] {binary("b"), 1L, null}); + list.putAlignedValue(1, new Object[] {binary("a"), 2L, null}); + list.sort(); + AlignedTVList.PartialClonePlan plan = list.preparePartialClone(Set.of()); + plan.commit(); + Assert.assertFalse(list.isSameRowKey(0, 1)); + Assert.assertEquals(2, list.getLogicalDeviceCount()); + Assert.assertTrue(list.isNullValue(0, 1)); + Assert.assertEquals(2L, plan.getCloneList().getLongByValueIndex(1, 1)); + } + + @Test + public void testRepeatedDevicesUseLogicalDeviceMetadata() { + FlatDeviceAlignedTVList list = list(); + AlignedTVList ordinary = AlignedTVListFactory.createAlignedList(list.getTsDataTypes()); + for (int i = 0; i < 4096; i++) { + Object[] row = {binary("device" + (i % 4)), (long) i, binary("region")}; + list.putAlignedValue(4096 - i, row); + ordinary.putAlignedValue(4096 - i, row); + } + Assert.assertEquals(4, list.getLogicalDeviceCount()); + Assert.assertEquals(1, list.getValues().size()); + Assert.assertTrue(list.getValues().get(0).stream().allMatch(v -> v instanceof long[])); + Assert.assertTrue(list.getRamSize() < ordinary.getRamSize()); + list.sort(); + for (int row = 1; row < list.rowCount(); row++) { + Assert.assertTrue(list.compareRowKey(row - 1, list, row) < 0); + } + Assert.assertEquals(binary("device0"), list.getBinaryByValueIndex(list.getValueIndex(0), 0)); + Assert.assertEquals(4092L, list.getLongByValueIndex(list.getValueIndex(0), 1)); + } + + @Test + public void testDeviceInfoTracksRowsAndTimeRange() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(20, new Object[] {binary("b"), 20L, binary("x")}); + list.putAlignedValue(5, new Object[] {binary("a"), 5L, binary("y")}); + list.putAlignedValue(10, new Object[] {binary("b"), 10L, binary("x")}); + + Map<IDeviceID, FlatDeviceAlignedTVList.DeviceInfo> deviceInfoMap = list.getDeviceInfoMap(); + Assert.assertEquals(2, deviceInfoMap.size()); + IDeviceID deviceA = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table", "a", "y"}); + IDeviceID deviceB = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table", "b", "x"}); + Assert.assertArrayEquals(new int[] {1}, deviceInfoMap.get(deviceA).getRowIndices()); + Assert.assertEquals(1, deviceInfoMap.get(deviceA).getRowCount()); + Assert.assertEquals(5, deviceInfoMap.get(deviceA).getStartTime()); + Assert.assertEquals(5, deviceInfoMap.get(deviceA).getEndTime()); + Assert.assertArrayEquals(new int[] {0, 2}, deviceInfoMap.get(deviceB).getRowIndices()); + Assert.assertEquals(2, deviceInfoMap.get(deviceB).getRowCount()); + Assert.assertEquals(10, deviceInfoMap.get(deviceB).getStartTime()); + Assert.assertEquals(20, deviceInfoMap.get(deviceB).getEndTime()); + } + + @Test + public void testPhysicalDeviceTableNameIsKeptInLogicalDeviceKeys() throws Exception { + IDeviceID physicalDevice = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table_name"}); + FlatDeviceAlignedTVList list = + (FlatDeviceAlignedTVList) + AlignedTVListFactory.createFlatDeviceAlignedList( + new ArrayList<>(List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING)), + List.of(0, 2), + physicalDevice); + list.putAlignedValue(1, new Object[] {binary("device"), 1L, binary("region")}); + + IDeviceID logicalDevice = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table_name", "device", "region"}); + Assert.assertTrue(list.getDeviceInfoMap().containsKey(logicalDevice)); + + ByteBuffer buffer = ByteBuffer.allocate(list.serializedSize()); + list.serializeToWAL(new WALByteBufferForTest(buffer)); + FlatDeviceAlignedTVList recovered = + (FlatDeviceAlignedTVList) + AlignedTVListFactory.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), physicalDevice); + Assert.assertTrue(recovered.getDeviceInfoMap().containsKey(logicalDevice)); + } + + @Test + public void testNullEmptyAndLiteralNullAreDistinct() { + FlatDeviceAlignedTVList list = list(); + for (Binary tag : new Binary[] {null, binary(""), binary("null")}) { + list.putAlignedValue(1, new Object[] {tag, null, null}); + } + Assert.assertEquals(3, list.getLogicalDeviceCount()); + Assert.assertTrue(list.isNullValue(0, 0)); + Assert.assertFalse(list.isNullValue(1, 0)); + Assert.assertTrue(list.getAllValueColDeletedMap().isMarked(0)); + Assert.assertFalse(list.getAllValueColDeletedMap().isMarked(1)); + list.sort(); + Assert.assertFalse(list.isSameRowKey(0, 1)); + Assert.assertFalse(list.isSameRowKey(1, 2)); + } + + @Test + public void testTabletRangeNullsFailuresAndBlockBoundary() { + FlatDeviceAlignedTVList list = list(); + int count = 1029; + long[] times = new long[count]; + long[] fields = new long[count]; + Binary[] first = new Binary[count]; + Binary[] second = new Binary[count]; + BitMap nulls = new BitMap(count); + TSStatus[] results = new TSStatus[count]; + for (int i = 0; i < count; i++) { + times[i] = i; + fields[i] = i; + first[i] = binary("a"); + second[i] = binary("b"); + } + nulls.mark(65); + results[130] = new TSStatus(500); + list.putAlignedValue(0, new Object[] {binary("initial"), 0L, null}); + list.putAlignedValues( + times, + new Object[] {first, fields, second}, + new BitMap[] {nulls, null, null}, + 3, + count - 2, + results); + Assert.assertEquals(count - 4, list.rowCount()); + Assert.assertEquals(4, list.getLogicalDeviceCount()); + Assert.assertTrue(list.isNullValue(63, 0)); + Assert.assertEquals(binary("b"), list.getBinaryByValueIndex(63, 2)); + Assert.assertTrue(list.isNullValue(128, 0)); + Assert.assertTrue(list.isNullValue(128, 1)); + Assert.assertTrue(list.isNullValue(128, 2)); + Assert.assertEquals(1026L, list.getLongByValueIndex(list.rowCount() - 1, 1)); + } + + @Test + public void testLateTagPreservesDeviceIdentity() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(3, new Object[] {binary("a"), 3L, null}); + list.addTagColumn(3, TSDataType.STRING); + list.putAlignedValue(1, new Object[] {binary("a"), 1L, null, null}); + list.putAlignedValue(2, new Object[] {binary("a"), 2L, null, binary("new")}); + Assert.assertEquals(2, list.getLogicalDeviceCount()); + Assert.assertTrue(list.isNullValue(0, 3)); + list.sort(); + Assert.assertEquals(1, list.getTime(0)); + Assert.assertEquals(3, list.getTime(1)); + Assert.assertEquals(2, list.getTime(2)); + } + + @Test + public void testTagDeletionIsRejectedWithoutChangingRows() { + for (boolean sorted : new boolean[] {false, true}) { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(2, new Object[] {binary("a"), 2L, binary("b")}); + list.putAlignedValue(1, new Object[] {null, 1L, null}); + if (sorted) { + list.sort(); + } + List<IDeviceID> devices = new ArrayList<>(list.getDeviceInfoMap().keySet()); + List<int[]> deviceRows = + list.getDeviceInfoMap().values().stream() + .map(FlatDeviceAlignedTVList.DeviceInfo::getRowIndices) + .toList(); + UnsupportedOperationException markNullException = + Assert.assertThrows( + UnsupportedOperationException.class, () -> list.markNullValue(0, 0, 0)); + Assert.assertEquals( + String.format( + DataNodeMiscMessages + .EXCEPTION_DELETING_TAG_COLUMN_ARG_IS_NOT_SUPPORTED_FOR_FLAT_DEVICE_TABLE_ARG_9B440C82, + 0, + "table"), + markNullException.getMessage()); + Assert.assertThrows(UnsupportedOperationException.class, () -> list.markNullValue(0, 0, 1)); + Assert.assertThrows(UnsupportedOperationException.class, () -> list.delete(2, 2, 0)); + UnsupportedOperationException deleteColumnException = + Assert.assertThrows(UnsupportedOperationException.class, () -> list.deleteColumn(2)); + Assert.assertEquals( + String.format( + DataNodeMiscMessages + .EXCEPTION_DELETING_TAG_COLUMN_ARG_IS_NOT_SUPPORTED_FOR_FLAT_DEVICE_TABLE_ARG_9B440C82, + 2, + "table"), + deleteColumnException.getMessage()); + Assert.assertThrows( + UnsupportedOperationException.class, + () -> list.deleteMatchingRows(1, 2, row -> true, List.of(1, 0))); + Assert.assertEquals(devices, new ArrayList<>(list.getDeviceInfoMap().keySet())); + for (int i = 0; i < devices.size(); i++) { + Assert.assertArrayEquals( + deviceRows.get(i), list.getDeviceInfoMap().get(devices.get(i)).getRowIndices()); + } + Assert.assertEquals(binary("a"), list.getBinaryByValueIndex(0, 0)); + Assert.assertEquals(binary("b"), list.getBinaryByValueIndex(0, 2)); + Assert.assertTrue(list.isNullValue(1, 0)); + Assert.assertTrue(list.isNullValue(1, 2)); + Assert.assertFalse(list.isNullValue(0, 1)); + Assert.assertFalse(list.isNullValue(1, 1)); + Assert.assertEquals(sorted, list.isSorted()); + } + } + + @Test + public void testTagDeletionIsRejectedWithoutMatchingData() { + FlatDeviceAlignedTVList list = list(); + Assert.assertThrows(UnsupportedOperationException.class, () -> list.deleteColumn(0)); + list.putAlignedValue(1, new Object[] {null, 1L, null}); + Assert.assertThrows(UnsupportedOperationException.class, () -> list.delete(2, 3, 0)); + Assert.assertThrows( + UnsupportedOperationException.class, + () -> list.deleteMatchingRows(1, 1, row -> true, List.of(0))); + } + + @Test + public void testFieldAndRowDeletionPreserveTags() { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(2, new Object[] {binary("b"), 2L, binary("x")}); + list.putAlignedValue(1, new Object[] {binary("a"), 1L, null}); + list.sort(); + Assert.assertEquals(Integer.valueOf(1), list.delete(2, 2, 1).left); + Assert.assertEquals(Integer.valueOf(0), list.delete(2, 2, 1).left); + Assert.assertTrue(list.isNullValue(0, 1)); + Assert.assertEquals(1, list.delete(1, 1)); + Assert.assertTrue(list.isNullValue(1, 1)); + Assert.assertFalse(list.isTimeDeleted(0)); + Assert.assertEquals(1, list.deleteTime(2, 2)); + Assert.assertEquals(0, list.deleteTime(2, 2)); + Assert.assertTrue(list.isTimeDeleted(1)); + Assert.assertEquals(binary("b"), list.getBinaryByValueIndex(0, 0)); + Assert.assertEquals(binary("x"), list.getBinaryByValueIndex(0, 2)); + Assert.assertEquals(binary("a"), list.getBinaryByValueIndex(1, 0)); + Assert.assertEquals(2, list.getLogicalDeviceCount()); + Assert.assertEquals(1, list.getMinTime()); + Assert.assertEquals(1, list.getMaxTime()); + list.putAlignedValue(3, new Object[] {binary("b"), 3L, binary("x")}); + Assert.assertEquals(2, list.getLogicalDeviceCount()); + Assert.assertEquals(3L, list.getLongByValueIndex(2, 1)); + } + + @Test + public void testWalBeforeAndAfterSort() throws Exception { + FlatDeviceAlignedTVList list = list(); + list.putAlignedValue(2, new Object[] {binary("b"), 2L, binary("x")}); + list.putAlignedValue(3, new Object[] {binary("a"), 3L, null}); + list.putAlignedValue(1, new Object[] {binary("a"), 1L, binary("")}); + list.delete(2, 2, 1); + assertWal(list); + list.sort(); + assertWal(list); + } + + private static void assertWal(FlatDeviceAlignedTVList source) throws Exception { + ByteBuffer buffer = ByteBuffer.allocate(source.serializedSize()); + source.serializeToWAL(new WALByteBufferForTest(buffer)); + Assert.assertEquals(0, buffer.remaining()); + AlignedTVList recovered = + AlignedTVListFactory.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), PHYSICAL_DEVICE); + Assert.assertTrue(recovered instanceof FlatDeviceAlignedTVList); + for (int i = 0; i < source.rowCount(); i++) { + Assert.assertEquals(source.getTime(i), recovered.getTime(i)); + int valueIndex = source.getValueIndex(i); + for (int column : new int[] {0, 2}) { + Assert.assertEquals( + source.isNullValue(valueIndex, column), recovered.isNullValue(i, column)); + if (!source.isNullValue(valueIndex, column)) { + Assert.assertEquals( + source.getBinaryByValueIndex(valueIndex, column), + recovered.getBinaryByValueIndex(i, column)); + } + } + Assert.assertEquals(source.isNullValue(valueIndex, 1), recovered.isNullValue(i, 1)); + if (!source.isNullValue(valueIndex, 1)) { + Assert.assertEquals( + source.getLongByValueIndex(valueIndex, 1), recovered.getLongByValueIndex(i, 1)); + } + } + } + + @Test + public void testCloneAndPartialOwnershipTransfer() { + for (Set<Integer> retained : List.of(Set.of(0), Set.of(1), Set.<Integer>of())) { + FlatDeviceAlignedTVList source = list(); + source.putAlignedValue(1, new Object[] {binary("a"), 1L, binary("b")}); + AlignedTVList full = source.clone(); + AlignedTVList.PartialClonePlan plan = source.preparePartialClone(retained); + AlignedTVList working = plan.getCloneList(); + plan.commit(); + Assert.assertEquals(1, source.getLogicalDeviceCount()); + working.putAlignedValue(2, new Object[] {binary("c"), 2L, binary("d")}); + source.clear(); + Assert.assertEquals(binary("a"), full.getBinaryByValueIndex(0, 0)); + Assert.assertEquals(binary("b"), working.getBinaryByValueIndex(0, 2)); + Assert.assertEquals(binary("c"), working.getBinaryByValueIndex(1, 0)); + } + } + + @Test + public void testClearReleasesDictionaryAndReusesList() { + FlatDeviceAlignedTVList list = list(); + long emptySize = list.getRamSize(); + Binary tag = binary("original"); + list.putAlignedValue(1, new Object[] {tag, 1L, null}); + tag.getValues()[0] = 'X'; + Assert.assertEquals(binary("original"), list.getBinaryByValueIndex(0, 0)); + list.clear(); + Assert.assertEquals(0, list.getLogicalDeviceCount()); + Assert.assertEquals(emptySize, list.getRamSize()); + list.putAlignedValue(1, new Object[] {binary("new"), 1L, null}); + Assert.assertEquals(binary("new"), list.getBinaryByValueIndex(0, 0)); + } + + @Test + public void testQueryProjectionAndFlushSortSnapshot() { + FlatDeviceAlignedTVList source = list(); + source.putAlignedValue(1, new Object[] {binary("b"), 10L, null}); + source.putAlignedValue(9, new Object[] {binary("a"), 90L, binary("x")}); + source.putAlignedValue(1, new Object[] {binary("b"), 11L, null}); + AlignedTVList snapshot = source.cloneForFlushSort(); + snapshot.sort(); + Assert.assertEquals(1, source.getTime(0)); + Assert.assertEquals(9, snapshot.getTime(0)); + AlignedTVList projected = + (AlignedTVList) + snapshot.getTvListByColumnIndex( + List.of(2, 0, 1), + List.of(TSDataType.STRING, TSDataType.STRING, TSDataType.INT64), + false); + Assert.assertEquals(binary("x"), projected.getBinaryByValueIndex(snapshot.getValueIndex(0), 0)); + Assert.assertEquals( + binary("x"), projected.getTimeValuePair(0).getValue().getVector()[0].getBinary()); + Assert.assertEquals(90L, projected.getTimeValuePair(0).getValue().getVector()[2].getLong()); + Assert.assertTrue(projected.isNullValue(snapshot.getValueIndex(1), 0)); + AlignedTVList.AlignedTVListIterator iterator = + snapshot.iterator( + Ordering.ASC, + snapshot.rowCount(), + null, + snapshot.getTsDataTypes(), + List.of(0, 1, 2), + null, + null, + 0, + List.of(TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN), + false, + 100, + new QueryContext(false, false)); + Assert.assertTrue(iterator.hasNextBatch()); + TsBlock block = iterator.nextBatch(); + Assert.assertEquals(2, block.getPositionCount()); + Assert.assertEquals(binary("a"), block.getColumn(0).getBinary(0)); + Assert.assertEquals(9, block.getTimeByIndex(0)); + Assert.assertEquals(binary("b"), block.getColumn(0).getBinary(1)); + Assert.assertEquals(11, block.getColumn(1).getLong(1)); + Assert.assertTrue(block.getColumn(2).isNull(1)); + Assert.assertFalse(iterator.hasNextBatch()); + } +}
