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/07/01 04:14:20 UTC
(iotdb) 03/08: add DEVICE_IN filter
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch delete_with_attribute
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1ebbd84d8c0e7c0606ad909e779c3bd64d4b08e3
Author: Tian Jiang <jt...@163.com>
AuthorDate: Tue Jun 30 18:21:20 2026 +0800
add DEVICE_IN filter
---
.../dataregion/modification/TagPredicate.java | 103 ++++++++++++++++++++-
.../modification/ModificationFileTest.java | 21 ++++-
.../modification/TableDeletionEntryTest.java | 38 ++++++++
3 files changed, 159 insertions(+), 3 deletions(-)
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 5af4a9b95fe..3a6f0a6ec35 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
@@ -35,9 +35,12 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
public abstract class TagPredicate implements StreamSerializable, BufferSerializable, Accountable {
@@ -51,7 +54,8 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ
NOP,
FULL_EXACT_MATCH,
SEGMENT_EXACT_MATCH,
- AND;
+ AND,
+ DEVICE_IN;
public long serialize(OutputStream stream) throws IOException {
stream.write((byte) ordinal());
@@ -105,6 +109,8 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ
predicate = new SegmentExactMatch();
} else if (Objects.requireNonNull(type) == TagPredicateType.AND) {
predicate = new And();
+ } else if (Objects.requireNonNull(type) == TagPredicateType.DEVICE_IN) {
+ predicate = new DeviceIn();
} else {
throw new IllegalArgumentException(StorageEngineMessages.UNRECOGNIZED_PREDICATE_TYPE + type);
}
@@ -123,6 +129,8 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ
predicate = new SegmentExactMatch();
} else if (Objects.requireNonNull(type) == TagPredicateType.AND) {
predicate = new And();
+ } else if (Objects.requireNonNull(type) == TagPredicateType.DEVICE_IN) {
+ predicate = new DeviceIn();
} else {
throw new IllegalArgumentException(StorageEngineMessages.UNRECOGNIZED_PREDICATE_TYPE + type);
}
@@ -341,6 +349,99 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ
}
}
+ public static class DeviceIn extends TagPredicate {
+
+ public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(DeviceIn.class);
+ private final Set<IDeviceID> deviceIDs = new HashSet<>();
+
+ public DeviceIn(Collection<IDeviceID> deviceIDs) {
+ super(TagPredicateType.DEVICE_IN);
+ this.deviceIDs.addAll(deviceIDs);
+ }
+
+ public DeviceIn() {
+ super(TagPredicateType.DEVICE_IN);
+ }
+
+ @Override
+ public int serializedSize() {
+ int serializedSize = super.serializedSize();
+ serializedSize += ReadWriteForEncodingUtils.varIntSize(deviceIDs.size());
+ for (IDeviceID deviceID : deviceIDs) {
+ serializedSize += deviceID.serializedSize();
+ }
+ return serializedSize;
+ }
+
+ @Override
+ public long serialize(OutputStream stream) throws IOException {
+ long size = super.serialize(stream);
+ size += ReadWriteForEncodingUtils.writeVarInt(deviceIDs.size(), stream);
+ for (IDeviceID deviceID : deviceIDs) {
+ size += deviceID.serialize(stream);
+ }
+ return size;
+ }
+
+ @Override
+ public long serialize(ByteBuffer buffer) {
+ long size = super.serialize(buffer);
+ size += ReadWriteForEncodingUtils.writeVarInt(deviceIDs.size(), buffer);
+ for (IDeviceID deviceID : deviceIDs) {
+ size += deviceID.serialize(buffer);
+ }
+ return size;
+ }
+
+ @Override
+ public void deserialize(InputStream stream) throws IOException {
+ int size = ReadWriteForEncodingUtils.readVarInt(stream);
+ for (int i = 0; i < size; i++) {
+ deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(stream));
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ int size = ReadWriteForEncodingUtils.readVarInt(buffer);
+ for (int i = 0; i < size; i++) {
+ deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(buffer));
+ }
+ }
+
+ @Override
+ public boolean matches(IDeviceID deviceID) {
+ return deviceIDs.contains(deviceID);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DeviceIn deviceIn = (DeviceIn) o;
+ return Objects.equals(deviceIDs, deviceIn.deviceIDs);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(deviceIDs);
+ }
+
+ @Override
+ public String toString() {
+ return "DeviceIn{" + "deviceIDs=" + deviceIDs + '}';
+ }
+
+ @Override
+ public long ramBytesUsed() {
+ return SHALLOW_SIZE + RamUsageEstimator.sizeOfHashSet(deviceIDs);
+ }
+ }
+
public static class And extends TagPredicate {
public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(And.class);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModificationFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModificationFileTest.java
index 65abd59e293..58fe5e90245 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModificationFileTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModificationFileTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.service.metrics.FileMetrics;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.recover.CompactionRecoverManager;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.DeviceIn;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.SegmentExactMatch;
@@ -131,6 +132,14 @@ public class ModificationFileTest {
new FullExactMatch(Factory.DEFAULT_FACTORY.create(new String[] {"id1", "id2"}))),
new TimeRange(5, 6)),
new TableDeletionEntry(new DeletionPredicate("table4"), new TimeRange(7, 8)),
+ new TableDeletionEntry(
+ new DeletionPredicate(
+ "table5",
+ new DeviceIn(
+ Arrays.asList(
+ Factory.DEFAULT_FACTORY.create(new String[] {"table5", "id1", "id2"}),
+ Factory.DEFAULT_FACTORY.create(new String[] {"table5", "id3", "id4"})))),
+ new TimeRange(9, 10)),
};
try (ModificationFile mFile = new ModificationFile(tempFileName, false)) {
for (int i = 0; i < 4; i++) {
@@ -141,11 +150,11 @@ public class ModificationFileTest {
assertEquals(modifications[i], modificationList.get(i));
}
- for (int i = 4; i < 8; i++) {
+ for (int i = 4; i < modifications.length; i++) {
mFile.write(modifications[i]);
}
modificationList = mFile.getAllMods();
- for (int i = 0; i < 8; i++) {
+ for (int i = 0; i < modifications.length; i++) {
assertEquals(modifications[i], modificationList.get(i));
}
} catch (IOException e) {
@@ -172,6 +181,14 @@ public class ModificationFileTest {
new FullExactMatch(Factory.DEFAULT_FACTORY.create(new String[] {"id1", "id2"}))),
new TimeRange(5, 6)),
new TableDeletionEntry(new DeletionPredicate("table4"), new TimeRange(7, 8)),
+ new TableDeletionEntry(
+ new DeletionPredicate(
+ "table5",
+ new DeviceIn(
+ Arrays.asList(
+ Factory.DEFAULT_FACTORY.create(new String[] {"table5", "id1", "id2"}),
+ Factory.DEFAULT_FACTORY.create(new String[] {"table5", "id3", "id4"})))),
+ new TimeRange(9, 10)),
};
try (ModificationFile mFile = new ModificationFile(tempFileName, false)) {
mFile.write(Arrays.asList(modifications));
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 315432c03a1..99a889e2e5f 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
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.modification;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.And;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.DeviceIn;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP;
import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.SegmentExactMatch;
@@ -72,6 +73,27 @@ public class TableDeletionEntryTest {
() -> TagPredicate.TagPredicateType.deserialize(new ByteArrayInputStream(new byte[0])));
}
+ @Test
+ public void testDeviceInSerialization() throws IOException {
+ DeviceIn predicate =
+ new DeviceIn(
+ Arrays.asList(
+ Factory.DEFAULT_FACTORY.create(new String[] {"table1", "id1", "id2"}),
+ Factory.DEFAULT_FACTORY.create(new String[] {"table1", "id3", "id4"})));
+
+ ByteBuffer buffer = ByteBuffer.allocate(predicate.serializedSize());
+ predicate.serialize(buffer);
+ assertEquals(predicate.serializedSize(), buffer.position());
+ buffer.flip();
+ assertEquals(predicate, TagPredicate.createFrom(buffer));
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ predicate.serialize(bos);
+ byte[] byteArray = bos.toByteArray();
+ assertEquals(predicate.serializedSize(), byteArray.length);
+ assertEquals(predicate, TagPredicate.createFrom(new ByteArrayInputStream(byteArray)));
+ }
+
@Test
public void testAffectDevice() {
TableDeletionEntry entry1 =
@@ -92,6 +114,15 @@ public class TableDeletionEntryTest {
"table1",
new And(new SegmentExactMatch("id1", 1), new SegmentExactMatch("id2", 2))),
new TimeRange(1, 5));
+ TableDeletionEntry entry5 =
+ new TableDeletionEntry(
+ new DeletionPredicate(
+ "table1",
+ new DeviceIn(
+ Arrays.asList(
+ Factory.DEFAULT_FACTORY.create(new String[] {"table1", "id1", "id2"}),
+ Factory.DEFAULT_FACTORY.create(new String[] {"table1", "id1"})))),
+ new TimeRange(1, 5));
IDeviceID deviceID1 = Factory.DEFAULT_FACTORY.create(new String[] {"table1", "id1", "id2"});
IDeviceID deviceID2 =
@@ -128,6 +159,13 @@ public class TableDeletionEntryTest {
assertFalse(entry4.affects(deviceID4));
assertFalse(entry4.affects(deviceID5));
assertFalse(entry4.affects(deviceID6));
+
+ assertTrue(entry5.affects(deviceID1));
+ assertFalse(entry5.affects(deviceID2));
+ assertFalse(entry5.affects(deviceID3));
+ assertTrue(entry5.affects(deviceID4));
+ assertFalse(entry5.affects(deviceID5));
+ assertFalse(entry5.affects(deviceID6));
}
@Test