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
The following commit(s) were added to refs/heads/add_flat_device by this push:
new 87eb52e66b8 tmp
87eb52e66b8 is described below
commit 87eb52e66b886275ae760f418efbaba48017ea7e
Author: Tian Jiang <[email protected]>
AuthorDate: Wed Oct 14 17:36:07 2026 +0800
tmp
---
...ileInsertionEventTableParserTabletIterator.java | 27 ++-
.../tsfile/parser/util/ModsOperationUtil.java | 53 ++++-
.../compaction/repair/RepairDataFileScanUtil.java | 5 +-
.../memtable/AlignedWritableMemChunk.java | 24 ++-
.../dataregion/memtable/TsFileProcessor.java | 33 ++-
.../dataregion/modification/TagPredicate.java | 8 +-
.../datastructure/FlatDeviceAlignedTVList.java | 27 ++-
.../db/utils/datastructure/FlatDeviceEncoder.java | 24 ++-
.../table/FlatDeviceTableParserDeletionTest.java | 226 +++++++++++++++++++++
.../repair/RepairDataFileScanUtilTest.java | 51 +++++
.../dataregion/memtable/TsFileProcessorTest.java | 30 +++
.../utils/datastructure/FlatDeviceEncoderTest.java | 42 +++-
.../schema/table/TsFileTableSchemaUtil.java | 15 +-
.../schema/table/TsFileTableSchemaUtilTest.java | 47 +++++
14 files changed, 567 insertions(+), 45 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
index e7aa8387080..d641a9d617b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
@@ -480,7 +480,8 @@ public class TsFileInsertionEventTableParserTabletIterator
implements Iterator<T
this.chunkReader = new TableChunkReader(timeChunk, valueChunkList, null);
this.modsInfoList =
- ModsOperationUtil.initializeMeasurementMods(deviceID, measurementList,
modifications);
+ ModsOperationUtil.initializeMeasurementMods(
+ deviceID, measurementList, modifications, isFlatDeviceTable);
}
private boolean areAllFieldsDeletedByMods(
@@ -522,15 +523,29 @@ public class
TsFileInsertionEventTableParserTabletIterator implements Iterator<T
Objects.nonNull(data.getVector()) ? data.getVector() : new
TsPrimitiveType[0];
boolean needFillTime = false;
final int valueColumnOffset = isFlatDeviceTable ? 0 : deviceIdSize;
- boolean hasNonDeletedField = dataTypeList.size() == valueColumnOffset;
+ boolean hasNonDeletedField =
+ isFlatDeviceTable ? fieldSchemaList.isEmpty() : dataTypeList.size() ==
valueColumnOffset;
+ if (isFlatDeviceTable
+ && fieldSchemaList.isEmpty()
+ && !modsInfoList.isEmpty()
+ && ModsOperationUtil.isDelete(data.currentTime(), modsInfoList.get(0),
primitiveTypes)) {
+ return false;
+ }
for (int i = valueColumnOffset, size = dataTypeList.size(); i < size; i++)
{
final TsPrimitiveType primitiveType =
i - valueColumnOffset < primitiveTypes.length
? primitiveTypes[i - valueColumnOffset]
: null;
- final boolean isDeleted = ModsOperationUtil.isDelete(data.currentTime(),
modsInfoList.get(i));
- if (!isDeleted) {
+ // TAGs define identity and must survive FIELD deletion. Only surviving
FIELDs keep a row.
+ final boolean isTag = isFlatDeviceTable && i < deviceIdSize;
+ final boolean isDeleted =
+ !isTag
+ && (isFlatDeviceTable
+ ? ModsOperationUtil.isDelete(
+ data.currentTime(), modsInfoList.get(i), primitiveTypes)
+ : ModsOperationUtil.isDelete(data.currentTime(),
modsInfoList.get(i)));
+ if (!isTag && !isDeleted) {
hasNonDeletedField = true;
}
if (primitiveType == null || isDeleted) {
@@ -543,7 +558,9 @@ public class TsFileInsertionEventTableParserTabletIterator
implements Iterator<T
PipeTabletUtils.markNullValue(tablet, rowIndex, i);
continue;
}
- needFillTime = true;
+ if (!isTag || fieldSchemaList.isEmpty()) {
+ needFillTime = true;
+ }
switch (dataTypeList.get(i)) {
case BOOLEAN:
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java
index e2b65b5415c..bbcdc25e7a1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/util/ModsOperationUtil.java
@@ -21,18 +21,23 @@ package
org.apache.iotdb.db.pipe.event.common.tsfile.parser.util;
import org.apache.iotdb.commons.path.PatternTreeMap;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+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.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.TagPredicateType;
import org.apache.iotdb.db.utils.ModificationUtils;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.TsPrimitiveType;
import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
@@ -103,6 +108,8 @@ public class ModsOperationUtil {
modification ->
modification.getTimeRange().contains(startTime, endTime)
&& modification.affects(deviceID)
+ && (!(modification instanceof FlatDeviceDeletionEntry
deletion)
+ || deletion.getPredicate().getTagPredicateType() ==
TagPredicateType.NOP)
&& modification.affects(measurementID));
} else {
// For tree model: check if any modification covers the time range
@@ -124,7 +131,18 @@ public class ModsOperationUtil {
List<String> measurements,
PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
modifications) {
+ return initializeMeasurementMods(deviceID, measurements, modifications,
false);
+ }
+
+ public static List<ModsInfo> initializeMeasurementMods(
+ IDeviceID deviceID,
+ List<String> measurements,
+ PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
modifications,
+ boolean isFlatDevice) {
+
List<ModsInfo> modsInfos = new ArrayList<>(measurements.size());
+ Map<FlatDeviceDeletionEntry, int[]> tagColumnMappings =
+ isFlatDevice ? new IdentityHashMap<>() : Collections.emptyMap();
for (final String measurement : measurements) {
final List<ModEntry> mods = modifications.getOverlapped(deviceID,
measurement);
@@ -134,7 +152,6 @@ public class ModsOperationUtil {
continue;
}
- // Sort by time range for efficient lookup
// Different filtering logic for tree model and table model
final List<ModEntry> filteredMods;
if (deviceID.isTableModel()) {
@@ -149,13 +166,42 @@ public class ModsOperationUtil {
// For tree model: no additional filtering needed
filteredMods = mods;
}
- // Store sorted mods and start index
- modsInfos.add(new ModsInfo(ModificationUtils.sortAndMerge(filteredMods),
0));
+ if (isFlatDevice) {
+ // Different TAG predicates can have overlapping ranges. Keep each
predicate and its
+ // column-name mapping rather than treating the ranges as one
monotonic time cursor.
+ ModsInfo info = new ModsInfo(filteredMods, 0);
+ info.tagColumnIndices = tagColumnMappings;
+ for (ModEntry mod : filteredMods) {
+ if (mod instanceof FlatDeviceDeletionEntry deletion) {
+ tagColumnMappings.computeIfAbsent(
+ deletion,
+ entry ->
+
entry.getTagColumnNames().stream().mapToInt(measurements::indexOf).toArray());
+ }
+ }
+ modsInfos.add(info);
+ } else {
+ modsInfos.add(new
ModsInfo(ModificationUtils.sortAndMerge(filteredMods), 0));
+ }
}
return modsInfos;
}
+ /** Flat rows are TAG-first, so timestamps may restart when the logical
device changes. */
+ public static boolean isDelete(long time, ModsInfo modsInfo,
TsPrimitiveType[] values) {
+ for (ModEntry mod : modsInfo.getMods()) {
+ if (mod.getTimeRange().contains(time)
+ && (!(mod instanceof FlatDeviceDeletionEntry deletion)
+ || deletion
+ .getPredicate()
+ .matches(values, modsInfo.tagColumnIndices.get(deletion)))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Check if data at the specified time point is deleted
*
@@ -272,6 +318,7 @@ public class ModsOperationUtil {
public static class ModsInfo {
private List<ModEntry> mods;
private int currentIndex;
+ private Map<FlatDeviceDeletionEntry, int[]> tagColumnIndices =
Collections.emptyMap();
public ModsInfo(List<ModEntry> mods, int currentIndex) {
this.mods = Objects.requireNonNull(mods);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java
index 309aed2ca27..cfc8b9fc03b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java
@@ -70,6 +70,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -305,7 +306,9 @@ public class RepairDataFileScanUtil {
for (int i = 0; i < previousValues.length; i++) {
int compare =
Objects.compare(
- previousValues[i].getBinary(), currentValues[i].getBinary(),
Binary::compareTo);
+ previousValues[i] == null ? null : previousValues[i].getBinary(),
+ currentValues[i] == null ? null : currentValues[i].getBinary(),
+ Comparator.nullsFirst(Binary::compareTo));
if (compare != 0) {
return compare;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index 50c6698973c..ad5957df04b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.AlignedTVListFactory;
import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo;
import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList;
+import
org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList.EncodingStatistics;
import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
import org.apache.iotdb.db.utils.datastructure.MemPointIteratorFactory;
import org.apache.iotdb.db.utils.datastructure.TVList;
@@ -75,6 +76,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted;
public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
+ private EncodingStatistics flatDeviceEncodingStatistics;
private Map<String, Integer> measurementIndexMap;
private final Map<String, TsTableColumnCategory> measurementCategoryMap;
@@ -766,6 +768,10 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
return useColumnSortedList;
}
+ public EncodingStatistics getFlatDeviceEncodingStatistics() {
+ return flatDeviceEncodingStatistics;
+ }
+
public int deleteFlatDevice(FlatDeviceDeletionEntry deletion) {
List<String> tagColumnNames = deletion.getTagColumnNames();
// Logical TAG ordinals do not necessarily match physical positions:
failed/omitted TAGs
@@ -1345,14 +1351,15 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
}
lists.add((FlatDeviceAlignedTVList) workingListForFlush);
List<IMeasurementSchema> activeSchemas = getActiveSchemaList();
- FlatDeviceAlignedTVList.encode(
- lists,
- activeSchemas,
- buildColumnIndexList(activeSchemas),
- encryptParameter,
- ioTaskQueue,
- encodeInfo.maxNumberOfPointsInChunk,
- encodeInfo.maxNumberOfPointsInPage);
+ flatDeviceEncodingStatistics =
+ FlatDeviceAlignedTVList.encode(
+ lists,
+ activeSchemas,
+ buildColumnIndexList(activeSchemas),
+ encryptParameter,
+ ioTaskQueue,
+ encodeInfo.maxNumberOfPointsInChunk,
+ encodeInfo.maxNumberOfPointsInPage);
encodeInfo.reset();
return;
}
@@ -1421,6 +1428,7 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
@Override
public void release() {
+ flatDeviceEncodingStatistics = null;
maybeReleaseTvList(list);
for (AlignedTVList alignedTvList : sortedList) {
maybeReleaseTvList(alignedTvList);
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 808e4a87131..8c01f490df1 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
@@ -419,16 +419,7 @@ public class TsFileProcessor {
private void updateResourceTime(
IDeviceID deviceID, long startTime, long endTime, boolean flatDevice) {
- updateResourceTime(deviceID, startTime, endTime, flatDevice, 1);
- }
-
- private void updateResourceTime(
- IDeviceID deviceID, long startTime, long endTime, boolean flatDevice,
long rowCount) {
if (flatDevice) {
- flatDeviceLogicalDevices
- .computeIfAbsent(deviceID.getTableName(), key -> new HashSet<>())
- .add(deviceID.toString());
- flatDeviceRowCounts.merge(deviceID.getTableName(), rowCount, Long::sum);
tsFileResource.updateTableStartTime(deviceID.getTableName(), startTime);
tsFileResource.updateTableEndTime(deviceID.getTableName(), endTime);
} else {
@@ -789,8 +780,7 @@ public class TsFileProcessor {
deviceEndOffset.left,
insertTabletNode.getTimes()[deviceStartOffset],
insertTabletNode.getTimes()[deviceEndOffset.right - 1],
- insertTabletNode.isFlatDevice(),
- deviceEndOffset.right - deviceStartOffset);
+ insertTabletNode.isFlatDevice());
deviceStartOffset = deviceEndOffset.right;
}
recordFlatDeviceWriteCost(
@@ -2010,6 +2000,7 @@ public class TsFileProcessor {
dataRegionName,
dataRegionInfo.getDataRegion().getDataRegionIdString());
flushTask.syncFlushMemTable();
+ recordFlatDeviceEncodingStatistics(memTableToFlush);
memTableFlushPointCount = memTableToFlush.getTotalPointsNum();
} catch (Throwable e) {
if (writer == null) {
@@ -2215,6 +2206,26 @@ public class TsFileProcessor {
}
}
+ private void recordFlatDeviceEncodingStatistics(IMemTable memTable) {
+ // Publish only after encoding and IO both succeed, so a failed flush is
never counted.
+ memTable
+ .getMemTableMap()
+ .forEach(
+ (device, group) -> {
+ if (group instanceof AlignedWritableMemChunkGroup alignedGroup) {
+ var statistics =
+
alignedGroup.getAlignedMemChunk().getFlatDeviceEncodingStatistics();
+ if (statistics != null) {
+ String table = device.getTableName();
+ Set<String> devices =
+ flatDeviceLogicalDevices.computeIfAbsent(table, key ->
new HashSet<>());
+ statistics.getDevices().forEach(id ->
devices.add(id.toString()));
+ flatDeviceRowCounts.merge(table, statistics.getRowCount(),
Long::sum);
+ }
+ }
+ });
+ }
+
/** end file and write some meta */
private void endFile() throws IOException, TsFileProcessorException {
if (logger.isDebugEnabled()) {
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 6c2331f6290..49667a19fbb 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
@@ -30,6 +30,7 @@ import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.apache.tsfile.utils.TsPrimitiveType;
import java.io.EOFException;
import java.io.IOException;
@@ -108,7 +109,12 @@ public abstract class TagPredicate implements
StreamSerializable, BufferSerializ
}
int columnIndex = tagColumnIndices[tagIndex];
Object value = columnIndex >= 0 && columnIndex < values.length ?
values[columnIndex] : null;
- return value == null ? null : value.toString();
+ if (value instanceof TsPrimitiveType primitive) {
+ value = primitive.getBinary();
+ }
+ return value instanceof Binary binary
+ ? binary.getStringValue(TSFileConfig.STRING_CHARSET)
+ : value == null ? null : value.toString();
}
private static List<String> tagValuesKey(Object[] values, int[]
tagColumnIndices) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java
index 186c42fc56e..21cbc50ba8f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java
@@ -1026,7 +1026,7 @@ public class FlatDeviceAlignedTVList extends
AlignedTVList {
}
/** Encodes lists in device order, merging only row IDs within the current
device. */
- public static void encode(
+ public static EncodingStatistics encode(
List<FlatDeviceAlignedTVList> lists,
List<IMeasurementSchema> schemas,
List<Integer> columns,
@@ -1034,7 +1034,30 @@ public class FlatDeviceAlignedTVList extends
AlignedTVList {
BlockingQueue<Object> output,
long chunkLimit,
int pageLimit) {
- FlatDeviceEncoder.encode(lists, schemas, columns, encryption, output,
chunkLimit, pageLimit);
+ return FlatDeviceEncoder.encode(
+ lists, schemas, columns, encryption, output, chunkLimit, pageLimit);
+ }
+
+ /** Counts physical output rows after duplicate resolution and time
deletion. */
+ public static final class EncodingStatistics {
+ private long rowCount;
+ private final Set<IDeviceID> devices = new HashSet<>();
+
+ void recordRow() {
+ rowCount++;
+ }
+
+ void recordDevice(IDeviceID device) {
+ devices.add(device);
+ }
+
+ public long getRowCount() {
+ return rowCount;
+ }
+
+ public Set<IDeviceID> getDevices() {
+ return Collections.unmodifiableSet(devices);
+ }
}
AlignedTVList fieldStorage() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java
index b82c9bee802..52ac16a3223 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoder.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.utils.datastructure;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter;
import
org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList.DeviceInfo;
+import
org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList.EncodingStatistics;
import org.apache.tsfile.encrypt.EncryptParameter;
import org.apache.tsfile.utils.Binary;
@@ -36,7 +37,7 @@ import java.util.concurrent.BlockingQueue;
final class FlatDeviceEncoder {
private FlatDeviceEncoder() {}
- static void encode(
+ static EncodingStatistics encode(
List<FlatDeviceAlignedTVList> lists,
List<IMeasurementSchema> schemas,
List<Integer> columnIndices,
@@ -45,10 +46,10 @@ final class FlatDeviceEncoder {
long chunkLimit,
int pageLimit) {
if (lists.size() == 1) {
- encodeSingleList(
+ return encodeSingleList(
lists.get(0), schemas, columnIndices, encryption, output,
chunkLimit, pageLimit);
- return;
}
+ EncodingStatistics statistics = new EncodingStatistics();
PriorityQueue<Cursor> devices =
new PriorityQueue<>(
(a, b) -> FlatDeviceAlignedTVList.compareDevices(a.device.id,
b.device.id));
@@ -74,6 +75,7 @@ final class FlatDeviceEncoder {
while (!devices.isEmpty()) {
// All rows in this group share TAGs; retain them even when a cursor
advances devices.
DeviceInfo currentDevice = devices.peek().device;
+ statistics.recordDevice(currentDevice.id);
do {
Cursor cursor = devices.remove();
rows.add(cursor);
@@ -131,6 +133,7 @@ final class FlatDeviceEncoder {
nil);
}
writer.write(time);
+ statistics.recordRow();
chunkRows++;
// The chunk writer may already have sealed on its byte-size threshold.
if (writer.getTimeChunkWriter().getPageWriter().getPointNumber() >=
pageLimit) {
@@ -138,7 +141,7 @@ final class FlatDeviceEncoder {
}
if (chunkRows >= chunkLimit) {
if (!enqueue(writer, output)) {
- return;
+ return statistics;
}
writer = new FlatDeviceAlignedChunkWriter(schemas, encryption);
chunkRows = 0;
@@ -148,9 +151,10 @@ final class FlatDeviceEncoder {
if (chunkRows > 0) {
enqueue(writer, output);
}
+ return statistics;
}
- private static void encodeSingleList(
+ private static EncodingStatistics encodeSingleList(
FlatDeviceAlignedTVList list,
List<IMeasurementSchema> schemas,
List<Integer> columnIndices,
@@ -159,9 +163,11 @@ final class FlatDeviceEncoder {
long chunkLimit,
int pageLimit) {
list.sort();
+ EncodingStatistics statistics = new EncodingStatistics();
FlatDeviceAlignedChunkWriter writer = new
FlatDeviceAlignedChunkWriter(schemas, encryption);
long chunkRows = 0;
for (DeviceInfo device : list.getDeviceInfoMap().values()) {
+ boolean recordedDevice = false;
int position = 0;
while (position < device.rowCount) {
int start = position++;
@@ -206,13 +212,18 @@ final class FlatDeviceEncoder {
nil);
}
writer.write(time);
+ statistics.recordRow();
+ if (!recordedDevice) {
+ statistics.recordDevice(device.id);
+ recordedDevice = true;
+ }
chunkRows++;
if (writer.getTimeChunkWriter().getPageWriter().getPointNumber() >=
pageLimit) {
writer.sealCurrentPage();
}
if (chunkRows >= chunkLimit) {
if (!enqueue(writer, output)) {
- return;
+ return statistics;
}
writer = new FlatDeviceAlignedChunkWriter(schemas, encryption);
chunkRows = 0;
@@ -222,6 +233,7 @@ final class FlatDeviceEncoder {
if (chunkRows > 0) {
enqueue(writer, output);
}
+ return statistics;
}
private static void writeValue(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/FlatDeviceTableParserDeletionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/FlatDeviceTableParserDeletionTest.java
new file mode 100644
index 00000000000..fd7a3881242
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/FlatDeviceTableParserDeletionTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.pipe.event.common.tsfile.parser.table;
+
+import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter;
+import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.encrypt.EncryptUtils;
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.writer.TsFileIOWriter;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+public class FlatDeviceTableParserDeletionTest {
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ private static final TsFileInsertionEventParserMemoryBlock MEMORY =
+ new TsFileInsertionEventParserMemoryBlock() {
+ @Override
+ public long getMemoryUsageInBytes() {
+ return 1L << 30;
+ }
+
+ @Override
+ public void forceResize(long newSizeInBytes) {}
+
+ @Override
+ public void close() {}
+ };
+
+ @Test
+ public void testTagScopedDeletionCannotSkipWholeChunk() throws Exception {
+ FlatDeviceDeletionEntry deletion =
+ deletion(new TagPredicate.SegmentExactMatch("a", 1), List.of());
+ List<Tablet> tablets = parse(new String[] {"a", "b"}, new long[] {1, 2},
deletion);
+ Assert.assertEquals(List.of(2L), times(tablets));
+ Assert.assertEquals(binary("b"), ((Binary[])
tablets.get(0).getValues()[0])[0]);
+ }
+
+ @Test
+ public void testDeletionSurvivesTimeResetAcrossLogicalDevices() throws
Exception {
+ List<Tablet> tablets =
+ parse(
+ new String[] {"a", "b", "c"},
+ new long[] {3, 1, 2},
+ deletion(new TagPredicate.NOP(), List.of()));
+ Assert.assertEquals(List.of(3L), times(tablets));
+ }
+
+ @Test
+ public void testScopedFieldDeletionKeepsOtherFieldsAndTags() throws
Exception {
+ List<Tablet> tablets =
+ parse(
+ new String[] {"a", "b"},
+ new long[] {1, 2},
+ deletion(new TagPredicate.SegmentExactMatch("a", 1),
List.of("s0")));
+ Assert.assertEquals(List.of(1L, 2L), times(tablets));
+ Tablet tablet = tablets.get(0);
+ Assert.assertEquals(binary("a"), ((Binary[]) tablet.getValues()[0])[0]);
+ Assert.assertTrue(tablet.getBitMaps()[2].isMarked(0));
+ Assert.assertEquals(101L, ((long[]) tablet.getValues()[3])[0]);
+ Assert.assertFalse(tablet.getBitMaps()[2].isMarked(1));
+ }
+
+ @Test
+ public void testTagMappingUsesDeletionColumnNames() throws Exception {
+ FlatDeviceDeletionEntry deletion =
+ deletion(
+ new TagPredicate.FullExactMatch(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"t",
"x", "a"})),
+ List.of());
+ deletion.setTagColumnNames(List.of("tag1", "tag0"));
+ Assert.assertEquals(
+ List.of(2L), times(parse(new String[] {"a", "b"}, new long[] {1, 2},
deletion)));
+ }
+
+ @Test
+ public void testNullAndMissingTagsMatchNullPredicate() throws Exception {
+ FlatDeviceDeletionEntry deletion =
+ deletion(new TagPredicate.SegmentExactMatch(null, 1), List.of());
+ Assert.assertEquals(
+ List.of(2L), times(parse(new String[] {null, "b"}, new long[] {1, 2},
deletion)));
+ deletion.setTagColumnNames(List.of("new_tag", "tag0"));
+ Assert.assertTrue(times(parse(new String[] {"a", "b"}, new long[] {1, 2},
deletion)).isEmpty());
+ }
+
+ @Test
+ public void testOverlappingRangesRetainIndependentTagPredicates() throws
Exception {
+ FlatDeviceDeletionEntry first = deletion(new
TagPredicate.SegmentExactMatch("a", 1), List.of());
+ FlatDeviceDeletionEntry second =
+ new FlatDeviceDeletionEntry(
+ new DeletionPredicate("t", new TagPredicate.SegmentExactMatch("b",
1)),
+ new TimeRange(1, 1));
+ second.setTagColumnNames(List.of("tag0", "tag1"));
+ Assert.assertEquals(
+ List.of(2L),
+ times(parse(new String[] {"a", "b", "b"}, new long[] {1, 1, 2}, first,
second)));
+ }
+
+ private FlatDeviceDeletionEntry deletion(TagPredicate predicate,
List<String> fields) {
+ FlatDeviceDeletionEntry deletion =
+ new FlatDeviceDeletionEntry(
+ new DeletionPredicate("t", predicate, fields), new TimeRange(1,
2));
+ deletion.setTagColumnNames(List.of("tag0", "tag1"));
+ return deletion;
+ }
+
+ private List<Tablet> parse(String[] tags, long[] times,
FlatDeviceDeletionEntry... deletions)
+ throws Exception {
+ File file = folder.newFile();
+ List<IMeasurementSchema> schemas =
+ List.of(
+ new MeasurementSchema("tag0", TSDataType.STRING),
+ new MeasurementSchema("tag1", TSDataType.STRING),
+ new MeasurementSchema("s0", TSDataType.INT64),
+ new MeasurementSchema("s1", TSDataType.INT64));
+ try (TsFileIOWriter writer = new TsFileIOWriter(file)) {
+ writer
+ .getSchema()
+ .registerTableSchema(
+ new TableSchema(
+ "t",
+ schemas,
+ List.of(
+ ColumnCategory.TAG,
+ ColumnCategory.TAG,
+ ColumnCategory.FIELD,
+ ColumnCategory.FIELD)));
+ TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "t", true);
+ writer.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create(new
String[] {"t"}));
+ FlatDeviceAlignedChunkWriter chunk =
+ new FlatDeviceAlignedChunkWriter(schemas,
EncryptUtils.getEncryptParameter());
+ for (int rowIndex = 0; rowIndex < times.length; rowIndex++) {
+ long time = times[rowIndex];
+ chunk.write(
+ time,
+ tags[rowIndex] == null ? Binary.EMPTY_VALUE :
binary(tags[rowIndex]),
+ tags[rowIndex] == null);
+ chunk.write(time, binary("x"), false);
+ chunk.write(time, time, false);
+ chunk.write(time, 100L + time, false);
+ chunk.write(time);
+ }
+ chunk.writeToFileWriter(writer);
+ writer.endChunkGroup();
+ writer.endFile();
+ }
+ var mods = PatternTreeMapFactory.getModsPatternTreeMap();
+ for (FlatDeviceDeletionEntry deletion : deletions) {
+ mods.append(deletion.keyOfPatternTree(), deletion);
+ }
+ List<Tablet> tablets = new ArrayList<>();
+ try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.getPath())) {
+ TsFileInsertionEventTableParserTabletIterator parser =
+ new TsFileInsertionEventTableParserTabletIterator(
+ reader,
+ entry -> true,
+ MEMORY,
+ MEMORY,
+ MEMORY,
+ MEMORY,
+ MEMORY,
+ mods,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ while (parser.hasNext()) {
+ Tablet tablet = parser.next();
+ if (tablet.getRowSize() > 0) {
+ tablets.add(tablet);
+ }
+ }
+ }
+ return tablets;
+ }
+
+ private List<Long> times(List<Tablet> tablets) {
+ List<Long> result = new ArrayList<>();
+ for (Tablet tablet : tablets) {
+ for (int rowIndex = 0; rowIndex < tablet.getRowSize(); rowIndex++) {
+ result.add(tablet.getTimestamp(rowIndex));
+ }
+ }
+ return result;
+ }
+
+ private static Binary binary(String value) {
+ return new Binary(value, TSFileConfig.STRING_CHARSET);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtilTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtilTest.java
index 90ae850afa1..433b99dd087 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtilTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtilTest.java
@@ -20,24 +20,35 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction.repair;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil;
import org.apache.iotdb.db.exception.StorageEngineException;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex;
+import org.apache.tsfile.encrypt.EncryptUtils;
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.file.metadata.ChunkMetadata;
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.common.TimeRange;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.writer.TsFileIOWriter;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
@@ -83,6 +94,46 @@ public class RepairDataFileScanUtilTest extends
AbstractCompactionTest {
Assert.assertFalse(scanUtil.hasUnsortedDataOrWrongStatistics());
}
+ @Test
+ public void testFlatDeviceNullTagOrder() throws IOException {
+ for (boolean unsorted : List.of(false, true)) {
+ TsFileResource resource = createEmptyFileAndResource(true);
+ List<IMeasurementSchema> schemas =
+ List.of(
+ new MeasurementSchema("tag", TSDataType.STRING),
+ new MeasurementSchema("s", TSDataType.INT64));
+ try (TsFileIOWriter writer = new TsFileIOWriter(resource.getTsFile())) {
+ writer
+ .getSchema()
+ .registerTableSchema(
+ new TableSchema("t", schemas, List.of(ColumnCategory.TAG,
ColumnCategory.FIELD)));
+ TsFileTableSchemaUtil.addFlatDeviceProperty(writer, "t", true);
+ writer.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create(new
String[] {"t"}));
+ FlatDeviceAlignedChunkWriter chunk =
+ new FlatDeviceAlignedChunkWriter(schemas,
EncryptUtils.getEncryptParameter());
+ // Null TAGs come first and may repeat at increasing times. A later
non-null TAG may
+ // restart time; placing a null TAG after it must be unsorted, never a
broken file.
+ String[] tags = unsorted ? new String[] {"a", null} : new String[]
{null, null, "a"};
+ for (int i = 0; i < tags.length; i++) {
+ long time = tags[i] == null ? i + 1 : 1;
+ chunk.write(
+ time,
+ tags[i] == null ? Binary.EMPTY_VALUE : new Binary(tags[i],
StandardCharsets.UTF_8),
+ tags[i] == null);
+ chunk.write(time, time, false);
+ chunk.write(time);
+ }
+ chunk.writeToFileWriter(writer);
+ writer.endChunkGroup();
+ writer.endFile();
+ }
+ RepairDataFileScanUtil scan = new RepairDataFileScanUtil(resource);
+ scan.scanTsFile(false);
+ Assert.assertFalse(scan.isBrokenFile());
+ Assert.assertEquals(unsorted, scan.hasUnsortedDataOrWrongStatistics());
+ }
+ }
+
@Test
public void testWrongChunkStatisticsWithNonAlignedSeries() throws
IOException {
TsFileResource resource = createEmptyFileAndResource(true);
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 598a67d83da..16ffa8761ad 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
@@ -27,6 +27,7 @@ import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.NonAlignedFullPath;
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.TsFileTableSchemaUtil;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -88,6 +89,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
@@ -1571,11 +1573,39 @@ public class TsFileProcessorTest {
assertEquals(Set.of(physicalDevice),
processor.getWorkMemTable().getMemTableMap().keySet());
assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70),
sequence, false));
+ processor.asyncFlush();
+ processor.getCloseFuture().get(30, TimeUnit.SECONDS);
+ assertTrue(processor.getFlushingMemTable().isEmpty());
+
// An all-null logical ID equals the table-name ID. It must extend the
table range,
// never narrow it to the null-TAG device alone.
processor.insert(flatDeviceRow(90, null, schemas), new long[5]);
processor.insert(flatDeviceRow(100, "d2", schemas), new long[5]);
+ // Repeated keys must increase write metrics but not the final TsFile row
count.
+ for (int i = 0; i < 5; i++) {
+ processor.insert(flatDeviceRow(10, "d1", schemas), new long[5]);
+ }
+ // This standalone processor reuses the future slot for flush and close
requests.
+ processor.setCloseFuture(null);
processor.asyncClose().get(30, TimeUnit.SECONDS);
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(filePath)) {
+ Map<String, byte[]> properties = reader.getTsFileProperties();
+ assertEquals(
+ // Eight rows in the first flush, then three physical rows in the
second. The five
+ // repeated writes at (d1,10) resolve to one row in that second
flush.
+ 11,
+ ByteBuffer.wrap(
+ properties.get(
+
TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey("flat_table")))
+ .getLong());
+ assertEquals(
+ 3,
+ ByteBuffer.wrap(
+ properties.get(
+
TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey(
+ "flat_table")))
+ .getLong());
+ }
assertTrue(resource.isClosed());
assertEquals(Set.of(physicalDevice), resource.getDevices());
assertResourceTimes(resource, physicalDevice, 10, 100);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java
index c4bdc712eb0..7a6f3d4d4f2 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceEncoderTest.java
@@ -98,6 +98,46 @@ public class FlatDeviceEncoderTest {
assertRow(rows.get(4), 3, "c", 30L, 300L);
}
+ @Test
+ public void testStatisticsCountEncodedRowsAfterDuplicateMergeAndDeletion()
throws Exception {
+ List<TSDataType> types = List.of(TSDataType.STRING, TSDataType.INT64,
TSDataType.INT64);
+ // Exercise both the single-list fast path and the merge across historical
lists.
+ for (boolean multipleLists : List.of(false, true)) {
+ FlatDeviceAlignedTVList old = list(types, List.of(0));
+ FlatDeviceAlignedTVList current = multipleLists ? list(types,
List.of(0)) : old;
+ old.putAlignedValue(1, new Object[] {binary("a"), 10L, null});
+ current.putAlignedValue(1, new Object[] {binary("a"), null, 20L});
+ current.putAlignedValue(1, new Object[] {binary("a"), 11L, null});
+ current.putAlignedValue(2, new Object[] {null, null, null});
+ current.putAlignedValue(3, new Object[] {binary("deleted"), 30L, 30L});
+ current.deleteMatchingRows(3, 3, row -> true, null);
+ BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
+ List<IMeasurementSchema> schemas =
+ IntStream.range(0, types.size())
+ .mapToObj(i -> (IMeasurementSchema) new MeasurementSchema("s" +
i, types.get(i)))
+ .collect(Collectors.toList());
+ FlatDeviceAlignedTVList.EncodingStatistics statistics =
+ FlatDeviceAlignedTVList.encode(
+ multipleLists ? List.of(old, current) : List.of(old),
+ schemas,
+ List.of(0, 1, 2),
+ EncryptDBUtils.getDefaultFirstEncryptParam(),
+ queue,
+ 1,
+ 1);
+ List<Row> rows = read(queue);
+ Assert.assertEquals(2, rows.size());
+ Assert.assertEquals(rows.size(), statistics.getRowCount());
+ Assert.assertEquals(2, statistics.getDevices().size());
+ Assert.assertFalse(
+ statistics
+ .getDevices()
+ .contains(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"table", "deleted"})));
+ assertRow(rows.get(1), 1, "a", 11L, 20L);
+ }
+ }
+
@Test
public void testSingleListLatestNonNullDeletionAndPageChunkBoundaries()
throws Exception {
List<TSDataType> types = List.of(TSDataType.STRING, TSDataType.INT64,
TSDataType.INT64);
@@ -292,7 +332,7 @@ public class FlatDeviceEncoderTest {
}
List<Row> rows = new ArrayList<>();
try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.getPath())) {
- for (AbstractAlignedChunkMetadata metadata :
reader.getAlignedChunkMetadata(device, true)) {
+ for (AbstractAlignedChunkMetadata metadata :
reader.getAlignedChunkMetadata(device, false)) {
Chunk time = reader.readMemChunk((ChunkMetadata)
metadata.getTimeChunkMetadata());
List<Chunk> values = new ArrayList<>();
for (IChunkMetadata value : metadata.getValueChunkMetadataList()) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java
index 17c2acb0168..2c3a250ab71 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtil.java
@@ -46,7 +46,7 @@ import static
org.apache.iotdb.commons.schema.table.TsTable.TIME_COLUMN_NAME;
public class TsFileTableSchemaUtil {
private static final String FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX =
- TsTable.FLAT_DEVICE_PROPERTY + ".";
+ "iotdb." + TsTable.FLAT_DEVICE_PROPERTY + ".";
private static final byte[] FLAT_DEVICE_PROPERTY_BYTES =
TsTable.FLAT_DEVICE_PROPERTY.getBytes(StandardCharsets.UTF_8);
private static final byte[] TRUE_BYTES = {1};
@@ -267,25 +267,26 @@ public class TsFileTableSchemaUtil {
}
public static String getFlatDevicePropertyKey(final String tableName) {
- return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + tableName;
+ // Disjoint prefixes prevent a table named point_count_t from shadowing
t's statistics.
+ return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "layout." + tableName;
}
public static String getFlatDeviceLogicalDeviceCountPropertyKey(final String
tableName) {
- return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "logical_device_count_" +
tableName;
+ return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "logical_device_count." +
tableName;
}
public static String getFlatDevicePointCountPropertyKey(final String
tableName) {
- return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "point_count_" + tableName;
+ return FLAT_DEVICE_TS_FILE_PROPERTY_PREFIX + "row_count." + tableName;
}
public static void addFlatDeviceStatistics(
final TsFileIOWriter writer,
final String tableName,
final long logicalDeviceCount,
- final long pointCount) {
+ final long rowCount) {
writer.addTsFileProperty(
getFlatDeviceLogicalDeviceCountPropertyKey(tableName),
longBytes(logicalDeviceCount));
- writer.addTsFileProperty(getFlatDevicePointCountPropertyKey(tableName),
longBytes(pointCount));
+ writer.addTsFileProperty(getFlatDevicePointCountPropertyKey(tableName),
longBytes(rowCount));
}
private static byte[] longBytes(long value) {
@@ -298,7 +299,7 @@ public class TsFileTableSchemaUtil {
if (Objects.isNull(tsFileProperties)) {
return false;
}
- final byte[] value =
tsFileProperties.get(getFlatDevicePropertyKey(tableName));
+ byte[] value = tsFileProperties.get(getFlatDevicePropertyKey(tableName));
return value != null && value.length == 1 && value[0] == 1;
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java
index 07a83a4838d..237bae5bd76 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsFileTableSchemaUtilTest.java
@@ -82,6 +82,53 @@ public class TsFileTableSchemaUtilTest {
Assert.assertFalse(TsFileTableSchemaUtil.isFlatDevice(new HashMap<>(),
"table"));
}
+ @Test
+ public void testFlatDeviceStatisticsDoNotShadowTableMarkers() throws
IOException {
+ File file = temporaryFolder.newFile("collision-properties.tsfile");
+ List<String> tables =
+ List.of("t", "point_count_t", "logical_device_count_t", "layout.t",
"row_count.t");
+ try (TsFileIOWriter writer = new TsFileIOWriter(file)) {
+ for (String table : tables) {
+ TsFileTableSchemaUtil.addFlatDeviceProperty(writer, table, true);
+ TsFileTableSchemaUtil.addFlatDeviceStatistics(writer, table, 2, 3);
+ }
+ writer.endFile();
+ }
+ try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.getAbsolutePath())) {
+ Map<String, byte[]> properties = reader.getTsFileProperties();
+ Assert.assertEquals(
+ tables.size() * 3,
+ properties.keySet().stream().filter(key ->
key.startsWith("iotdb.flat_device.")).count());
+ for (String table : tables) {
+ Assert.assertTrue(TsFileTableSchemaUtil.isFlatDevice(properties,
table));
+ Assert.assertEquals(
+ 2,
+ ByteBuffer.wrap(
+ properties.get(
+
TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey(table)))
+ .getLong());
+ Assert.assertEquals(
+ 3,
+ ByteBuffer.wrap(
+
properties.get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey(table)))
+ .getLong());
+ }
+ }
+ }
+
+ @Test
+ public void testLegacyFlatDeviceMarker() {
+ Assert.assertTrue(
+ TsFileTableSchemaUtil.isFlatDevice(Map.of("flat_device.t", new byte[]
{1}), "t"));
+ Assert.assertFalse(
+ TsFileTableSchemaUtil.isFlatDevice(
+ Map.of("flat_device.t",
ByteBuffer.allocate(Long.BYTES).putLong(1).array()), "t"));
+ // An old table called layout.t must not be mistaken for t's new layout
property.
+ Assert.assertFalse(
+ TsFileTableSchemaUtil.isFlatDevice(
+ Map.of("flat_device.t", new byte[] {0}, "flat_device.layout.t",
new byte[] {1}), "t"));
+ }
+
@Test
public void testInvalidFlatDevicePropertyDefaultsToFalse() {
final Map<String, byte[]> properties = new HashMap<>();
