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 fc5a333f8259143194f93b92e74f22e801c7542c Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 10 11:05:17 2026 +0800 support flat_device deletion --- .../operator/source/FlatDeviceDeletionFilter.java | 189 +++++++++++ .../execution/operator/source/SeriesScanUtil.java | 4 +- .../dataregion/modification/DeletionPredicate.java | 4 + .../modification/FlatDeviceDeletionEntry.java | 126 +++++++ .../dataregion/modification/TagPredicate.java | 106 +++++- .../memtable/FlatDeviceMemTableDeletionTest.java | 375 +++++++++++++++++++++ .../modification/TableDeletionEntryTest.java | 68 ++++ 7 files changed, 868 insertions(+), 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDeviceDeletionFilter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDeviceDeletionFilter.java new file mode 100644 index 00000000000..87ea6cceff9 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDeviceDeletionFilter.java @@ -0,0 +1,189 @@ +/* + * 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.queryengine.execution.operator.source; + +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; + +import org.apache.tsfile.file.metadata.IMetadata; +import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.filter.basic.OperatorType; +import org.apache.tsfile.utils.Binary; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** Row filter for TAG predicates stored in a flat-device TsFile modification file. */ +final class FlatDeviceDeletionFilter extends Filter { + private final int[] tagColumnIndices; + private final List<FlatDeviceDeletionEntry> deletions; + + FlatDeviceDeletionFilter(int[] tagColumnIndices, List<FlatDeviceDeletionEntry> deletions) { + this.tagColumnIndices = tagColumnIndices; + this.deletions = deletions; + } + + private boolean keep(long time, Object[] values) { + for (FlatDeviceDeletionEntry deletion : deletions) { + TimeRange range = deletion.getTimeRange(); + if (range.contains(time) && deletion.getPredicate().matches(values, tagColumnIndices)) { + return false; + } + } + return true; + } + + @Override + public boolean satisfy(long time, Object value) { + return true; + } + + @Override + public boolean satisfyBoolean(long time, boolean value) { + return true; + } + + @Override + public boolean satisfyInteger(long time, int value) { + return true; + } + + @Override + public boolean satisfyLong(long time, long value) { + return true; + } + + @Override + public boolean satisfyFloat(long time, float value) { + return true; + } + + @Override + public boolean satisfyDouble(long time, double value) { + return true; + } + + @Override + public boolean satisfyBinary(long time, Binary value) { + return true; + } + + @Override + public boolean satisfyString(long time, String value) { + return true; + } + + @Override + public boolean satisfyRow(long time, Object[] values) { + return keep(time, values); + } + + @Override + public boolean satisfyBooleanRow(long time, boolean[] values) { + return true; + } + + @Override + public boolean satisfyIntegerRow(long time, int[] values) { + return true; + } + + @Override + public boolean satisfyLongRow(long time, long[] values) { + return true; + } + + @Override + public boolean satisfyFloatRow(long time, float[] values) { + return true; + } + + @Override + public boolean satisfyDoubleRow(long time, double[] values) { + return true; + } + + @Override + public boolean satisfyBinaryRow(long time, Binary[] values) { + return keep(time, values); + } + + @Override + public boolean[] satisfyTsBlock(TsBlock tsBlock) { + return satisfyTsBlock(null, tsBlock); + } + + @Override + public boolean[] satisfyTsBlock(boolean[] selection, TsBlock tsBlock) { + boolean[] result = + selection == null ? new boolean[tsBlock.getPositionCount()] : selection.clone(); + if (selection == null) { + Arrays.fill(result, true); + } + Object[] values = new Object[tsBlock.getValueColumnCount()]; + for (int row = 0; row < result.length; row++) { + if (result[row]) { + for (int column = 0; column < values.length; column++) { + values[column] = + tsBlock.getColumn(column).isNull(row) + ? null + : tsBlock.getColumn(column).getObject(row); + } + result[row] = keep(tsBlock.getTimeByIndex(row), values); + } + } + return result; + } + + @Override + public boolean canSkip(IMetadata metadata) { + return false; + } + + @Override + public boolean allSatisfy(IMetadata metadata) { + return false; + } + + @Override + public boolean satisfyStartEndTime(long startTime, long endTime) { + return true; + } + + @Override + public boolean containStartEndTime(long startTime, long endTime) { + return false; + } + + @Override + public List<TimeRange> getTimeRanges() { + return Collections.emptyList(); + } + + @Override + public Filter reverse() { + throw new UnsupportedOperationException(); + } + + @Override + public OperatorType getOperatorType() { + return OperatorType.FALSE_LITERAL; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index 7729b866a53..a786170cb43 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -799,9 +799,7 @@ public class SeriesScanUtil implements Accountable { throw new IllegalStateException(e); } if (!deletions.isEmpty()) { - Filter filter = - new FlatDeviceDeletionFilter( - seriesPath.getDeviceId().getTableName(), rowKeyColumnIndices, deletions); + Filter filter = new FlatDeviceDeletionFilter(rowKeyColumnIndices, deletions); pageReaderList.forEach(pageReader -> pageReader.addRecordFilter(filter)); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionPredicate.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionPredicate.java index 7a6ec6f15f8..91d42b0eb68 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionPredicate.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionPredicate.java @@ -80,6 +80,10 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable && tagPredicate.matches(deviceSegments); } + public boolean matches(Object[] values, int[] tagColumnIndices) { + return tagPredicate.matches(values, tagColumnIndices); + } + public void setTagPredicate(TagPredicate tagPredicate) { this.tagPredicate = tagPredicate; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java new file mode 100644 index 00000000000..ab117841034 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java @@ -0,0 +1,126 @@ +/* + * 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.modification; + +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.utils.ReadWriteForEncodingUtils; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** Deletion entry whose predicate is evaluated against rows in a flat-device table. */ +public class FlatDeviceDeletionEntry extends TableDeletionEntry { + private List<String> tagColumnNames = Collections.emptyList(); + + public FlatDeviceDeletionEntry() { + super(); + modType = ModType.FLAT_DEVICE_DELETION; + } + + public FlatDeviceDeletionEntry(DeletionPredicate predicate, TimeRange timeRange) { + super(predicate, timeRange); + // Preserve the row-scoped interpretation across plan, WAL and mods serialization. + modType = ModType.FLAT_DEVICE_DELETION; + } + + public void setTagColumnNames(List<String> names) { + tagColumnNames = names == null ? Collections.emptyList() : List.copyOf(names); + } + + public List<String> getTagColumnNames() { + return tagColumnNames; + } + + @Override + public long serialize(OutputStream stream) throws IOException { + long size = super.serialize(stream); + size += ReadWriteForEncodingUtils.writeVarInt(tagColumnNames.size(), stream); + for (String name : tagColumnNames) { + size += ReadWriteIOUtils.writeVar(name, stream); + } + return size; + } + + @Override + public long serialize(ByteBuffer buffer) { + long size = super.serialize(buffer); + size += ReadWriteForEncodingUtils.writeVarInt(tagColumnNames.size(), buffer); + for (String name : tagColumnNames) { + size += ReadWriteIOUtils.writeVar(name, buffer); + } + return size; + } + + @Override + public void deserialize(InputStream stream) throws IOException { + super.deserialize(stream); + int size = ReadWriteForEncodingUtils.readVarInt(stream); + List<String> names = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + names.add(ReadWriteIOUtils.readVarIntString(stream)); + } + tagColumnNames = names; + } + + @Override + public void deserialize(ByteBuffer buffer) { + super.deserialize(buffer); + int size = ReadWriteForEncodingUtils.readVarInt(buffer); + List<String> names = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + names.add(ReadWriteIOUtils.readVarIntString(buffer)); + } + tagColumnNames = names; + } + + @Override + public int serializedSize() { + int size = super.serializedSize() + ReadWriteForEncodingUtils.varIntSize(tagColumnNames.size()); + for (String name : tagColumnNames) { + size += ModEntry.sizeToWriteVarString(name); + } + return size; + } + + @Override + public FlatDeviceDeletionEntry clone() { + FlatDeviceDeletionEntry clone = + new FlatDeviceDeletionEntry(getPredicate(), new TimeRange(getStartTime(), getEndTime())); + clone.setTagColumnNames(tagColumnNames); + return clone; + } + + /** A physical flat-device group is addressed by table name; TAG matching is row-scoped. */ + @Override + public boolean affects(IDeviceID deviceID) { + return getTableName().equals(deviceID.getTableName()); + } + + @Override + public boolean affectsAll(IDeviceID deviceID) { + return affects(deviceID) + && getPredicate().getTagPredicateType() == TagPredicate.TagPredicateType.NOP; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TagPredicate.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TagPredicate.java index 16ac02a1ba0..6c2331f6290 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TagPredicate.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TagPredicate.java @@ -36,6 +36,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; +import java.util.AbstractList; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -95,6 +96,55 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ /** Index zero is the table name; missing trailing TAGs have the same meaning as null TAGs. */ public abstract boolean matches(Binary[] deviceSegments); + /** + * Matches TAGs in rows already scoped to the predicate's table. Entry i maps device segment i + 1 + * to a value column; missing columns and trailing TAGs are treated as null. + */ + public abstract boolean matches(Object[] values, int[] tagColumnIndices); + + private static String tagValue(Object[] values, int[] tagColumnIndices, int tagIndex) { + if (tagIndex < 0 || tagIndex >= tagColumnIndices.length) { + return null; + } + int columnIndex = tagColumnIndices[tagIndex]; + Object value = columnIndex >= 0 && columnIndex < values.length ? values[columnIndex] : null; + return value == null ? null : value.toString(); + } + + private static List<String> tagValuesKey(Object[] values, int[] tagColumnIndices) { + int length = tagColumnIndices.length; + while (length > 0 && tagValue(values, tagColumnIndices, length - 1) == null) { + length--; + } + int size = length; + // A temporary lookup view avoids copying the row's TAGs. Never retain it in the key cache. + return new AbstractList<>() { + @Override + public String get(int index) { + Objects.checkIndex(index, size); + return tagValue(values, tagColumnIndices, index); + } + + @Override + public int size() { + return size; + } + }; + } + + private static List<String> tagValuesKey(IDeviceID deviceID) { + int length = deviceID.segmentNum(); + while (length > 1 && deviceID.segment(length - 1) == null) { + length--; + } + List<String> tags = new ArrayList<>(); + for (int i = 1; i < length; i++) { + Object segment = deviceID.segment(i); + tags.add(segment == null ? null : segment.toString()); + } + return tags; + } + private static List<Binary> binaryDeviceKey(Binary[] segments) { int length = segments.length; while (length > 1 && segments[length - 1] == null) { @@ -168,6 +218,11 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ } public static class NOP extends TagPredicate { + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + return true; + } + @Override public boolean matches(Binary[] deviceSegments) { return true; @@ -222,6 +277,19 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ private IDeviceID deviceID; private volatile List<Binary> binaryDeviceID; + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + int tagCount = Math.max(deviceID.segmentNum() - 1, tagColumnIndices.length); + for (int i = 0; i < tagCount; i++) { + Object segment = i + 1 < deviceID.segmentNum() ? deviceID.segment(i + 1) : null; + if (!Objects.equals( + segment == null ? null : segment.toString(), tagValue(values, tagColumnIndices, i))) { + return false; + } + } + return true; + } + @Override public boolean matches(Binary[] deviceSegments) { if (binaryDeviceID == null) { @@ -313,6 +381,11 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ private volatile Binary binaryPattern; private int segmentIndex; + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + return Objects.equals(pattern, tagValue(values, tagColumnIndices, segmentIndex - 1)); + } + @Override public boolean matches(Binary[] deviceSegments) { if (pattern != null && binaryPattern == null) { @@ -416,6 +489,19 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(DeviceIn.class); private final Set<IDeviceID> deviceIDs = new HashSet<>(); private volatile Set<List<Binary>> binaryDeviceIDs; + private volatile Set<List<String>> stringDeviceIDs; + + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + if (stringDeviceIDs == null) { + Set<List<String>> keys = new HashSet<>(); + for (IDeviceID deviceID : deviceIDs) { + keys.add(tagValuesKey(deviceID)); + } + stringDeviceIDs = keys; + } + return stringDeviceIDs.contains(tagValuesKey(values, tagColumnIndices)); + } @Override public boolean matches(Binary[] deviceSegments) { @@ -476,6 +562,7 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(stream)); } binaryDeviceIDs = null; + stringDeviceIDs = null; } @Override @@ -485,6 +572,7 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(buffer)); } binaryDeviceIDs = null; + stringDeviceIDs = null; } @Override @@ -518,11 +606,17 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ public long ramBytesUsed() { return SHALLOW_SIZE + RamUsageEstimator.sizeOfHashSet(deviceIDs) - + RamUsageEstimator.sizeOfObject(binaryDeviceIDs); + + RamUsageEstimator.sizeOfObject(binaryDeviceIDs) + + RamUsageEstimator.sizeOfObject(stringDeviceIDs); } } public static class SegmentNotNull extends TagPredicate { + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + return tagValue(values, tagColumnIndices, segmentIndex - 1) != null; + } + @Override public boolean matches(Binary[] deviceSegments) { return segmentIndex < deviceSegments.length && deviceSegments[segmentIndex] != null; @@ -604,6 +698,16 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ } public static class And extends TagPredicate { + @Override + public boolean matches(Object[] values, int[] tagColumnIndices) { + for (TagPredicate predicate : predicates) { + if (!predicate.matches(values, tagColumnIndices)) { + return false; + } + } + return true; + } + @Override public boolean matches(Binary[] deviceSegments) { for (TagPredicate predicate : predicates) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java new file mode 100644 index 00000000000..820bc150ec6 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java @@ -0,0 +1,375 @@ +/* + * 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.memtable; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.chunk.ValueChunkWriter; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Before; +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.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class FlatDeviceMemTableDeletionTest { + private static final String DATABASE = "flat_mem_delete"; + private static final String TABLE = "t"; + private final List<PrimitiveMemTable> memTables = new ArrayList<>(); + + @Before + public void setUp() { + TsTable table = new TsTable(TABLE); + table.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + table.addColumnSchema(new TagColumnSchema("tag2", TSDataType.STRING)); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, TABLE, null); + } + + @After + public void tearDown() { + memTables.forEach(PrimitiveMemTable::release); + DataNodeTableCache.getInstance().invalid(DATABASE); + } + + // Equal timestamps and duplicate compound keys must only delete matching TAGs inside the + // inclusive time bounds, regardless of whether a query sorted the TVList before DELETE. + @Test + public void testTagAndTimeDeletionBeforeAndAfterSort() { + for (boolean sorted : new boolean[] {false, true}) { + PrimitiveMemTable memTable = newMemTable(); + insert(memTable, 2, "b", "x", 20); + insert(memTable, 1, "a", "x", 10); + insert(memTable, 1, "b", "x", 11); + insert(memTable, 1, "b", "x", 12); + insert(memTable, 0, "b", "x", 5); + AlignedTVList list = chunk(memTable).getWorkingTVList(); + if (sorted) { + list.sort(); + } + TableDeletionEntry deletion = deletion(new TagPredicate.SegmentExactMatch("b", 1), 1, 1); + assertEquals(2, memTable.delete(deletion)); + assertEquals(0, memTable.delete(deletion)); + list.sort(); + assertEquals(List.of(5L, 10L, 20L), remainingValues(list, 2)); + assertEquals(0, list.getMinTime()); + assertEquals(2, list.getMaxTime()); + } + } + + // IS NULL must remain distinct from an empty string and may be combined with another TAG. + // Deleting boundary rows must also refresh the surviving TVList time range. + @Test + public void testNullAndEmptyTags() { + PrimitiveMemTable memTable = newMemTable(); + insert(memTable, 0, "a", null, 1); + insert(memTable, 1, "a", "", 2); + insert(memTable, 2, "b", null, 3); + insert(memTable, 3, null, "x", 4); + assertEquals( + 1, + memTable.delete( + deletion( + new TagPredicate.And( + new TagPredicate.SegmentExactMatch("a", 1), + new TagPredicate.SegmentExactMatch(null, 2)), + 0, + 3))); + assertEquals(1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch(null, 1), 0, 3))); + AlignedTVList list = chunk(memTable).getWorkingTVList(); + assertEquals(List.of(2L, 3L), remainingValues(list, 2)); + assertEquals(1, list.getMinTime()); + assertEquals(2, list.getMaxTime()); + } + + // A failed leading TAG is physically absent. A later write appends it after FIELD, while a + // handed-over TVList stays narrower. Both lists must use the table's logical TAG ordinals. + @Test + public void testMissingAndLateTagsAcrossHandover() { + PrimitiveMemTable memTable = newMemTable(); + RelationalInsertRowNode first = row(1, "a", "x", 10); + first.markFailedMeasurement(0); + memTable.insertAlignedRow(first); + AlignedWritableMemChunk chunk = chunk(memTable); + chunk.handoverAlignedTvList(); + insert(memTable, 1, "b", "x", 20); + insert(memTable, 1, "a", "x", 30); + assertEquals(1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch(null, 1), 1, 1))); + assertTrue(chunk.getSortedList().get(0).isTimeDeleted(0)); + assertEquals(1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch("b", 1), 1, 1))); + chunk.getWorkingTVList().sort(); + assertEquals(List.of(30L), remainingValues(chunk.getWorkingTVList(), 1)); + } + + // A measurement-specific entry clears only the requested cells of matching rows. Keeping TAGs + // and timestamps allows another field update without changing the logical device identity. + @Test + public void testMeasurementDeletion() { + PrimitiveMemTable memTable = newMemTable(); + insert(memTable, 1, "a", "x", 10); + insert(memTable, 1, "b", "x", 20); + TableDeletionEntry deletion = + new TableDeletionEntry( + new DeletionPredicate( + TABLE, new TagPredicate.SegmentExactMatch("a", 1), List.of("field1")), + new TimeRange(1, 1)); + assertEquals(1, memTable.delete(deletion)); + assertEquals(0, memTable.delete(deletion)); + AlignedTVList list = chunk(memTable).getWorkingTVList(); + assertFalse(list.isTimeDeleted(0)); + assertTrue(list.isNullValue(0, 2)); + assertFalse(list.isNullValue(0, 0)); + assertEquals(20, list.getLongByValueIndex(1, 2)); + } + + // Time-deletion bitmaps must survive a WAL snapshot, and deleting the final logical row must + // release the group so a later write to the same compound key is visible again. + @Test + public void testDeletionWalSnapshotAndReinsert() throws Exception { + PrimitiveMemTable memTable = newMemTable(); + insert(memTable, 1, "b", "x", 20); + insert(memTable, 1, "a", "x", 10); + assertEquals(1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch("a", 1), 1, 1))); + AlignedWritableMemChunk chunk = chunk(memTable); + chunk.getWorkingTVList().sort(); + ByteBuffer buffer = ByteBuffer.allocate(chunk.serializedSize()); + chunk.serializeToWAL(new WALByteBufferForTest(buffer)); + assertFalse(buffer.hasRemaining()); + AlignedWritableMemChunk recovered = + AlignedWritableMemChunk.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + try { + recovered.getWorkingTVList().sort(); + assertEquals(List.of(20L), remainingValues(recovered.getWorkingTVList(), 2)); + } finally { + recovered.release(); + } + assertEquals(1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch("b", 1), 1, 1))); + assertTrue(memTable.getMemTableMap().isEmpty()); + insert(memTable, 1, "a", "x", 30); + assertEquals(List.of(30L), remainingValues(chunk(memTable).getWorkingTVList(), 2)); + } + + // Internal recovery removes whole physical IDs before a database/cache is available. NOP time + // deletion and ordinary table-device matching must also retain their existing behavior. + @Test + public void testPhysicalAndOrdinaryDeletionWithoutSchema() { + PrimitiveMemTable flat = newMemTable(); + insert(flat, 1, "a", "x", 10); + insert(flat, 2, "b", "x", 20); + PrimitiveMemTable ordinary = newMemTable(); + RelationalInsertRowNode row = row(1, "a", "x", 10); + row.setFlatDevice(false); + ordinary.insertAlignedRow(row); + DataNodeTableCache.getInstance().invalid(DATABASE); + flat.setDatabaseAndDataRegionId(null, "0"); + assertEquals( + 1, + flat.delete( + deletion( + new TagPredicate.FullExactMatch( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {TABLE})), + 1, + 1))); + assertEquals(1, flat.delete(deletion(new TagPredicate.NOP(), 2, 2))); + assertTrue(flat.getMemTableMap().isEmpty()); + assertEquals(1, ordinary.delete(deletion(new TagPredicate.SegmentExactMatch("a", 1), 1, 1))); + assertTrue(ordinary.getMemTableMap().isEmpty()); + } + + // Sorting changes value indices before a snapshot. Both aligned TVList layouts must serialize + // timestamps, values and deletion bits in the same order and restore the deleted-row count. + @Test + public void testSortedWalDeletionForBothLayouts() throws Exception { + for (boolean flat : new boolean[] {false, true}) { + List<IMeasurementSchema> schemas = + new ArrayList<>( + List.of( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("field1", TSDataType.INT64))); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(schemas, true, flat ? 1 : 0, flat); + try { + chunk.putAlignedRow(3, new Object[] {binary("b"), 30L}); + chunk.putAlignedRow(1, new Object[] {binary("a"), 10L}); + chunk.putAlignedRow(2, new Object[] {binary("c"), 20L}); + assertEquals(1, chunk.getWorkingTVList().deleteMatchingRows(1, 1, index -> true, null)); + chunk.getWorkingTVList().sort(); + ByteBuffer buffer = ByteBuffer.allocate(chunk.serializedSize()); + chunk.serializeToWAL(new WALByteBufferForTest(buffer)); + assertFalse(buffer.hasRemaining()); + AlignedWritableMemChunk recovered = + AlignedWritableMemChunk.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + try { + AlignedTVList list = recovered.getWorkingTVList(); + assertTrue(list.isTimeDeleted(0)); + assertEquals(List.of(20L, 30L), remainingValues(list, 1)); + assertEquals(2, list.getMinTime()); + assertEquals(3, list.getMaxTime()); + assertEquals(binary("c"), list.getBinaryByValueIndex(list.getValueIndex(1), 0)); + assertEquals(2, recovered.deleteTime(2, 3)); + assertTrue(recovered.isAllDeleted()); + } finally { + recovered.release(); + } + } finally { + chunk.release(); + } + } + } + + // A selective TAG-column deletion changes the compound key, unlike a whole-row deletion. + // The next sort must restore TAG order without clearing the other row's TAG or FIELD. + @Test + public void testSelectiveTagColumnDeletionInvalidatesSort() { + PrimitiveMemTable memTable = newMemTable(); + insert(memTable, 1, "a", "x", 10); + insert(memTable, 1, "b", "x", 20); + AlignedTVList list = chunk(memTable).getWorkingTVList(); + list.sort(); + assertEquals( + 1, + memTable.delete( + new TableDeletionEntry( + new DeletionPredicate( + TABLE, new TagPredicate.SegmentExactMatch("b", 1), List.of("tag1")), + new TimeRange(1, 1)))); + assertFalse(list.isSorted()); + list.sort(); + assertTrue(list.isNullValue(list.getValueIndex(0), 0)); + assertEquals(List.of(20L, 10L), remainingValues(list, 2)); + } + + // Deletions at the first, middle and last sorted positions must not contribute to page sizes + // or value chunks. Exercise both flush encoders, including a physically dropped TAG column. + @Test + public void testFlushSkipsDeletedRowsInEveryColumn() { + for (boolean dropColumn : new boolean[] {false, true}) { + PrimitiveMemTable memTable = newMemTable(); + for (int i = 0; i < 5; i++) { + insert(memTable, 1, String.valueOf((char) ('a' + i)), "x", i); + } + for (String tag : List.of("a", "c", "e")) { + assertEquals( + 1, memTable.delete(deletion(new TagPredicate.SegmentExactMatch(tag, 1), 1, 1))); + } + AlignedWritableMemChunk chunk = chunk(memTable); + if (dropColumn) { + chunk.removeColumn("tag2"); + } + chunk.sortTvListForFlush(); + BlockingQueue<Object> queue = new LinkedBlockingQueue<>(); + chunk.encodeWorkingAlignedTVList(queue, 2, 1); + assertEquals(1, queue.size()); + AlignedChunkWriterImpl writer = (AlignedChunkWriterImpl) queue.remove(); + assertEquals(2, writer.getTimeChunkWriter().getStatistics().getCount()); + assertEquals(dropColumn ? 2 : 3, writer.getValueChunkWriterList().size()); + for (ValueChunkWriter valueWriter : writer.getValueChunkWriterList()) { + assertEquals(2, valueWriter.getStatistics().getCount()); + } + } + } + + private PrimitiveMemTable newMemTable() { + PrimitiveMemTable memTable = new PrimitiveMemTable(DATABASE, "0"); + memTables.add(memTable); + return memTable; + } + + private static TableDeletionEntry deletion(TagPredicate predicate, long start, long end) { + return new TableDeletionEntry( + new DeletionPredicate(TABLE, predicate), new TimeRange(start, end)); + } + + private static void insert( + PrimitiveMemTable memTable, long time, String tag1, String tag2, long value) { + memTable.insertAlignedRow(row(time, tag1, tag2, value)); + } + + private static RelationalInsertRowNode row(long time, String tag1, String tag2, long value) { + RelationalInsertRowNode node = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath(TABLE, false), + true, + new String[] {"tag1", "tag2", "field1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.STRING, TSDataType.INT64}, + new MeasurementSchema[] { + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("field1", TSDataType.INT64) + }, + time, + new Object[] {binary(tag1), binary(tag2), value}, + false, + new TsTableColumnCategory[] { + TsTableColumnCategory.TAG, TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD + }); + node.setFlatDevice(true); + return node; + } + + private static Binary binary(String value) { + return value == null ? null : new Binary(value, TSFileConfig.STRING_CHARSET); + } + + private static AlignedWritableMemChunk chunk(PrimitiveMemTable memTable) { + return ((AlignedWritableMemChunkGroup) memTable.getMemTableMap().values().iterator().next()) + .getAlignedMemChunk(); + } + + private static List<Long> remainingValues(AlignedTVList list, int fieldIndex) { + List<Long> values = new ArrayList<>(); + for (int i = 0; i < list.rowCount(); i++) { + if (!list.isTimeDeleted(i)) { + values.add(list.getLongByValueIndex(list.getValueIndex(i), fieldIndex)); + } + } + return values; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java index 426fa0b7354..9d67c640acf 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java @@ -162,6 +162,74 @@ public class TableDeletionEntryTest { } } + @Test + public void testMappedTagValuesMatching() throws IOException { + IDeviceID device = Factory.DEFAULT_FACTORY.create(new String[] {"t", "a", ""}); + List<TagPredicate> predicates = + Arrays.asList( + new NOP(), + new FullExactMatch(device), + new FullExactMatch(Factory.DEFAULT_FACTORY.create(new String[] {"t"})), + new DeviceIn( + Arrays.asList(device, Factory.DEFAULT_FACTORY.create(new String[] {"t", "b"}))), + new SegmentExactMatch("a", 1), + new SegmentExactMatch("", 2), + new SegmentExactMatch(null, 3), + new TagPredicate.SegmentNotNull(2), + new And(new SegmentExactMatch("a", 1), new SegmentExactMatch(null, 3))); + // TAG order differs from value-column order; include absent columns and trailing null TAGs. + int[][] mappings = {{2, 0}, {2, 0, -1}, {2, 4}, {-1, 0}, {2, 0, 1}, {}}; + Object[] values = {new Binary("", TSFileConfig.STRING_CHARSET), "extra", "a", 42L}; + for (TagPredicate predicate : predicates) { + DeletionPredicate deletion = new DeletionPredicate("t", predicate); + for (int[] mapping : mappings) { + String[] segments = new String[mapping.length + 1]; + segments[0] = "t"; + for (int i = 0; i < mapping.length; i++) { + int column = mapping[i]; + segments[i + 1] = + column >= 0 && column < values.length ? values[column].toString() : null; + } + boolean expected = deletion.matches(Factory.DEFAULT_FACTORY.create(segments)); + assertEquals(expected, predicate.matches(values, mapping)); + assertEquals(expected, deletion.matches(values, mapping)); + + ByteBuffer buffer = ByteBuffer.allocate(deletion.serializedSize()); + deletion.serialize(buffer); + assertEquals(deletion.serializedSize(), buffer.position()); + buffer.flip(); + DeletionPredicate restored = new DeletionPredicate(); + restored.deserialize(buffer); + assertEquals(expected, restored.matches(values, mapping)); + + ByteArrayOutputStream output = new ByteArrayOutputStream(); + deletion.serialize(output); + restored.deserialize(new ByteArrayInputStream(output.toByteArray())); + assertEquals(expected, restored.matches(values, mapping)); + } + } + } + + @Test + public void testReusableTagValues() { + IDeviceID device = Factory.DEFAULT_FACTORY.create(new String[] {"t", "a"}); + Object[] values = {null, 42L}; + int[] mapping = {0, -1}; + for (TagPredicate predicate : + Arrays.asList(new FullExactMatch(device), new DeviceIn(Collections.singleton(device)))) { + values[0] = "a"; + assertTrue(predicate.matches(values, mapping)); + values[0] = null; + assertFalse(predicate.matches(values, mapping)); + values[0] = new Binary("a", TSFileConfig.STRING_CHARSET); + assertTrue(predicate.matches(values, mapping)); + mapping[0] = 1; + assertFalse(predicate.matches(values, mapping)); + mapping[0] = 0; + assertTrue(predicate.matches(values, mapping)); + } + } + @Test public void testSerialization() throws IOException { TableDeletionEntry entry =
