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 <[email protected]> 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
