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 96bc4165baf05196d3a4eff95fe7de5f9a55b54b Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 9 18:06:04 2026 +0800 tmp --- .../execution/operator/source/SeriesScanUtil.java | 33 ++++ .../dataregion/memtable/TsFileProcessor.java | 96 ++++++----- .../dataregion/modification/DeletionPredicate.java | 15 ++ .../dataregion/modification/ModEntry.java | 48 +++--- .../modification/TableDeletionEntry.java | 5 + .../dataregion/modification/TagPredicate.java | 94 ++++++++++- .../read/reader/chunk/DiskAlignedChunkLoader.java | 4 + .../apache/iotdb/db/utils/ModificationUtils.java | 10 ++ .../dataregion/memtable/TsFileProcessorTest.java | 186 +++++++++++++++++++++ .../modification/TableDeletionEntryTest.java | 119 +++++++++++++ 10 files changed, 540 insertions(+), 70 deletions(-) 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 c33a27e0db0..7729b866a53 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.db.queryengine.execution.operator.source; import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; +import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.db.exception.CorruptedTsFileException; @@ -32,6 +33,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOpt import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.TagPredicateType; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskAlignedChunkLoader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskChunkLoader; @@ -110,6 +113,7 @@ public class SeriesScanUtil implements Accountable { private final IDeviceID deviceID; protected boolean isAligned = false; + protected int[] rowKeyColumnIndices = new int[0]; private final TSDataType dataType; // inner class of SeriesReader for order purpose @@ -721,6 +725,12 @@ public class SeriesScanUtil implements Accountable { FileLoaderUtils.loadPageReaderList( chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList(), context); + // Flat-device TAG deletions are row-scoped and must be evaluated while decoding each page. + if (rowKeyColumnIndices.length > 0 + && chunkLoader instanceof DiskAlignedChunkLoader diskAlignedLoader) { + addFlatDeviceDeletionFilter(pageReaderList, diskAlignedLoader.getResource()); + } + // init TsBlockBuilder for each page reader pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); @@ -773,6 +783,29 @@ public class SeriesScanUtil implements Accountable { } } + private void addFlatDeviceDeletionFilter( + List<IPageReader> pageReaderList, TsFileResource resource) { + List<FlatDeviceDeletionEntry> deletions; + try { + deletions = + context.getPathModifications(resource, seriesPath.getDeviceId()).stream() + .filter( + mod -> + mod instanceof FlatDeviceDeletionEntry deletion + && deletion.getPredicate().getTagPredicateType() != TagPredicateType.NOP) + .map(mod -> (FlatDeviceDeletionEntry) mod) + .toList(); + } catch (IllegalPathException e) { + throw new IllegalStateException(e); + } + if (!deletions.isEmpty()) { + Filter filter = + new FlatDeviceDeletionFilter( + seriesPath.getDeviceId().getTableName(), rowKeyColumnIndices, deletions); + pageReaderList.forEach(pageReader -> pageReader.addRecordFilter(filter)); + } + } + private void unpackOneFakeMemChunkMetaData( IChunkMetadata chunkMetaData, MemChunkLoader chunkLoader, long timestampInFileName) { ReadOnlyMemChunk readOnlyMemChunk = chunkLoader.getReadOnlyMemChunk(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 809e696ade9..4bc4c0d007b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -398,14 +398,11 @@ public class TsFileProcessor { reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); } - // Update start time of this memtable - IDeviceID memTableDeviceID = insertRowNode.getMemTableDeviceID(insertRowNode.getDeviceID()); - tsFileResource.updateStartTime(memTableDeviceID, insertRowNode.getTime()); - // For sequence tsfile, we update the endTime only when the file is prepared to be closed. - // For unsequence tsfile, we have to update the endTime for each insertion. - if (!sequence) { - tsFileResource.updateEndTime(memTableDeviceID, insertRowNode.getTime()); - } + updateResourceTime( + insertRowNode.getDeviceID(), + insertRowNode.getTime(), + insertRowNode.getTime(), + insertRowNode.isFlatDevice()); tsFileResource.updateProgressIndex(insertRowNode.getProgressIndex()); // RecordScheduleMemTableCost @@ -414,6 +411,16 @@ public class TsFileProcessor { infoForMetrics[4] += pointInserted; } + private void updateResourceTime( + IDeviceID deviceID, long startTime, long endTime, boolean flatDevice) { + tsFileResource.updateStartTime(deviceID, startTime); + // A flat-device MemTable only retains physical device groups, so capture logical end times + // while the insert plan still provides each device's range, even for sequence files. + if (!sequence || flatDevice) { + tsFileResource.updateEndTime(deviceID, endTime); + } + } + public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) throws WriteProcessException { @@ -519,14 +526,11 @@ public class TsFileProcessor { } else { pointInserted += workMemTable.insert(insertRowNode); } - // update start time of this memtable - IDeviceID memTableDeviceID = insertRowNode.getMemTableDeviceID(insertRowNode.getDeviceID()); - tsFileResource.updateStartTime(memTableDeviceID, insertRowNode.getTime()); - // for sequence tsfile, we update the endTime only when the file is prepared to be closed. - // for unsequence tsfile, we have to update the endTime for each insertion. - if (!sequence) { - tsFileResource.updateEndTime(memTableDeviceID, insertRowNode.getTime()); - } + updateResourceTime( + insertRowNode.getDeviceID(), + insertRowNode.getTime(), + insertRowNode.getTime(), + insertRowNode.isFlatDevice()); } } finally { reconcileAlignedTVListRamCost(alignedRamCostSnapshot, alignedMemTableIncrement); @@ -738,26 +742,14 @@ public class TsFileProcessor { final List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs = insertTabletNode.splitByDevice(start, end); - tsFileResource.updateStartTime( - insertTabletNode.getMemTableDeviceID(deviceEndOffsetPairs.get(0).left), - insertTabletNode.getTimes()[start]); - if (!sequence) { - // For sequence tsfile, we update the endTime only when the file is prepared to be closed. - // For unsequence tsfile, we have to update the endTime for each insertion. - tsFileResource.updateEndTime( - insertTabletNode.getMemTableDeviceID(deviceEndOffsetPairs.get(0).left), - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(0).right - 1]); - } - for (int i = 1; i < deviceEndOffsetPairs.size(); i++) { - // the end offset of i - 1 is the start offset of i - tsFileResource.updateStartTime( - insertTabletNode.getMemTableDeviceID(deviceEndOffsetPairs.get(i).left), - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i - 1).right]); - if (!sequence) { - tsFileResource.updateEndTime( - insertTabletNode.getMemTableDeviceID(deviceEndOffsetPairs.get(i).left), - insertTabletNode.getTimes()[deviceEndOffsetPairs.get(i).right - 1]); - } + int deviceStartOffset = start; + for (Pair<IDeviceID, Integer> deviceEndOffset : deviceEndOffsetPairs) { + updateResourceTime( + deviceEndOffset.left, + insertTabletNode.getTimes()[deviceStartOffset], + insertTabletNode.getTimes()[deviceEndOffset.right - 1], + insertTabletNode.isFlatDevice()); + deviceStartOffset = deviceEndOffset.right; } } } finally { @@ -1775,16 +1767,42 @@ public class TsFileProcessor { */ private Future<?> addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOException { final Map<IDeviceID, Long> lastTimeForEachDevice = tobeFlushed.getMaxTime(); + final Map<IDeviceID, Long> resourceEndTimes = new HashMap<>(lastTimeForEachDevice); + final Set<String> flatDeviceTables = new HashSet<>(); + // Flat-device groups aggregate several logical devices. Their physical end times are only + // used by the flush-time callback; resource times were recorded with logical IDs on insertion. + tobeFlushed + .getMemTableMap() + .forEach( + (deviceID, group) -> { + if (group instanceof AlignedWritableMemChunkGroup alignedGroup + && alignedGroup.getAlignedMemChunk().isFlatDevice()) { + if (resourceEndTimes.remove(deviceID) != null) { + flatDeviceTables.add(deviceID.getTableName()); + } + } + }); + // Each file is flushed once. Only flat-device groups still in this MemTable need their + // logical end times copied from the resource; other devices use the MemTable's end times. + if (!flatDeviceTables.isEmpty()) { + for (IDeviceID deviceID : tsFileResource.getDevices()) { + if (flatDeviceTables.contains(deviceID.getTableName())) { + tsFileResource + .getEndTime(deviceID) + .ifPresent(endTime -> resourceEndTimes.put(deviceID, endTime)); + } + } + } // If some devices have been removed in MemTable, the number of device in MemTable and // tsFileResource will not be the same. And the endTime of these devices in resource will be // Long.minValue. // In the case, we need to delete the removed devices in tsFileResource. - if (lastTimeForEachDevice.size() != tsFileResource.getDevices().size()) { - tsFileResource.deleteRemovedDeviceAndUpdateEndTime(lastTimeForEachDevice); + if (resourceEndTimes.size() != tsFileResource.getDevices().size()) { + tsFileResource.deleteRemovedDeviceAndUpdateEndTime(resourceEndTimes); } else { if (sequence) { - tsFileResource.updateEndTime(lastTimeForEachDevice); + tsFileResource.updateEndTime(resourceEndTimes); } } 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 b8722f6b414..7a6ec6f15f8 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 @@ -22,8 +22,10 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NO import org.apache.iotdb.db.utils.io.BufferSerializable; import org.apache.iotdb.db.utils.io.StreamSerializable; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.utils.Accountable; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; @@ -42,6 +44,7 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(DeletionPredicate.class); private String tableName; + private volatile Binary binaryTableName; private TagPredicate tagPredicate = new NOP(); // an empty list means affecting all columns private List<String> measurementNames = Collections.emptyList(); @@ -68,6 +71,15 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable return tableName.equals(deviceID.getTableName()) && tagPredicate.matches(deviceID); } + public boolean matches(Binary[] deviceSegments) { + if (binaryTableName == null) { + binaryTableName = new Binary(tableName, TSFileConfig.STRING_CHARSET); + } + return deviceSegments.length > 0 + && binaryTableName.equals(deviceSegments[0]) + && tagPredicate.matches(deviceSegments); + } + public void setTagPredicate(TagPredicate tagPredicate) { this.tagPredicate = tagPredicate; } @@ -117,6 +129,7 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable @Override public void deserialize(InputStream stream) throws IOException { tableName = ReadWriteIOUtils.readVarIntString(stream); + binaryTableName = null; tagPredicate = TagPredicate.createFrom(stream); int measurementLength = ReadWriteForEncodingUtils.readVarInt(stream); @@ -133,6 +146,7 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable @Override public void deserialize(ByteBuffer buffer) { tableName = ReadWriteIOUtils.readVarIntString(buffer); + binaryTableName = null; tagPredicate = TagPredicate.createFrom(buffer); int measurementLength = ReadWriteForEncodingUtils.readVarInt(buffer); @@ -194,6 +208,7 @@ public class DeletionPredicate implements StreamSerializable, BufferSerializable public long ramBytesUsed() { return SHALLOW_SIZE + RamUsageEstimator.sizeOf(tableName) + + RamUsageEstimator.sizeOfObject(binaryTableName) + RamUsageEstimator.sizeOfObject(tagPredicate) + RamUsageEstimator.sizeOfArrayList(measurementNames); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModEntry.java index 5e9ba381c30..3f8fb53ed31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/ModEntry.java @@ -146,6 +146,7 @@ public abstract class ModEntry public enum ModType { TABLE_DELETION((byte) 0x00), + FLAT_DEVICE_DELETION((byte) 0x02), TREE_DELETION((byte) 0x01); private final byte typeNum; @@ -163,30 +164,22 @@ public abstract class ModEntry } public ModEntry newEntry() { - ModEntry entry; - switch (this) { - case TREE_DELETION: - entry = new TreeDeletionEntry(); - break; - case TABLE_DELETION: - entry = new TableDeletionEntry(); - break; - default: - throw new IllegalArgumentException(StorageEngineMessages.UNSUPPORTED_MOD_TYPE + this); - } - return entry; + return switch (this) { + case TREE_DELETION -> new TreeDeletionEntry(); + case TABLE_DELETION -> new TableDeletionEntry(); + case FLAT_DEVICE_DELETION -> new FlatDeviceDeletionEntry(); + }; } public static ModType deserialize(ByteBuffer buffer) { byte typeNum = buffer.get(); - switch (typeNum) { - case 0x00: - return TABLE_DELETION; - case 0x01: - return TREE_DELETION; - default: - throw new IllegalArgumentException(StorageEngineMessages.UNKNOWN_MOD_TYPE + typeNum); - } + return switch (typeNum) { + case 0x00 -> TABLE_DELETION; + case 0x01 -> TREE_DELETION; + case 0x02 -> FLAT_DEVICE_DELETION; + default -> + throw new IllegalArgumentException(StorageEngineMessages.UNKNOWN_MOD_TYPE + typeNum); + }; } public static ModType deserialize(InputStream stream) throws IOException { @@ -197,14 +190,13 @@ public abstract class ModEntry if (typeNum == -1) { throw new EOFException(); } - switch (typeNum) { - case 0x00: - return TABLE_DELETION; - case 0x01: - return TREE_DELETION; - default: - throw new IllegalArgumentException(StorageEngineMessages.UNKNOWN_MOD_TYPE + typeNum); - } + return switch (typeNum) { + case 0x00 -> TABLE_DELETION; + case 0x01 -> TREE_DELETION; + case 0x02 -> FLAT_DEVICE_DELETION; + default -> + throw new IllegalArgumentException(StorageEngineMessages.UNKNOWN_MOD_TYPE + typeNum); + }; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntry.java index b2c7ccee4e5..f4240a10a7f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntry.java @@ -26,6 +26,7 @@ import org.apache.iotdb.db.utils.ModificationUtils; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; import java.io.IOException; @@ -99,6 +100,10 @@ public class TableDeletionEntry extends ModEntry { return predicate.matches(deviceID); } + public boolean affects(Binary[] deviceSegments) { + return predicate.matches(deviceSegments); + } + @Override public boolean affects(String measurementID) { return predicate.affects(measurementID); 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 8d41d23ca06..16ac02a1ba0 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 @@ -22,9 +22,11 @@ import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.utils.io.BufferSerializable; import org.apache.iotdb.db.utils.io.StreamSerializable; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.IDeviceID.Deserializer; import org.apache.tsfile.utils.Accountable; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; @@ -35,6 +37,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashSet; @@ -89,6 +92,27 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ public abstract boolean matches(IDeviceID deviceID); + /** Index zero is the table name; missing trailing TAGs have the same meaning as null TAGs. */ + public abstract boolean matches(Binary[] deviceSegments); + + private static List<Binary> binaryDeviceKey(Binary[] segments) { + int length = segments.length; + while (length > 1 && segments[length - 1] == null) { + length--; + } + return Arrays.asList(segments).subList(0, length); + } + + private static List<Binary> binaryDeviceKey(IDeviceID deviceID) { + Binary[] segments = new Binary[deviceID.segmentNum()]; + for (int i = 0; i < segments.length; i++) { + Object segment = deviceID.segment(i); + segments[i] = + segment == null ? null : new Binary(segment.toString(), TSFileConfig.STRING_CHARSET); + } + return binaryDeviceKey(segments); + } + @Override public long serialize(OutputStream stream) throws IOException { return type.serialize(stream); @@ -144,6 +168,11 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ } public static class NOP extends TagPredicate { + @Override + public boolean matches(Binary[] deviceSegments) { + return true; + } + public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(NOP.class); public NOP() { @@ -191,6 +220,15 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(FullExactMatch.class); private IDeviceID deviceID; + private volatile List<Binary> binaryDeviceID; + + @Override + public boolean matches(Binary[] deviceSegments) { + if (binaryDeviceID == null) { + binaryDeviceID = binaryDeviceKey(deviceID); + } + return binaryDeviceID.equals(binaryDeviceKey(deviceSegments)); + } public FullExactMatch(IDeviceID deviceID) { super(TagPredicateType.FULL_EXACT_MATCH); @@ -223,11 +261,13 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ @Override public void deserialize(InputStream stream) throws IOException { deviceID = Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(stream); + binaryDeviceID = null; } @Override public void deserialize(ByteBuffer buffer) { deviceID = Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(buffer); + binaryDeviceID = null; } @Override @@ -259,7 +299,9 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ @Override public long ramBytesUsed() { - return SHALLOW_SIZE + RamUsageEstimator.sizeOfObject(deviceID); + return SHALLOW_SIZE + + RamUsageEstimator.sizeOfObject(deviceID) + + RamUsageEstimator.sizeOfObject(binaryDeviceID); } } @@ -268,8 +310,19 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(SegmentExactMatch.class); private String pattern; + private volatile Binary binaryPattern; private int segmentIndex; + @Override + public boolean matches(Binary[] deviceSegments) { + if (pattern != null && binaryPattern == null) { + binaryPattern = new Binary(pattern, TSFileConfig.STRING_CHARSET); + } + return Objects.equals( + binaryPattern, + segmentIndex < deviceSegments.length ? deviceSegments[segmentIndex] : null); + } + public SegmentExactMatch(String pattern, int segmentIndex) { super(TagPredicateType.SEGMENT_EXACT_MATCH); this.pattern = pattern; @@ -306,12 +359,14 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ @Override public void deserialize(InputStream stream) throws IOException { pattern = ReadWriteIOUtils.readVarIntString(stream); + binaryPattern = null; segmentIndex = ReadWriteForEncodingUtils.readVarInt(stream); } @Override public void deserialize(ByteBuffer buffer) { pattern = ReadWriteIOUtils.readVarIntString(buffer); + binaryPattern = null; segmentIndex = ReadWriteForEncodingUtils.readVarInt(buffer); } @@ -350,7 +405,9 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ @Override public long ramBytesUsed() { - return SHALLOW_SIZE + RamUsageEstimator.sizeOf(pattern); + return SHALLOW_SIZE + + RamUsageEstimator.sizeOf(pattern) + + RamUsageEstimator.sizeOfObject(binaryPattern); } } @@ -358,6 +415,20 @@ 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; + + @Override + public boolean matches(Binary[] deviceSegments) { + if (binaryDeviceIDs == null) { + Set<List<Binary>> keys = new HashSet<>(); + for (IDeviceID deviceID : deviceIDs) { + keys.add(binaryDeviceKey(deviceID)); + } + binaryDeviceIDs = keys; + } + // The lookup view is never retained: callers may reuse the row's segments array. + return binaryDeviceIDs.contains(binaryDeviceKey(deviceSegments)); + } public DeviceIn(Collection<IDeviceID> deviceIDs) { super(TagPredicateType.DEVICE_IN); @@ -404,6 +475,7 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ for (int i = 0; i < size; i++) { deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(stream)); } + binaryDeviceIDs = null; } @Override @@ -412,6 +484,7 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ for (int i = 0; i < size; i++) { deviceIDs.add(Deserializer.DEFAULT_DESERIALIZER.deserializeFrom(buffer)); } + binaryDeviceIDs = null; } @Override @@ -443,11 +516,17 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ @Override public long ramBytesUsed() { - return SHALLOW_SIZE + RamUsageEstimator.sizeOfHashSet(deviceIDs); + return SHALLOW_SIZE + + RamUsageEstimator.sizeOfHashSet(deviceIDs) + + RamUsageEstimator.sizeOfObject(binaryDeviceIDs); } } public static class SegmentNotNull extends TagPredicate { + @Override + public boolean matches(Binary[] deviceSegments) { + return segmentIndex < deviceSegments.length && deviceSegments[segmentIndex] != null; + } public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(SegmentNotNull.class); @@ -525,6 +604,15 @@ public abstract class TagPredicate implements StreamSerializable, BufferSerializ } public static class And extends TagPredicate { + @Override + public boolean matches(Binary[] deviceSegments) { + for (TagPredicate predicate : predicates) { + if (!predicate.matches(deviceSegments)) { + return false; + } + } + return true; + } public static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(And.class); private final List<TagPredicate> predicates = new ArrayList<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java index 5af55250285..ceee05ba807 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java @@ -144,6 +144,10 @@ public class DiskAlignedChunkLoader implements IChunkLoader { return resource.getTsFileID(); } + public TsFileResource getResource() { + return resource; + } + public File getTsFile() { return resource.getTsFile(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java index 0ea1abee866..9afd8a189aa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java @@ -27,8 +27,10 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionPathUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.SettleSelectorImpl; import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; 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.modification.TreeDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory; @@ -323,6 +325,14 @@ public class ModificationUtils { } private static void doModifyChunkMetaData(ModEntry modification, IChunkMetadata metaData) { + // A flat-device deletion with a tag predicate is row-scoped. Chunk metadata does not carry + // per-row tag values, so applying its time range here would incorrectly hide rows belonging + // to other logical devices sharing this physical flat-device chunk. Such deletions are + // applied while decoding page data instead. + if (modification instanceof FlatDeviceDeletionEntry deletion + && deletion.getPredicate().getTagPredicateType() != TagPredicate.TagPredicateType.NOP) { + return; + } metaData.insertIntoSortedDeletions(modification.getTimeRange()); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index e1b402a64dd..f8c7375b036 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -35,13 +35,18 @@ import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest; +import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; @@ -51,9 +56,11 @@ import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.read.TimeValuePair; @@ -61,7 +68,9 @@ import org.apache.tsfile.read.TsFileReader; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Path; import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.filter.factory.TimeFilterApi; import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.Binary; @@ -83,9 +92,13 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import static junit.framework.TestCase.assertTrue; import static org.apache.iotdb.db.storageengine.dataregion.DataRegionTest.buildInsertRowNodeByTSRecord; @@ -1475,6 +1488,179 @@ public class TsFileProcessorTest { assertTrue(processor.getTsFileResource().isClosed()); } + @Test + public void testFlatDeviceSequenceResourceTimes() throws Exception { + assertFlatDeviceResourceTimes(true); + } + + @Test + public void testFlatDeviceUnsequenceResourceTimes() throws Exception { + assertFlatDeviceResourceTimes(false); + } + + private void assertFlatDeviceResourceTimes(boolean sequence) throws Exception { + Map<IDeviceID, Long> flushedTimes = new HashMap<>(); + processor = + new TsFileProcessor( + storageGroup, + new File(filePath), + sgInfo, + this::closeTsFileProcessor, + (tsFileProcessor, updateMap, systemFlushTime) -> flushedTimes.putAll(updateMap), + sequence); + processor.setTsFileProcessorInfo(new TsFileProcessorInfo(sgInfo)); + sgInfo.initTsFileProcessorInfo(processor); + SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor); + MeasurementSchema[] schemas = { + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64) + }; + processor.registerToTsFile( + "flat_table", + table -> + new TableSchema( + table, Arrays.asList(schemas), List.of(ColumnCategory.TAG, ColumnCategory.FIELD)), + true); + + // Different logical devices share one MemTable group, but keep independent resource ranges + // for all three insertion APIs. Tablet ranges exclude the first and last input rows. + RelationalInsertRowNode first = flatDeviceRow(10, "d1", schemas); + RelationalInsertRowNode second = flatDeviceRow(30, "d2", schemas); + processor.insert(first, new long[5]); + processor.insert(second, new long[5]); + TsFileResource resource = processor.getTsFileResource(); + assertResourceTimes(resource, first.getDeviceID(), 10, 10); + assertResourceTimes(resource, second.getDeviceID(), 30, 30); + processor.insertRows( + new RelationalInsertRowsNode( + new PlanNodeId("rows"), + List.of(0, 1), + List.of(flatDeviceRow(20, "d1", schemas), flatDeviceRow(40, "d2", schemas))), + new long[5]); + assertResourceTimes(resource, first.getDeviceID(), 10, 20); + assertResourceTimes(resource, second.getDeviceID(), 30, 40); + RelationalInsertTabletNode tablet = + new RelationalInsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath("flat_table", false), + true, + new String[] {"tag1", "s1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + schemas, + new long[] {1, 50, 60, 70, 80, 999}, + null, + new Object[] { + Arrays.stream(new String[] {"excluded", "d1", "d1", "d2", "d2", "excluded"}) + .map(tag -> new Binary(tag, StandardCharsets.UTF_8)) + .toArray(Binary[]::new), + new long[] {1, 50, 60, 70, 80, 999} + }, + 6, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + tablet.setFlatDevice(true); + processor.insertTablet( + tablet, List.of(new int[] {1, 3}, new int[] {3, 5}), new TSStatus[6], false, new long[5]); + assertEquals(Set.of(first.getDeviceID(), second.getDeviceID()), resource.getDevices()); + assertResourceTimes(resource, first.getDeviceID(), 10, 60); + assertResourceTimes(resource, second.getDeviceID(), 30, 80); + IDeviceID physicalDevice = first.getMemTableDeviceID(first.getDeviceID()); + assertEquals(Set.of(physicalDevice), processor.getWorkMemTable().getMemTableMap().keySet()); + assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70), sequence, false)); + + // Flush and close the file once. An all-null logical ID equals the physical table-name ID, + // but must retain its own range instead of the entire physical group's range. + processor.insert(flatDeviceRow(90, null, schemas), new long[5]); + processor.insert(flatDeviceRow(100, "d2", schemas), new long[5]); + processor.asyncClose().get(30, TimeUnit.SECONDS); + assertTrue(resource.isClosed()); + assertEquals( + Set.of(first.getDeviceID(), second.getDeviceID(), physicalDevice), resource.getDevices()); + assertResourceTimes(resource, first.getDeviceID(), 10, 60); + assertResourceTimes(resource, second.getDeviceID(), 30, 100); + assertResourceTimes(resource, physicalDevice, 90, 90); + assertEquals(100L, flushedTimes.get(physicalDevice).longValue()); + assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70), sequence, false)); + assertFalse(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(200), sequence, false)); + assertEquals(10, resource.getOrderTimeForSeq(physicalDevice, true)); + assertEquals(100, resource.getOrderTimeForUnseq(physicalDevice, false)); + + TsFileResource restored = new TsFileResource(new File(filePath)); + restored.deserialize(); + assertEquals(resource.getDevices(), restored.getDevices()); + assertResourceTimes(restored, first.getDeviceID(), 10, 60); + assertResourceTimes(restored, second.getDeviceID(), 30, 100); + assertResourceTimes(restored, physicalDevice, 90, 90); + + // Read through the query scanner, which uses the physical ID even though the resource has + // logical ranges. The null-TAG device's [90, 90] range must not prune d2's row at time 70. + SeriesScanOptions.Builder scanOptions = new SeriesScanOptions.Builder(); + scanOptions.withAllSensors(Set.of("tag1", "s1")); + scanOptions.withGlobalTimeFilter(TimeFilterApi.eq(70)); + AlignedSeriesScanUtil scan = + new AlignedSeriesScanUtil( + new AlignedFullPath(physicalDevice, List.of("tag1", "s1"), Arrays.asList(schemas)), + Ordering.ASC, + scanOptions.build(), + EnvironmentUtils.TEST_QUERY_FI_CONTEXT, + true, + List.of(TSDataType.STRING, TSDataType.INT64), + List.of(0)); + scan.initQueryDataSource( + new QueryDataSource( + sequence ? List.of(resource) : Collections.emptyList(), + sequence ? Collections.emptyList() : List.of(resource))); + int rows = 0; + Optional<Boolean> hasNextFile; + while ((hasNextFile = scan.hasNextFile()).orElse(true)) { + if (hasNextFile.isEmpty()) { + continue; + } + Optional<Boolean> hasNextChunk; + while ((hasNextChunk = scan.hasNextChunk()).orElse(true)) { + if (hasNextChunk.isEmpty()) { + continue; + } + while (scan.hasNextPage()) { + TsBlock block = scan.nextPage(); + if (block == null) { + continue; + } + for (int i = 0; i < block.getPositionCount(); i++) { + assertEquals(70, block.getTimeByIndex(i)); + assertEquals(new Binary("d2", StandardCharsets.UTF_8), block.getColumn(0).getBinary(i)); + assertEquals(70, block.getColumn(1).getLong(i)); + rows++; + } + } + } + } + assertEquals(1, rows); + } + + private RelationalInsertRowNode flatDeviceRow(long time, String tag, MeasurementSchema[] schemas) + throws IllegalPathException { + RelationalInsertRowNode row = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath("flat_table", false), + true, + new String[] {"tag1", "s1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + schemas, + time, + new Object[] {tag == null ? null : new Binary(tag, StandardCharsets.UTF_8), time}, + false, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + row.setFlatDevice(true); + return row; + } + + private void assertResourceTimes( + TsFileResource resource, IDeviceID device, long start, long end) { + assertEquals(start, resource.getStartTime(device).get().longValue()); + assertEquals(end, resource.getEndTime(device).get().longValue()); + } + private void closeTsFileProcessor(TsFileProcessor unsealedTsFileProcessor) throws TsFileProcessorException { TsFileResource resource = unsealedTsFileProcessor.getTsFileResource(); 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 99a889e2e5f..426fa0b7354 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 @@ -23,10 +23,18 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.De 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; +import org.apache.iotdb.db.utils.ModificationUtils; +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.IDeviceID.Factory; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.file.metadata.statistics.LongStatistics; import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.utils.Binary; import org.junit.Test; import java.io.ByteArrayInputStream; @@ -34,8 +42,10 @@ import java.io.ByteArrayOutputStream; import java.io.EOFException; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.List; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -43,6 +53,115 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class TableDeletionEntryTest { + @Test + public void testChunkDeletionForNormalAndFlatDevices() { + // Normal-table predicates are already matched against the chunk's logical device. Only + // flat-device TAG predicates must defer deletion until individual rows can be inspected. + DeletionPredicate tagPredicate = new DeletionPredicate("t", new SegmentExactMatch("a", 1)); + for (TimeRange range : List.of(new TimeRange(2, 3), new TimeRange(1, 5))) { + List<ModEntry> deletions = + List.of( + new TableDeletionEntry(tagPredicate, range), + new FlatDeviceDeletionEntry(tagPredicate, range), + new FlatDeviceDeletionEntry(new DeletionPredicate("t"), range)); + for (ModEntry deletion : deletions) { + LongStatistics statistics = new LongStatistics(); + statistics.update(1, 10L); + statistics.update(5, 50L); + ChunkMetadata chunk = + new ChunkMetadata( + "s1", + TSDataType.INT64, + TSEncoding.PLAIN, + CompressionType.UNCOMPRESSED, + 0, + statistics); + List<ChunkMetadata> chunks = new ArrayList<>(List.of(chunk)); + ModificationUtils.modifyChunkMetaData(chunks, List.of(deletion)); + if (deletion == deletions.get(1)) { + assertEquals(List.of(chunk), chunks); + assertTrue( + chunk.getDeleteIntervalList() == null || chunk.getDeleteIntervalList().isEmpty()); + } else { + assertEquals(List.of(range), chunk.getDeleteIntervalList()); + if (range.contains(1, 5)) { + assertTrue(chunks.isEmpty()); + } else { + assertEquals(List.of(chunk), chunks); + assertTrue(chunk.isModified()); + } + } + } + } + } + + // All predicate types must agree with device-ID matching for empty, missing and null TAGs, + // including after both serialization paths; warming the Binary caches must not alter the wire + // format. + @Test + public void testBinarySegmentsMatching() throws IOException { + IDeviceID device = Factory.DEFAULT_FACTORY.create(new String[] {"t", "a", ""}); + TagPredicate[] predicates = { + new NOP(), + new FullExactMatch(device), + new DeviceIn(Collections.singleton(device)), + 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)) + }; + String[][] rows = { + {"t", "a", ""}, {"t", "a", "", null}, {"t", "a"}, + {"t", null, ""}, {"other", "a", ""}, {"t", "a", "", "b"} + }; + for (TagPredicate predicate : predicates) { + TableDeletionEntry entry = + new TableDeletionEntry(new DeletionPredicate("t", predicate), new TimeRange(1, 5)); + for (String[] row : rows) { + Binary[] segments = + Arrays.stream(row) + .map(value -> value == null ? null : new Binary(value, TSFileConfig.STRING_CHARSET)) + .toArray(Binary[]::new); + IDeviceID rowDevice = Factory.DEFAULT_FACTORY.create(row); + assertEquals(predicate.matches(rowDevice), predicate.matches(segments)); + assertEquals(entry.affects(rowDevice), entry.affects(segments)); + ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize()); + entry.serialize(buffer); + assertEquals(entry.serializedSize(), buffer.position()); + buffer.flip(); + assertEquals( + entry.affects(rowDevice), + ((TableDeletionEntry) ModEntry.createFrom(buffer)).affects(segments)); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + entry.serialize(output); + assertEquals( + entry.affects(rowDevice), + ((TableDeletionEntry) + ModEntry.createFrom(new ByteArrayInputStream(output.toByteArray()))) + .affects(segments)); + } + } + } + + // Exact-match predicates must not retain a caller's reusable row array in their lookup keys. + @Test + public void testReusableBinarySegments() { + IDeviceID device = Factory.DEFAULT_FACTORY.create(new String[] {"t", "a"}); + Binary[] segments = { + new Binary("t", TSFileConfig.STRING_CHARSET), new Binary("a", TSFileConfig.STRING_CHARSET) + }; + for (TagPredicate predicate : + Arrays.asList(new FullExactMatch(device), new DeviceIn(Collections.singleton(device)))) { + segments[1] = new Binary("a", TSFileConfig.STRING_CHARSET); + assertTrue(predicate.matches(segments)); + segments[1] = null; + assertFalse(predicate.matches(segments)); + segments[1] = new Binary("a", TSFileConfig.STRING_CHARSET); + assertTrue(predicate.matches(segments)); + } + } + @Test public void testSerialization() throws IOException { TableDeletionEntry entry =
