You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 06:50:47 UTC
(iotdb) 12/38: support flat_device deletion
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 <jt...@163.com>
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 =