This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch removeIsTableModelMethod in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e1a540722a65ba7ee973a5f40f17c8f12fc3ba1d Author: shuwenwei <[email protected]> AuthorDate: Fri Oct 9 18:24:50 2026 +0800 [Storage] Pass table model explicitly in memtable, flush and WAL replay --- .../apache/iotdb/db/i18n/ImportWALMessages.java | 6 ++- .../apache/iotdb/db/i18n/ImportWALMessages.java | 6 ++- .../builder/PipeTableModelTsFileBuilderV2.java | 3 +- .../util/builder/PipeTreeModelTsFileBuilderV2.java | 3 +- .../schemaregion/utils/ResourceByPathUtils.java | 5 +- .../dataregion/flush/MemTableFlushTask.java | 22 ++++++-- .../dataregion/memtable/AbstractMemTable.java | 54 +++++++++++-------- .../memtable/AbstractWritableMemChunk.java | 9 ++-- .../memtable/AlignedWritableMemChunk.java | 52 +++++++++--------- .../memtable/AlignedWritableMemChunkGroup.java | 37 +++++++------ .../dataregion/memtable/IMemTable.java | 8 +-- .../dataregion/memtable/IWritableMemChunk.java | 14 +++-- .../memtable/IWritableMemChunkGroup.java | 12 ++--- .../dataregion/memtable/TsFileProcessor.java | 12 +++-- .../dataregion/memtable/WritableMemChunk.java | 27 ++++++---- .../dataregion/memtable/WritableMemChunkGroup.java | 26 ++++----- .../wal/recover/file/TsFilePlanRedoer.java | 9 +++- .../file/UnsealedTsFileRecoverPerformer.java | 13 +++-- .../java/org/apache/iotdb/db/tools/ImportWAL.java | 61 ++++++++++++++++------ .../fragment/FragmentInstanceExecutionTest.java | 10 ++-- ...ignedBitmapMemoryAccountingPerformanceTest.java | 6 +-- .../memtable/MemChunkDeserializeTest.java | 12 ++--- .../dataregion/memtable/MemTableFlushTaskTest.java | 28 +++++----- .../dataregion/memtable/PrimitiveMemTableTest.java | 27 ++++++---- .../memtable/WritableMemChunkRegionScanTest.java | 8 +-- .../org/apache/iotdb/db/tools/ImportWALTest.java | 54 +++++++++++++++++-- 26 files changed, 338 insertions(+), 186 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java index d28ca7dbcfa..fefbfffe869 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -136,14 +136,16 @@ public final class ImportWALMessages { public static final String MESSAGE_SKIPPED_ARG_CORRUPTED_WAL_FILES_SOURCE_FILES_RETAINED_A889CCE2 = "Skipped %d corrupted WAL files; source files retained."; - public static final String MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_AND_ASK_FOR_CONFIRMATION_4B1E409D = - "Target database for table-model WAL entries. If omitted, infer from the WAL parent directory and ask for confirmation."; + public static final String MESSAGE_TARGET_DATABASE_FOR_WAL_REPLAY_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_IF_INFERENCE_FAILS_DB_DATABASE_IS_REQUIRED_6EEBC019 = + "Target database for WAL replay. If omitted, infer from the WAL parent directory; if inference fails, -db/--database is required."; public static final String MESSAGE_INFERRED_TABLE_DATABASE_ARG_FROM_WAL_DIRECTORY_ARG_REPLAY_INTO_THIS_DATABASE_Y_YES_A_ACCEPT_ALL_INFERRED_DATABASES_N_QUIT_5B59D833 = "Inferred table database %s from WAL directory %s. Replay into this database? [y] yes, [a] accept all inferred databases, [N] quit: "; public static final String EXCEPTION_DATABASE_CONFIRMATION_REQUIRED_FOR_WAL_DIRECTORY_ARG_INFERRED_DATABASE_ARG_SPECIFY_DB_DATABASE_OR_SKIP_DB_CONFIRMATION_WHEN_INTERACTIVE_INPUT_IS_UNAVAILABLE_14DF6D36 = "Database confirmation required for WAL directory %s (inferred database: %s). Specify -db/--database or --skip_db_confirmation when interactive input is unavailable."; public static final String EXCEPTION_REPLAY_INTO_INFERRED_DATABASE_ARG_WAS_NOT_CONFIRMED_SPECIFY_DB_DATABASE_TO_SELECT_THE_TARGET_EXPLICITLY_86F81190 = "Replay into inferred database %s was not confirmed. Specify -db/--database to select the target explicitly."; + public static final String EXCEPTION_CANNOT_DETERMINE_THE_TARGET_DATABASE_OF_WAL_DIRECTORIES_ARG_SPECIFY_DB_DATABASE_WHICH_APPLIES_TO_ALL_IMPORTED_DIRECTORIES_55B174E4 = + "Cannot determine the target database of WAL directories %s. Specify -db/--database, which applies to all imported directories."; public static final String MESSAGE_ACCEPT_ALL_INFERRED_DATABASE_NAMES_WITHOUT_CONFIRMATION_DB_DATABASE_STILL_TAKES_PRECEDENCE_FA49A73C = "Accept all inferred database names without confirmation; -db/--database still takes precedence."; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java index 5f9b6e18b02..fc65b1bec5b 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -135,14 +135,16 @@ public final class ImportWALMessages { public static final String MESSAGE_SKIPPED_ARG_CORRUPTED_WAL_FILES_SOURCE_FILES_RETAINED_A889CCE2 = "已跳过 %d 个损坏的 WAL 文件；源文件已保留。"; - public static final String MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_AND_ASK_FOR_CONFIRMATION_4B1E409D = - "表模型 WAL 条目的目标数据库。省略时尝试从 WAL 文件的父目录名推断，并请求确认。"; + public static final String MESSAGE_TARGET_DATABASE_FOR_WAL_REPLAY_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_IF_INFERENCE_FAILS_DB_DATABASE_IS_REQUIRED_6EEBC019 = + "WAL 重放的目标数据库。省略时尝试从 WAL 文件的父目录名推断；无法推断时必须指定 -db/--database。"; public static final String MESSAGE_INFERRED_TABLE_DATABASE_ARG_FROM_WAL_DIRECTORY_ARG_REPLAY_INTO_THIS_DATABASE_Y_YES_A_ACCEPT_ALL_INFERRED_DATABASES_N_QUIT_5B59D833 = "推断目标表模型数据库为 %s，来源 WAL 目录为 %s。是否向此数据库重放？[y] 同意，[a] 全部同意推断的数据库，[N] 退出："; public static final String EXCEPTION_DATABASE_CONFIRMATION_REQUIRED_FOR_WAL_DIRECTORY_ARG_INFERRED_DATABASE_ARG_SPECIFY_DB_DATABASE_OR_SKIP_DB_CONFIRMATION_WHEN_INTERACTIVE_INPUT_IS_UNAVAILABLE_14DF6D36 = "需要确认 WAL 目录 %s 的目标数据库（推断结果：%s）。无交互终端时请指定 -db/--database 或 --skip_db_confirmation。"; public static final String EXCEPTION_REPLAY_INTO_INFERRED_DATABASE_ARG_WAS_NOT_CONFIRMED_SPECIFY_DB_DATABASE_TO_SELECT_THE_TARGET_EXPLICITLY_86F81190 = "未确认向推断出的数据库 %s 重放。请使用 -db/--database 显式选择目标。"; + public static final String EXCEPTION_CANNOT_DETERMINE_THE_TARGET_DATABASE_OF_WAL_DIRECTORIES_ARG_SPECIFY_DB_DATABASE_WHICH_APPLIES_TO_ALL_IMPORTED_DIRECTORIES_55B174E4 = + "无法推断 WAL 目录 %s 的目标数据库，请使用 -db/--database 显式指定（-db 会作用于本次导入的所有目录）。"; public static final String MESSAGE_ACCEPT_ALL_INFERRED_DATABASE_NAMES_WITHOUT_CONFIRMATION_DB_DATABASE_STILL_TAKES_PRECEDENCE_FA49A73C = "自动接受所有推断出的数据库名，不再询问确认；-db/--database 仍优先。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java index ca4efdadb98..dc1a58ff012 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java @@ -329,7 +329,8 @@ public class PipeTableModelTsFileBuilderV2 extends PipeTsFileBuilder { } } - final MemTableFlushTask memTableFlushTask = new MemTableFlushTask(memTable, writer, null, null); + final MemTableFlushTask memTableFlushTask = + new MemTableFlushTask(memTable, writer, null, null, true); memTableFlushTask.syncFlushMemTable(); writer.endFile(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java index 9d9d8b0fb0a..21dd3bb9cb0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java @@ -257,7 +257,8 @@ public class PipeTreeModelTsFileBuilderV2 extends PipeTsFileBuilder { } } - final MemTableFlushTask memTableFlushTask = new MemTableFlushTask(memTable, writer, null, null); + final MemTableFlushTask memTableFlushTask = + new MemTableFlushTask(memTable, writer, null, null, false); memTableFlushTask.syncFlushMemTable(); writer.endFile(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java index f95fcfda473..f7271138724 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java @@ -816,7 +816,10 @@ class MeasurementResourceByPathUtils extends ResourceByPathUtils { return null; } IWritableMemChunk memChunk = - memTableMap.get(deviceID).getMemChunkMap().get(fullPath.getMeasurement()); + memTableMap + .get(deviceID) + .getMemChunkMap(context.isIgnoreAllNullRows()) + .get(fullPath.getMeasurement()); // check If data type matches if (memChunk.getSchema().getType() != fullPath.getMeasurementSchema().getType() && !fullPath diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java index 68727f03aa7..4a2dfd88cd6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java @@ -85,6 +85,12 @@ public class MemTableFlushTask { private String storageGroup; private String dataRegionId; + /** + * Whether rows whose values are all null should be ignored when flushing. All-null rows are only + * meaningful for table model databases, so it is the negation of {@code isTableModel}. + */ + private final boolean ignoreAllNullRows; + private IMemTable memTable; private volatile long memSerializeTime = 0L; @@ -97,16 +103,21 @@ public class MemTableFlushTask { * @param memTable the memTable to flush * @param writer the writer where memTable will be flushed to (current tsfile writer or vm writer) * @param storageGroup current database + * @param dataRegionId current data region id + * @param isTableModel whether the memTable belongs to a table model database, it cannot be + * inferred when the caller does not know the database, e.g. the pipe tsfile builders */ public MemTableFlushTask( IMemTable memTable, RestorableTsFileIOWriter writer, String storageGroup, - String dataRegionId) { + String dataRegionId, + boolean isTableModel) { this.memTable = memTable; this.writer = writer; this.storageGroup = storageGroup; this.dataRegionId = dataRegionId; + this.ignoreAllNullRows = !isTableModel; this.encodingTaskFuture = SUB_TASK_POOL_MANAGER.submit(encodingTask); this.ioTaskFuture = SUB_TASK_POOL_MANAGER.submit(ioTask); @@ -160,9 +171,10 @@ public class MemTableFlushTask { // sort the IDeviceID in lexicographical order Collections.sort(deviceIDList); for (IDeviceID deviceID : deviceIDList) { - final Map<String, IWritableMemChunk> value = memTableMap.get(deviceID).getMemChunkMap(); + final Map<String, IWritableMemChunk> value = + memTableMap.get(deviceID).getMemChunkMap(ignoreAllNullRows); // skip the empty device/chunk group - if (memTableMap.get(deviceID).isEmpty() || value.isEmpty()) { + if (memTableMap.get(deviceID).isEmpty(ignoreAllNullRows) || value.isEmpty()) { continue; } encodingTaskQueue.put(new StartFlushGroupIOTask(deviceID)); @@ -171,7 +183,7 @@ public class MemTableFlushTask { for (String seriesId : seriesInOrder) { long startTime = System.currentTimeMillis(); IWritableMemChunk series = value.get(seriesId); - if (series.count() == 0) { + if (series.count(ignoreAllNullRows) == 0) { continue; } /* @@ -278,7 +290,7 @@ public class MemTableFlushTask { if (writableMemChunk instanceof AlignedWritableMemChunk && times == null) { times = new long[MAX_NUMBER_OF_POINTS_IN_PAGE]; } - writableMemChunk.encode(ioTaskQueue, encodeInfo, times); + writableMemChunk.encode(ioTaskQueue, encodeInfo, times, ignoreAllNullRows); long subTaskTime = System.currentTimeMillis() - starTime; WRITING_METRICS.recordFlushSubTaskCost(WritingMetrics.ENCODING_TASK, subTaskTime); memSerializeTime += subTaskTime; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index 35a3696630e..2bdc6c29035 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -187,7 +187,7 @@ public abstract class AbstractMemTable implements IMemTable { deviceId, k -> { seriesNumber += filteredSchemaList.size(); - return new AlignedWritableMemChunkGroup(filteredSchemaList, k.isTableModel()); + return new AlignedWritableMemChunkGroup(filteredSchemaList); }); for (IMeasurementSchema schema : schemaList) { if (schema != null && !memChunkGroup.contains(schema.getMeasurementName())) { @@ -562,10 +562,10 @@ public abstract class AbstractMemTable implements IMemTable { } @Override - public long size() { + public long size(boolean ignoreAllNullRows) { long sum = 0; for (IWritableMemChunkGroup writableMemChunkGroup : memTableMap.values()) { - sum += writableMemChunkGroup.count(); + sum += writableMemChunkGroup.count(ignoreAllNullRows); } return sum; } @@ -617,6 +617,7 @@ public abstract class AbstractMemTable implements IMemTable { Filter globalTimeFilter) { IDeviceID deviceID = fullPath.getDeviceId(); + // Rows whose values are all null can never satisfy this read path, so they are always ignored. if (fullPath instanceof NonAlignedFullPath) { String measurementId = ((NonAlignedFullPath) fullPath).getMeasurement(); @@ -637,6 +638,7 @@ public abstract class AbstractMemTable implements IMemTable { chunkMetaDataMap, memChunkHandleMap, deletionList, + true, globalTimeFilter); } else { // check If MemTable Contains this path @@ -660,6 +662,7 @@ public abstract class AbstractMemTable implements IMemTable { chunkMetaDataMap, memChunkHandleMap, deletionList, + true, globalTimeFilter); } } @@ -682,6 +685,7 @@ public abstract class AbstractMemTable implements IMemTable { } IWritableMemChunkGroup writableMemChunkGroup = memTableMap.get(deviceID); + // Rows whose values are all null can never satisfy this read path, so they are always ignored. if (isAligned) { getMemAlignedChunkHandleFromMemTable( deviceID, @@ -690,6 +694,7 @@ public abstract class AbstractMemTable implements IMemTable { memChunkHandleMap, ttlLowerBound, modsToMemTabled, + true, globalTimeFilter); } else { getMemChunkHandleFromMemTable( @@ -699,6 +704,7 @@ public abstract class AbstractMemTable implements IMemTable { memChunkHandleMap, ttlLowerBound, modsToMemTabled, + true, globalTimeFilter); } } @@ -709,10 +715,12 @@ public abstract class AbstractMemTable implements IMemTable { Map<String, List<IChunkMetadata>> chunkMetadataMap, Map<String, List<IChunkHandle>> memChunkHandleMap, List<TimeRange> deletionList, + boolean ignoreAllNullRows, Filter globalTimeFilter) { WritableMemChunk memChunk = - (WritableMemChunk) memTableMap.get(deviceID).getMemChunkMap().get(measurementId); + (WritableMemChunk) + memTableMap.get(deviceID).getMemChunkMap(ignoreAllNullRows).get(measurementId); if (memChunk == null) { return; @@ -740,6 +748,7 @@ public abstract class AbstractMemTable implements IMemTable { Map<String, List<IChunkMetadata>> chunkMetadataList, Map<String, List<IChunkHandle>> memChunkHandleMap, List<List<TimeRange>> deletionList, + boolean ignoreAllNullRows, Filter globalTimeFilter) { AlignedWritableMemChunk alignedMemChunk = @@ -758,7 +767,8 @@ public abstract class AbstractMemTable implements IMemTable { List<BitMap> bitMaps = new ArrayList<>(); long[] timestamps = - alignedMemChunk.getAnySatisfiedTimestamp(deletionList, bitMaps, true, globalTimeFilter); + alignedMemChunk.getAnySatisfiedTimestamp( + deletionList, bitMaps, ignoreAllNullRows, globalTimeFilter); if (timestamps.length == 0) { return; } @@ -780,6 +790,7 @@ public abstract class AbstractMemTable implements IMemTable { Map<String, List<IChunkHandle>> memChunkHandleMap, long ttlLowerBound, List<Pair<ModEntry, IMemTable>> modsToMemTabled, + boolean ignoreAllNullRows, Filter globalTimeFilter) { AlignedWritableMemChunk memChunk = writableMemChunkGroup.getAlignedMemChunk(); @@ -796,7 +807,8 @@ public abstract class AbstractMemTable implements IMemTable { List<BitMap> bitMaps = new ArrayList<>(); long[] timestamps = - memChunk.getAnySatisfiedTimestamp(deletionList, bitMaps, true, globalTimeFilter); + memChunk.getAnySatisfiedTimestamp( + deletionList, bitMaps, ignoreAllNullRows, globalTimeFilter); if (timestamps.length == 0) { return; } @@ -817,10 +829,11 @@ public abstract class AbstractMemTable implements IMemTable { Map<String, List<IChunkHandle>> memChunkHandleMap, long ttlLowerBound, List<Pair<ModEntry, IMemTable>> modsToMemTabled, + boolean ignoreAllNullRows, Filter globalTimeFilter) { for (Entry<String, IWritableMemChunk> entry : - writableMemChunkGroup.getMemChunkMap().entrySet()) { + writableMemChunkGroup.getMemChunkMap(ignoreAllNullRows).entrySet()) { String measurementId = entry.getKey(); WritableMemChunk writableMemChunk = (WritableMemChunk) entry.getValue(); @@ -921,7 +934,7 @@ public abstract class AbstractMemTable implements IMemTable { } @Override - public long delete(ModEntry modEntry) { + public long delete(ModEntry modEntry, boolean ignoreAllNullRows) { List<Pair<IDeviceID, IWritableMemChunkGroup>> targetDeviceList = new ArrayList<>(); for (Entry<IDeviceID, IWritableMemChunkGroup> entry : memTableMap.entrySet()) { if (modEntry.affects(entry.getKey())) { @@ -932,11 +945,11 @@ public abstract class AbstractMemTable implements IMemTable { long pointDeleted = 0; for (Pair<IDeviceID, IWritableMemChunkGroup> pair : targetDeviceList) { if (modEntry.affectsAll(pair.left)) { - pointDeleted += pair.right.deleteTime(modEntry); + pointDeleted += pair.right.deleteTime(modEntry, ignoreAllNullRows); } else { - pointDeleted += pair.right.delete(modEntry); + pointDeleted += pair.right.delete(modEntry, ignoreAllNullRows); } - if (pair.right.isEmpty()) { + if (pair.right.isEmpty(ignoreAllNullRows)) { memTableMap.remove(pair.left).release(); } } @@ -1087,15 +1100,13 @@ public abstract class AbstractMemTable implements IMemTable { IWritableMemChunkGroup memChunkGroup; if (multiTvListInMemChunk) { if (isAligned) { - memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, deviceID.isTableModel()); + memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream); } else { memChunkGroup = WritableMemChunkGroup.deserialize(stream); } } else { if (isAligned) { - memChunkGroup = - AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks( - stream, deviceID.isTableModel()); + memChunkGroup = AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks(stream); } else { memChunkGroup = WritableMemChunkGroup.deserializeSingleTVListMemChunks(stream); } @@ -1130,15 +1141,13 @@ public abstract class AbstractMemTable implements IMemTable { IWritableMemChunkGroup memChunkGroup; if (multiTvListInMemChunk) { if (isAligned) { - memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream, deviceID.isTableModel()); + memChunkGroup = AlignedWritableMemChunkGroup.deserialize(stream); } else { memChunkGroup = WritableMemChunkGroup.deserialize(stream); } } else { if (isAligned) { - memChunkGroup = - AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks( - stream, deviceID.isTableModel()); + memChunkGroup = AlignedWritableMemChunkGroup.deserializeSingleTVListMemChunks(stream); } else { memChunkGroup = WritableMemChunkGroup.deserializeSingleTVListMemChunks(stream); } @@ -1148,11 +1157,12 @@ public abstract class AbstractMemTable implements IMemTable { } @Override - public Map<IDeviceID, Long> getMaxTime() { + public Map<IDeviceID, Long> getMaxTime(boolean ignoreAllNullRows) { Map<IDeviceID, Long> latestTimeForEachDevice = new HashMap<>(); for (Entry<IDeviceID, IWritableMemChunkGroup> entry : memTableMap.entrySet()) { - if (entry.getValue().count() > 0 && !entry.getValue().isEmpty()) { - latestTimeForEachDevice.put(entry.getKey(), entry.getValue().getMaxTime()); + if (entry.getValue().count(ignoreAllNullRows) > 0 + && !entry.getValue().isEmpty(ignoreAllNullRows)) { + latestTimeForEachDevice.put(entry.getKey(), entry.getValue().getMaxTime(ignoreAllNullRows)); } } return latestTimeForEachDevice; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java index 4f2bb0abd11..327944c2975 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java @@ -222,7 +222,7 @@ public abstract class AbstractWritableMemChunk implements IWritableMemChunk { TSStatus[] results); @Override - public abstract long count(); + public abstract long count(boolean ignoreAllNullRows); @Override public abstract long rowCount(); @@ -284,13 +284,16 @@ public abstract class AbstractWritableMemChunk implements IWritableMemChunk { @Override public abstract void encode( - BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo, long[] times); + BlockingQueue<Object> ioTaskQueue, + BatchEncodeInfo encodeInfo, + long[] times, + boolean ignoreAllNullRows); @Override public abstract void release(); @Override - public abstract boolean isEmpty(); + public abstract boolean isEmpty(boolean ignoreAllNullRows); @Override public abstract List<? extends TVList> getSortedList(); 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 ebdc024aac8..54b481de66a 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 @@ -75,7 +75,6 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { private AlignedTVList list; private List<AlignedTVList> sortedList; private long sortedRowCount = 0; - private final boolean ignoreAllNullRows; private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); private final int TVLIST_SORT_THRESHOLD = CONFIG.getTvListSortThreshold(); @@ -84,14 +83,12 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { private EncryptParameter encryptParameter; - public AlignedWritableMemChunk(List<IMeasurementSchema> schemaList, boolean isTableModel) { - this(schemaList, isTableModel, EncryptUtils.getEncryptParameter()); + public AlignedWritableMemChunk(List<IMeasurementSchema> schemaList) { + this(schemaList, EncryptUtils.getEncryptParameter()); } public AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, - boolean isTableModel, - EncryptParameter encryptParameter) { + List<IMeasurementSchema> schemaList, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); this.dataTypes = new ArrayList<>(); this.schemaList = schemaList; @@ -101,19 +98,16 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } this.list = AlignedTVList.newAlignedList(dataTypes); this.sortedList = new ArrayList<>(); - this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; } - private AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel) { - this(schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); + private AlignedWritableMemChunk(List<IMeasurementSchema> schemaList, AlignedTVList list) { + this(schemaList, list, EncryptUtils.getEncryptParameter()); } private AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, AlignedTVList list, - boolean isTableModel, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); this.schemaList = schemaList; @@ -123,7 +117,6 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.list = list; this.dataTypes = list.getTsDataTypes(); this.sortedList = new ArrayList<>(); - this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; } @@ -465,7 +458,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } @Override - public long count() { + public long count(boolean ignoreAllNullRows) { if (!ignoreAllNullRows && measurementIndexMap.isEmpty()) { return rowCount(); } @@ -487,8 +480,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } @Override - public long getMaxTime() { - if (isEmpty()) { + public long getMaxTime(boolean ignoreAllNullRows) { + if (isEmpty(ignoreAllNullRows)) { return Long.MIN_VALUE; } long maxTime = list.getMaxTime(); @@ -582,7 +575,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { public void encodeWorkingAlignedTVList( BlockingQueue<Object> ioTaskQueue, long maxNumberOfPointsInChunk, - int maxNumberOfPointsInPage) { + int maxNumberOfPointsInPage, + boolean ignoreAllNullRows) { List<IMeasurementSchema> activeSchemaList = getActiveSchemaList(); if (activeSchemaList.isEmpty() && ignoreAllNullRows) { return; @@ -1065,7 +1059,11 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } @Override - public void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo, long[] times) { + public void encode( + BlockingQueue<Object> ioTaskQueue, + BatchEncodeInfo encodeInfo, + long[] times, + boolean ignoreAllNullRows) { encodeInfo.maxNumberOfPointsInChunk = Math.min( encodeInfo.maxNumberOfPointsInChunk, @@ -1073,7 +1071,10 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { if (TVLIST_SORT_THRESHOLD == 0) { encodeWorkingAlignedTVList( - ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); + ioTaskQueue, + encodeInfo.maxNumberOfPointsInChunk, + encodeInfo.maxNumberOfPointsInPage, + ignoreAllNullRows); return; } @@ -1151,15 +1152,15 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } @Override - public long getLastPoint() { + public long getLastPoint(boolean ignoreAllNullRows) { if (rowCount() == 0) { return Long.MIN_VALUE; } - return getMaxTime(); + return getMaxTime(ignoreAllNullRows); } @Override - public boolean isEmpty() { + public boolean isEmpty(boolean ignoreAllNullRows) { if (rowCount() == 0) { return true; } @@ -1215,8 +1216,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { list.serializeToWAL(buffer); } - public static AlignedWritableMemChunk deserialize(DataInputStream stream, boolean isTableModel) - throws IOException { + public static AlignedWritableMemChunk deserialize(DataInputStream stream) throws IOException { int schemaListSize = stream.readInt(); List<IMeasurementSchema> schemaList = new ArrayList<>(schemaListSize); for (int i = 0; i < schemaListSize; i++) { @@ -1230,13 +1230,13 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { sortedList.add(tvList); } AlignedTVList list = AlignedTVList.deserialize(stream); - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemaList, list, isTableModel); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemaList, list); chunk.sortedList = sortedList; return chunk; } public static AlignedWritableMemChunk deserializeSingleTVListMemChunks( - DataInputStream stream, boolean isTableModel) throws IOException { + DataInputStream stream) throws IOException { int schemaListSize = stream.readInt(); List<IMeasurementSchema> schemaList = new ArrayList<>(schemaListSize); for (int i = 0; i < schemaListSize; i++) { @@ -1245,7 +1245,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } AlignedTVList list = AlignedTVList.deserialize(stream); - return new AlignedWritableMemChunk(schemaList, list, isTableModel); + return new AlignedWritableMemChunk(schemaList, list); } public List<IMeasurementSchema> getSchemaList() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java index a27973acdb1..e1235ca935b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java @@ -46,13 +46,13 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { private EncryptParameter encryptParameter; - public AlignedWritableMemChunkGroup(List<IMeasurementSchema> schemaList, boolean isTableModel) { - memChunk = new AlignedWritableMemChunk(schemaList, isTableModel); + public AlignedWritableMemChunkGroup(List<IMeasurementSchema> schemaList) { + memChunk = new AlignedWritableMemChunk(schemaList); } @TestOnly public AlignedWritableMemChunkGroup( - AlignedWritableMemChunk memChunk, List<IMeasurementSchema> schemaList, boolean isTableModel) { + AlignedWritableMemChunk memChunk, List<IMeasurementSchema> schemaList) { this.memChunk = memChunk; } @@ -78,8 +78,8 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public long count() { - return memChunk.count(); + public long count(boolean ignoreAllNullRows) { + return memChunk.count(ignoreAllNullRows); } /** @@ -101,20 +101,20 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public Map<String, IWritableMemChunk> getMemChunkMap() { - if (memChunk.count() == 0) { + public Map<String, IWritableMemChunk> getMemChunkMap(boolean ignoreAllNullRows) { + if (memChunk.count(ignoreAllNullRows) == 0) { return Collections.emptyMap(); } return Collections.singletonMap("", memChunk); } @Override - public boolean isEmpty() { - return memChunk.isEmpty() || memChunk.isAllDeleted(); + public boolean isEmpty(boolean ignoreAllNullRows) { + return memChunk.isEmpty(ignoreAllNullRows) || memChunk.isAllDeleted(); } @Override - public long delete(ModEntry modEntry) { + public long delete(ModEntry modEntry, boolean ignoreAllNullRows) { int deletedPointsNumber = 0; Set<String> measurements = memChunk.getAllMeasurements(); List<String> columnsToBeRemoved = new ArrayList<>(); @@ -138,7 +138,7 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { return deletedPointsNumber; } - public long deleteTime(ModEntry modEntry) { + public long deleteTime(ModEntry modEntry, boolean ignoreAllNullRows) { return memChunk.deleteTime(modEntry.getStartTime(), modEntry.getEndTime()); } @@ -148,8 +148,8 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public long getMaxTime() { - return memChunk.getMaxTime(); + public long getMaxTime(boolean ignoreAllNullRows) { + return memChunk.getMaxTime(ignoreAllNullRows); } public AlignedWritableMemChunk getAlignedMemChunk() { @@ -172,10 +172,10 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { memChunk.setEncryptParameter(encryptParameter); } - protected static AlignedWritableMemChunkGroup deserialize( - DataInputStream stream, boolean isTableModel) throws IOException { + protected static AlignedWritableMemChunkGroup deserialize(DataInputStream stream) + throws IOException { AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup(); - memChunkGroup.memChunk = AlignedWritableMemChunk.deserialize(stream, isTableModel); + memChunkGroup.memChunk = AlignedWritableMemChunk.deserialize(stream); return memChunkGroup; } @@ -185,10 +185,9 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { } protected static AlignedWritableMemChunkGroup deserializeSingleTVListMemChunks( - DataInputStream stream, boolean isTableModel) throws IOException { + DataInputStream stream) throws IOException { AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup(); - memChunkGroup.memChunk = - AlignedWritableMemChunk.deserializeSingleTVListMemChunks(stream, isTableModel); + memChunkGroup.memChunk = AlignedWritableMemChunk.deserializeSingleTVListMemChunks(stream); return memChunkGroup; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java index afa24083ff6..48d43644c8d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java @@ -68,9 +68,11 @@ public interface IMemTable extends WALEntryValue { Object[] objectValue); /** + * @param ignoreAllNullRows whether rows whose values are all null should be ignored, it is + * determined by the data model of the database the memtable belongs to * @return the number of points */ - long size(); + long size(boolean ignoreAllNullRows); /** * @return memory usage @@ -170,7 +172,7 @@ public interface IMemTable extends WALEntryValue { * * @return */ - long delete(ModEntry modEntry); + long delete(ModEntry modEntry, boolean ignoreAllNullRows); /** * Make a copy of this MemTable. @@ -214,7 +216,7 @@ public interface IMemTable extends WALEntryValue { void setFlushStatus(FlushStatus flushStatus); - Map<IDeviceID, Long> getMaxTime(); + Map<IDeviceID, Long> getMaxTime(boolean ignoreAllNullRows); String getDatabase(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java index e57f30680f9..71f205ef601 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java @@ -84,7 +84,7 @@ public interface IWritableMemChunk extends WALEntryValue { int end, TSStatus[] results); - long count(); + long count(boolean ignoreAllNullRows); long rowCount(); @@ -100,7 +100,7 @@ public interface IWritableMemChunk extends WALEntryValue { void releaseTemporaryTvListForFlush(); - default long getMaxTime() { + default long getMaxTime(boolean ignoreAllNullRows) { return Long.MAX_VALUE; } @@ -115,15 +115,19 @@ public interface IWritableMemChunk extends WALEntryValue { IChunkWriter createIChunkWriter(); - void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo, long[] times); + void encode( + BlockingQueue<Object> ioTaskQueue, + BatchEncodeInfo encodeInfo, + long[] times, + boolean ignoreAllNullRows); void release(); long getFirstPoint(); - long getLastPoint(); + long getLastPoint(boolean ignoreAllNullRows); - boolean isEmpty(); + boolean isEmpty(boolean ignoreAllNullRows); List<? extends TVList> getSortedList(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java index 46a21f05840..0cafd436329 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java @@ -47,21 +47,21 @@ public interface IWritableMemChunkGroup extends WALEntryValue { void release(); - long count(); + long count(boolean ignoreAllNullRows); boolean contains(String measurement); - Map<String, IWritableMemChunk> getMemChunkMap(); + Map<String, IWritableMemChunk> getMemChunkMap(boolean ignoreAllNullRows); - boolean isEmpty(); + boolean isEmpty(boolean ignoreAllNullRows); - long delete(ModEntry modEntry); + long delete(ModEntry modEntry, boolean ignoreAllNullRows); - long deleteTime(ModEntry modEntry); + long deleteTime(ModEntry modEntry, boolean ignoreAllNullRows); IWritableMemChunk getWritableMemChunk(String measurement); - long getMaxTime(); + long getMaxTime(boolean ignoreAllNullRows); void setEncryptParameter(EncryptParameter encryptParameter); 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 78c48a2702b..7a89a07a9af 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 @@ -1584,6 +1584,11 @@ public class TsFileProcessor { } } + /** Whether the data region of this processor belongs to a table model database. */ + private boolean isTableModel() { + return dataRegionInfo.getDataRegion().isTableModel(); + } + /** * Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which * <= 'timestamp' in the deletion. <br> @@ -1599,7 +1604,7 @@ public class TsFileProcessor { try { boolean deleted = false; if (workMemTable != null) { - long pointDeleted = workMemTable.delete(deletion); + long pointDeleted = workMemTable.delete(deletion, !isTableModel()); logger.info( StorageEngineMessages .STORAGE_LOG_DELETION_DELETION_WITH_IN_WORKMEMTABLE_POINTS_DELETED_00EA995A, @@ -1767,7 +1772,7 @@ public class TsFileProcessor { * flushManager again. */ private Future<?> addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOException { - final Map<IDeviceID, Long> lastTimeForEachDevice = tobeFlushed.getMaxTime(); + final Map<IDeviceID, Long> lastTimeForEachDevice = tobeFlushed.getMaxTime(!isTableModel()); // 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 @@ -1911,7 +1916,8 @@ public class TsFileProcessor { memTableToFlush, writer, dataRegionName, - dataRegionInfo.getDataRegion().getDataRegionIdString()); + dataRegionInfo.getDataRegion().getDataRegionIdString(), + isTableModel()); flushTask.syncFlushMemTable(); memTableFlushPointCount = memTableToFlush.getTotalPointsNum(); } catch (Throwable e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java index 7e54296c6a8..929108e9246 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java @@ -222,7 +222,12 @@ public class WritableMemChunk extends AbstractWritableMemChunk { } @Override - public long count() { + public long count(boolean ignoreAllNullRows) { + // a non-aligned chunk never stores all-null rows, so ignoreAllNullRows makes no difference + return pointCount(); + } + + private long pointCount() { long count = list.count(); for (TVList tvList : sortedList) { count += tvList.count(); @@ -241,7 +246,7 @@ public class WritableMemChunk extends AbstractWritableMemChunk { } @Override - public long getMaxTime() { + public long getMaxTime(boolean ignoreAllNullRows) { long maxTime = list.getMaxTime(); for (TVList tvList : sortedList) { maxTime = Math.max(maxTime, tvList.getMaxTime()); @@ -260,23 +265,23 @@ public class WritableMemChunk extends AbstractWritableMemChunk { @Override public long getFirstPoint() { - if (count() == 0) { + if (pointCount() == 0) { return Long.MAX_VALUE; } return getMinTime(); } @Override - public long getLastPoint() { - if (count() == 0) { + public long getLastPoint(boolean ignoreAllNullRows) { + if (pointCount() == 0) { return Long.MIN_VALUE; } - return getMaxTime(); + return getMaxTime(ignoreAllNullRows); } @Override - public boolean isEmpty() { - return count() == 0; + public boolean isEmpty(boolean ignoreAllNullRows) { + return pointCount() == 0; } @Override @@ -389,7 +394,11 @@ public class WritableMemChunk extends AbstractWritableMemChunk { } @Override - public void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo, long[] times) { + public void encode( + BlockingQueue<Object> ioTaskQueue, + BatchEncodeInfo encodeInfo, + long[] times, + boolean ignoreAllNullRows) { if (TVLIST_SORT_THRESHOLD == 0) { encodeWorkingTVList( ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.targetChunkSize); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java index 6f5c63824c4..9f27648b0ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java @@ -93,10 +93,10 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public long count() { + public long count(boolean ignoreAllNullRows) { long count = 0; for (IWritableMemChunk memChunk : memChunkMap.values()) { - count += memChunk.count(); + count += memChunk.count(ignoreAllNullRows); } return count; } @@ -118,17 +118,19 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public Map<String, IWritableMemChunk> getMemChunkMap() { + public Map<String, IWritableMemChunk> getMemChunkMap(boolean ignoreAllNullRows) { + // a group of non-aligned chunks never stores all-null rows, so ignoreAllNullRows makes no + // difference return memChunkMap; } @Override - public boolean isEmpty() { - return memChunkMap.isEmpty() || count() == 0; + public boolean isEmpty(boolean ignoreAllNullRows) { + return memChunkMap.isEmpty() || count(ignoreAllNullRows) == 0; } @Override - public long delete(ModEntry modEntry) { + public long delete(ModEntry modEntry, boolean ignoreAllNullRows) { Iterator<Entry<String, IWritableMemChunk>> iter = memChunkMap.entrySet().iterator(); long deletedPointsNumber = 0; while (iter.hasNext()) { @@ -140,11 +142,11 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup { IWritableMemChunk chunk = entry.getValue(); if (modEntry.getStartTime() == Long.MIN_VALUE && modEntry.getEndTime() == Long.MAX_VALUE) { iter.remove(); - deletedPointsNumber += chunk.count(); + deletedPointsNumber += chunk.count(ignoreAllNullRows); chunk.release(); } else { deletedPointsNumber += chunk.delete(modEntry.getStartTime(), modEntry.getEndTime()); - if (chunk.count() == 0) { + if (chunk.count(ignoreAllNullRows) == 0) { iter.remove(); } } @@ -153,8 +155,8 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public long deleteTime(ModEntry modEntry) { - return delete(modEntry); + public long deleteTime(ModEntry modEntry, boolean ignoreAllNullRows) { + return delete(modEntry, ignoreAllNullRows); } @Override @@ -166,10 +168,10 @@ public class WritableMemChunkGroup implements IWritableMemChunkGroup { } @Override - public long getMaxTime() { + public long getMaxTime(boolean ignoreAllNullRows) { long maxTime = Long.MIN_VALUE; for (IWritableMemChunk memChunk : memChunkMap.values()) { - maxTime = Math.max(maxTime, memChunk.getMaxTime()); + maxTime = Math.max(maxTime, memChunk.getMaxTime(ignoreAllNullRows)); } return maxTime; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java index 221d39cdfcc..8e191d475ba 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; @@ -55,11 +56,15 @@ import java.util.Optional; */ public class TsFilePlanRedoer { private final TsFileResource tsFileResource; + // whether rows whose values are all null should be ignored, which depends on the data model of + // the database this tsfile belongs to + private final boolean ignoreAllNullRows; // store data when redoing logs private IMemTable recoveryMemTable; public TsFilePlanRedoer(TsFileResource tsFileResource) { this.tsFileResource = tsFileResource; + this.ignoreAllNullRows = !PathUtils.isTableModelDatabase(tsFileResource.getDatabaseName()); this.recoveryMemTable = new PrimitiveMemTable(tsFileResource.getDatabaseName(), tsFileResource.getDataRegionId()); WritingMetrics.getInstance().recordActiveMemTableCount(tsFileResource.getDataRegionId(), 1); @@ -73,7 +78,7 @@ public class TsFilePlanRedoer { TreeDeletionEntry deletionEntry = new TreeDeletionEntry( path, deleteDataNode.getDeleteStartTime(), deleteDataNode.getDeleteEndTime()); - recoveryMemTable.delete(deletionEntry); + recoveryMemTable.delete(deletionEntry, ignoreAllNullRows); deletionEntries.add(deletionEntry); } tsFileResource.getModFileForWrite().write(deletionEntries); @@ -81,7 +86,7 @@ public class TsFilePlanRedoer { void redoDelete(RelationalDeleteDataNode node) throws IOException { for (TableDeletionEntry modEntry : node.getModEntries()) { - recoveryMemTable.delete(modEntry); + recoveryMemTable.delete(modEntry, ignoreAllNullRows); } tsFileResource.getModFileForWrite().write(node.getModEntries()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java index abce4d0ec03..d8369e2c406 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java @@ -192,13 +192,15 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform new DeletionPredicate(device.getTableName(), new FullExactMatch(device)), new TimeRange( tsFileResource.getStartTime(device).get(), - tsFileResource.getEndTime(device).get()))); + tsFileResource.getEndTime(device).get())), + !isTableModel); } else { memTable.delete( new TreeDeletionEntry( new MeasurementPath(device, "*"), tsFileResource.getStartTime(device).get(), - tsFileResource.getEndTime(device).get())); + tsFileResource.getEndTime(device).get()), + !isTableModel); } } } @@ -269,10 +271,10 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform for (Map.Entry<IDeviceID, IWritableMemChunkGroup> deviceEntry : memTableMap.entrySet()) { IDeviceID deviceId = deviceEntry.getKey(); for (Map.Entry<String, IWritableMemChunk> measurementEntry : - deviceEntry.getValue().getMemChunkMap().entrySet()) { + deviceEntry.getValue().getMemChunkMap(!isTableModel).entrySet()) { IWritableMemChunk memChunk = measurementEntry.getValue(); tsFileResource.updateStartTime(deviceId, memChunk.getFirstPoint()); - tsFileResource.updateEndTime(deviceId, memChunk.getLastPoint()); + tsFileResource.updateEndTime(deviceId, memChunk.getLastPoint(!isTableModel)); } } // flush memTable @@ -283,7 +285,8 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform recoveryMemTable, writer, databaseName + FILE_NAME_SEPARATOR + dataRegionId, - dataRegionId); + dataRegionId, + isTableModel); tableFlushTask.syncFlushMemTable(); tsFileResource.updatePlanIndexes(recoveryMemTable.getMinPlanIndex()); tsFileResource.updatePlanIndexes(recoveryMemTable.getMaxPlanIndex()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java index 265c027e8ac..005b78c5a0c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java @@ -163,6 +163,7 @@ public class ImportWAL { .MESSAGE_INFERRED_TABLE_DATABASE_ARG_FROM_WAL_DIRECTORY_ARG_REPLAY_INTO_THIS_DATABASE_Y_YES_A_ACCEPT_ALL_INFERRED_DATABASES_N_QUIT_5B59D833, inferredDatabase, directory)); + requireResolvedDatabases(directoryDatabases); final ReplayStatistics statistics = replayWALDirectories( walFiles, @@ -255,7 +256,7 @@ public class ImportWAL { .hasArg() .desc( ImportWALMessages - .MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_AND_ASK_FOR_CONFIRMATION_4B1E409D) + .MESSAGE_TARGET_DATABASE_FOR_WAL_REPLAY_IF_OMITTED_INFER_FROM_THE_WAL_PARENT_DIRECTORY_IF_INFERENCE_FAILS_DB_DATABASE_IS_REQUIRED_6EEBC019) .build()); options.addOption( Option.builder() @@ -444,8 +445,9 @@ public class ImportWAL { final WALReplayer.ReplayDecisionController replayDecisionController) throws IOException { final Session treeSession = createSession(host, port, username, password, null); + final boolean isTableModel = database != null && PathUtils.isTableModelDatabase(database); final Session tableSession = - database == null ? null : createSession(host, port, username, password, database); + isTableModel ? createSession(host, port, username, password, database) : null; try { treeSession.open(false); if (tableSession != null) { @@ -576,6 +578,26 @@ public class ImportWAL { return databases; } + /** + * The data model of every directory must be known before any replay starts, otherwise a WAL + * snapshot could only be replayed with a guessed model. + */ + static void requireResolvedDatabases(final Map<Path, String> directoryDatabases) { + final List<String> unresolvedDirectories = new ArrayList<>(); + for (final Map.Entry<Path, String> entry : directoryDatabases.entrySet()) { + if (entry.getValue() == null) { + unresolvedDirectories.add(entry.getKey().toString()); + } + } + if (!unresolvedDirectories.isEmpty()) { + throw new IllegalArgumentException( + String.format( + ImportWALMessages + .EXCEPTION_CANNOT_DETERMINE_THE_TARGET_DATABASE_OF_WAL_DIRECTORIES_ARG_SPECIFY_DB_DATABASE_WHICH_APPLIES_TO_ALL_IMPORTED_DIRECTORIES_55B174E4, + String.join(", ", unresolvedDirectories))); + } + } + private static final Comparator<Path> WAL_FILE_COMPARATOR = Comparator.comparing((Path path) -> Objects.toString(path.getParent(), "")) .thenComparingLong(ImportWAL::getWALVersion) @@ -936,6 +958,7 @@ public class ImportWAL { private final Session treeSession; private final Session tableSession; + private final boolean isTableModel; private final ConsensusLogToTabletConverter converter; private final ReplayDecisionPrompt replayDecisionPrompt; private final Map<String, TableSchema> tableSchemas = new HashMap<>(); @@ -949,6 +972,15 @@ public class ImportWAL { new ReplayDecisionController(System.console())); } + /** + * @param treeSession the session which replays tree model entries, it carries no database + * @param tableSession the session which replays table model entries, null when no table + * database is declared + * @param tableDatabaseName the target database resolved before replay; null means no table + * database is declared, so the entries are replayed with the tree model only + * @param replayDecisionPrompt the prompt which decides how to handle entries that cannot be + * replayed automatically + */ WALReplayer( final Session treeSession, final Session tableSession, @@ -956,10 +988,12 @@ public class ImportWAL { final ReplayDecisionPrompt replayDecisionPrompt) { this.treeSession = treeSession; this.tableSession = tableSession; + this.isTableModel = + tableDatabaseName != null && PathUtils.isTableModelDatabase(tableDatabaseName); this.replayDecisionPrompt = replayDecisionPrompt; converter = new ConsensusLogToTabletConverter( - null, null, ColumnFilterMatcher.matchAll(), tableDatabaseName); + null, null, ColumnFilterMatcher.matchAll(), isTableModel ? tableDatabaseName : null); } @Override @@ -972,9 +1006,8 @@ public class ImportWAL { throws IoTDBConnectionException, StatementExecutionException { if (entry.getType() == WALEntryType.MEMORY_TABLE_SNAPSHOT || entry.getType() == WALEntryType.OLD_MEMORY_TABLE_SNAPSHOT) { - return replayMemTableSnapshot((IMemTable) entry.getValue()) - ? ReplayResult.REPLAYED - : ReplayResult.IGNORED; + final IMemTable memTable = (IMemTable) entry.getValue(); + return replayMemTableSnapshot(memTable) ? ReplayResult.REPLAYED : ReplayResult.IGNORED; } if (entry.getValue() instanceof InsertNode insertNode) { replayInsert(insertNode); @@ -1318,8 +1351,8 @@ public class ImportWAL { memTable.getMemTableMap().entrySet()) { final IDeviceID deviceId = deviceEntry.getKey(); final IWritableMemChunkGroup group = deviceEntry.getValue(); - for (IWritableMemChunk chunk : group.getMemChunkMap().values()) { - if (chunk == null || chunk.isEmpty()) { + for (IWritableMemChunk chunk : group.getMemChunkMap(!isTableModel).values()) { + if (chunk == null || chunk.isEmpty(!isTableModel)) { continue; } if (chunk instanceof AlignedWritableMemChunk) { @@ -1336,8 +1369,7 @@ public class ImportWAL { final IDeviceID deviceId, final IWritableMemChunk chunk) throws IoTDBConnectionException, StatementExecutionException { final List<IMeasurementSchema> schemas = Collections.singletonList(chunk.getSchema()); - final boolean tableModel = deviceId.isTableModel(); - requireTableSessionIfNeeded(tableModel); + requireTableSessionIfNeeded(isTableModel); final TableTabletSchema tabletSchema = createTableTabletSchema(deviceId, schemas); final List<TVList> lists = new ArrayList<>(); lists.addAll(chunk.getSortedList()); @@ -1354,7 +1386,7 @@ public class ImportWAL { final int end = Math.min(start + SNAPSHOT_TABLET_ROW_LIMIT, list.rowCount()); final Tablet tablet = buildNonAlignedTablet(deviceId, tabletSchema, schemas, list, start, end); - sendTablet(tablet, tableModel, false); + sendTablet(tablet, isTableModel, false); replayed = true; } } @@ -1364,8 +1396,7 @@ public class ImportWAL { private boolean replayAlignedMemChunk( final IDeviceID deviceId, final AlignedWritableMemChunk chunk) throws IoTDBConnectionException, StatementExecutionException { - final boolean tableModel = deviceId.isTableModel(); - requireTableSessionIfNeeded(tableModel); + requireTableSessionIfNeeded(isTableModel); final List<IMeasurementSchema> schemas = chunk.getSchemaList(); final TableTabletSchema tabletSchema = createTableTabletSchema(deviceId, schemas); final List<AlignedTVList> lists = new ArrayList<>(); @@ -1384,7 +1415,7 @@ public class ImportWAL { final int end = Math.min(start + SNAPSHOT_TABLET_ROW_LIMIT, replayableRows.size()); final Tablet tablet = buildAlignedTablet(deviceId, tabletSchema, schemas, list, replayableRows, start, end); - sendTablet(tablet, tableModel, true); + sendTablet(tablet, isTableModel, true); replayed = true; } } @@ -1412,7 +1443,7 @@ public class ImportWAL { private TableTabletSchema createTableTabletSchema( final IDeviceID deviceId, final List<IMeasurementSchema> fieldSchemas) throws IoTDBConnectionException, StatementExecutionException { - if (!deviceId.isTableModel()) { + if (!isTableModel) { return new TableTabletSchema(fieldSchemas, null, 0); } final List<IMeasurementSchema> tagSchemas = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java index 79e66f92d9e..0caf1692f63 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java @@ -209,7 +209,7 @@ public class FragmentInstanceExecutionTest { new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.PLAIN))); - AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas, true); + AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas); for (int i = 0; i < 100; i++) { memChunk.putAlignedRow(i, new Object[] {(long) i, (long) i * 2, (long) i * 3}); } @@ -265,8 +265,8 @@ public class FragmentInstanceExecutionTest { IMemTable memTable = createMemTable(deviceId, measurementId); assertEquals(1, memTable.getMemTableMap().size()); IWritableMemChunkGroup memChunkGroup = memTable.getMemTableMap().values().iterator().next(); - assertEquals(1, memChunkGroup.getMemChunkMap().size()); - IWritableMemChunk memChunk = memChunkGroup.getMemChunkMap().values().iterator().next(); + assertEquals(1, memChunkGroup.getMemChunkMap(true).size()); + IWritableMemChunk memChunk = memChunkGroup.getMemChunkMap(true).values().iterator().next(); TVList tvList = memChunk.getWorkingTVList(); assertFalse(tvList.isSorted()); @@ -330,8 +330,8 @@ public class FragmentInstanceExecutionTest { assertEquals(1, memTable.getMemTableMap().size()); IWritableMemChunkGroup memChunkGroup = memTable.getMemTableMap().values().iterator().next(); - assertEquals(1, memChunkGroup.getMemChunkMap().size()); - IWritableMemChunk memChunk = memChunkGroup.getMemChunkMap().values().iterator().next(); + assertEquals(1, memChunkGroup.getMemChunkMap(true).size()); + IWritableMemChunk memChunk = memChunkGroup.getMemChunkMap(true).values().iterator().next(); AlignedTVList workingTvList = (AlignedTVList) memChunk.getWorkingTVList(); assertFalse(workingTvList.isSorted()); assertEquals(100, workingTvList.rowCount()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java index b18b543a2d5..b9b9fd53685 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java @@ -266,14 +266,14 @@ public class AlignedBitmapMemoryAccountingPerformanceTest { private static AlignedWritableMemChunk[] createMemChunks(Scenario scenario, int count) { AlignedWritableMemChunk[] memChunks = new AlignedWritableMemChunk[count]; for (int i = 0; i < count; i++) { - memChunks[i] = new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); + memChunks[i] = new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas)); } return memChunks; } private static AccountingTarget createAccountingTarget(Scenario scenario) { AlignedWritableMemChunk memChunk = - new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas), false); + new AlignedWritableMemChunk(new ArrayList<>(scenario.schemas)); memChunk.writeAlignedTablet( scenario.times, scenario.columns, @@ -290,7 +290,7 @@ public class AlignedBitmapMemoryAccountingPerformanceTest { Collections.singletonMap( deviceId, new AlignedWritableMemChunkGroup( - memChunk, new ArrayList<>(scenario.schemas), false))); + memChunk, new ArrayList<>(scenario.schemas)))); return new AccountingTarget(memTable, deviceId); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java index 5595390752a..682b4b287d2 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java @@ -239,7 +239,7 @@ public class MemChunkDeserializeTest { new MeasurementSchema("s4", TSDataType.FLOAT), new MeasurementSchema("s5", TSDataType.DOUBLE), new MeasurementSchema("s6", TSDataType.TEXT)); - AlignedWritableMemChunk series = new AlignedWritableMemChunk(schemaList, false); + AlignedWritableMemChunk series = new AlignedWritableMemChunk(schemaList); int count = 1000; for (int i = 0; i < count; i++) { @@ -261,7 +261,7 @@ public class MemChunkDeserializeTest { series.serializeToWAL(walBuffer); DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(walBuffer.getBuffer().array())); - AlignedWritableMemChunk memChunk = AlignedWritableMemChunk.deserialize(inputStream, false); + AlignedWritableMemChunk memChunk = AlignedWritableMemChunk.deserialize(inputStream); AlignedReadOnlyMemChunk readableChunk = (AlignedReadOnlyMemChunk) getAlignedReadOnlyChunk(memChunk, schemaList, measurementList); @@ -293,7 +293,7 @@ public class MemChunkDeserializeTest { series.writeNonAlignedPoint(1, 1); WritableMemChunkGroup group = new WritableMemChunkGroup(); - group.getMemChunkMap().put(measurement, series); + group.getMemChunkMap(true).put(measurement, series); WALByteBufferForTest walBuffer = new WALByteBufferForTest(ByteBuffer.allocate(group.serializedSize())); @@ -303,7 +303,7 @@ public class MemChunkDeserializeTest { DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(walBuffer.getBuffer().array())); WritableMemChunkGroup deserialized = WritableMemChunkGroup.deserialize(inputStream); - Assert.assertTrue(deserialized.getMemChunkMap().containsKey(measurement)); + Assert.assertTrue(deserialized.getMemChunkMap(true).containsKey(measurement)); } private WritableMemChunk createWritableMemChunkFromBytes(WritableMemChunk series) @@ -319,7 +319,7 @@ public class MemChunkDeserializeTest { private ReadOnlyMemChunk getReadOnlyChunk(WritableMemChunk memChunk, TSDataType dataType) throws QueryProcessException, IOException, MetadataException { WritableMemChunkGroup memChunkGroup = new WritableMemChunkGroup(); - memChunkGroup.getMemChunkMap().put("s1", memChunk); + memChunkGroup.getMemChunkMap(true).put("s1", memChunk); Map<IDeviceID, IWritableMemChunkGroup> memTableMap = new HashMap<>(); memTableMap.put(deviceID, memChunkGroup); IMemTable memTable = new PrimitiveMemTable(storageGroup, dataRegionId, memTableMap); @@ -343,7 +343,7 @@ public class MemChunkDeserializeTest { List<String> measurementList) throws QueryProcessException, IOException, MetadataException { AlignedWritableMemChunkGroup memChunkGroup = - new AlignedWritableMemChunkGroup(memChunk, schemaList, false); + new AlignedWritableMemChunkGroup(memChunk, schemaList); Map<IDeviceID, IWritableMemChunkGroup> memTableMap = new HashMap<>(); memTableMap.put(deviceID, memChunkGroup); IMemTable memTable = new PrimitiveMemTable(storageGroup, dataRegionId, memTableMap); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java index afe0384839b..ef4c520b135 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java @@ -97,7 +97,7 @@ public class MemTableFlushTaskTest { MemTableTestUtils.measurementId0, MemTableTestUtils.dataType0); MemTableFlushTask memTableFlushTask = - new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId); + new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId, true); assertTrue( writer .getVisibleMetadataList( @@ -134,7 +134,7 @@ public class MemTableFlushTaskTest { throws ExecutionException, InterruptedException, IllegalPathException, WriteProcessException { MemTableTestUtils.produceVectorData(memTable); MemTableFlushTask memTableFlushTask = - new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId); + new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId, true); assertTrue( writer .getVisibleMetadataList(MemTableTestUtils.deviceId0, "sensor0", TSDataType.BOOLEAN) @@ -162,7 +162,7 @@ public class MemTableFlushTaskTest { throws ExecutionException, InterruptedException, IllegalPathException, WriteProcessException { MemTableTestUtils.produceNullableVectorData(memTable); MemTableFlushTask memTableFlushTask = - new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId); + new MemTableFlushTask(memTable, writer, storageGroup, dataRegionId, true); assertTrue( writer .getVisibleMetadataList(MemTableTestUtils.deviceId0, "sensor0", TSDataType.BOOLEAN) @@ -192,7 +192,7 @@ public class MemTableFlushTaskTest { memChunk.sortTvListForFlush(); BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); - memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100); + memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100, true); assertFalse(memChunk.isColumnMappingBuilt()); assertFalse(ioTaskQueue.isEmpty()); @@ -206,7 +206,7 @@ public class MemTableFlushTaskTest { memChunk.sortTvListForFlush(); BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); - memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100); + memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100, true); assertTrue(memChunk.isColumnMappingBuilt()); assertFalse(ioTaskQueue.isEmpty()); @@ -228,7 +228,7 @@ public class MemTableFlushTaskTest { Arrays.asList( new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN)); - AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas, true); + AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas); String alignedFilePath = TestConstant.OUTPUT_DATA_DIR.concat( "duplicateAlignedRows-" + removeSecondMeasurement + ".tsfile"); @@ -245,7 +245,7 @@ public class MemTableFlushTaskTest { memChunk.sortTvListForFlush(); BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); - memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100); + memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100, false); try (TsFileIOWriter alignedWriter = new TsFileIOWriter(new File(alignedFilePath))) { alignedWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d")); Object task; @@ -304,7 +304,7 @@ public class MemTableFlushTaskTest { Arrays.asList( new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN)); - AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas, false); + AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas); String alignedFilePath = TestConstant.OUTPUT_DATA_DIR.concat("testAlignedFastPath.tsfile"); try { @@ -320,7 +320,7 @@ public class MemTableFlushTaskTest { memChunk.sortTvListForFlush(); BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); - memChunk.encodeWorkingAlignedTVList(ioTaskQueue, rowCount, 1024); + memChunk.encodeWorkingAlignedTVList(ioTaskQueue, rowCount, 1024, true); try (TsFileIOWriter alignedWriter = new TsFileIOWriter(new File(alignedFilePath))) { alignedWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d")); Object task; @@ -373,7 +373,7 @@ public class MemTableFlushTaskTest { Arrays.asList( new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN)); - AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas, false); + AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas); String alignedFilePath = TestConstant.OUTPUT_DATA_DIR.concat("testAlignedTimeDeletion" + pageSize + ".tsfile"); try { @@ -389,7 +389,7 @@ public class MemTableFlushTaskTest { BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); // Cover a single page and boundaries between pages and chunks. - memChunk.encodeWorkingAlignedTVList(ioTaskQueue, pageSize + 1, pageSize); + memChunk.encodeWorkingAlignedTVList(ioTaskQueue, pageSize + 1, pageSize, true); try (TsFileIOWriter alignedWriter = new TsFileIOWriter(new File(alignedFilePath))) { alignedWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d")); Object task; @@ -464,7 +464,7 @@ public class MemTableFlushTaskTest { schemas.add(new MeasurementSchema("s" + column, types.get(column), TSEncoding.PLAIN)); } schemas.add(new MeasurementSchema("empty", TSDataType.INT64, TSEncoding.PLAIN)); - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas); Object[][] expected = new Object[rowCount][types.size()]; String path = TestConstant.OUTPUT_DATA_DIR.concat("unmaterialized-" + moved + ".tsfile"); try { @@ -502,7 +502,7 @@ public class MemTableFlushTaskTest { } chunk.sortTvListForFlush(); BlockingQueue<Object> queue = new LinkedBlockingQueue<>(); - chunk.encodeWorkingAlignedTVList(queue, ARRAY_SIZE * 5 + 3, ARRAY_SIZE + 3); + chunk.encodeWorkingAlignedTVList(queue, ARRAY_SIZE * 5 + 3, ARRAY_SIZE + 3, false); try (TsFileIOWriter fileWriter = new TsFileIOWriter(new File(path))) { fileWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d")); Object task; @@ -559,7 +559,7 @@ public class MemTableFlushTaskTest { private boolean columnMappingBuilt; private TrackingAlignedWritableMemChunk(List<IMeasurementSchema> schemaList) { - super(schemaList, false); + super(schemaList); } @Override diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java index dffd4d941d0..5a0f8a1e800 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java @@ -183,11 +183,14 @@ public class PrimitiveMemTableTest { new StringArrayDeviceID("root.test.d1"), measurementSchemas, i, new Object[] {i, i, i}); } memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 150, 160), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 150, 160), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 150, 160), + true); ResourceByPathUtils resourcesByPathUtils = ResourceByPathUtils.getResourceInstance( new AlignedFullPath( @@ -230,11 +233,14 @@ public class PrimitiveMemTableTest { new StringArrayDeviceID("root.test.d1"), measurementSchemas, i, new Object[] {i, i, i}); } memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 150, 160), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 150, 160), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 150, 160)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 150, 160), + true); ResourceByPathUtils resourcesByPathUtils = ResourceByPathUtils.getResourceInstance( new AlignedFullPath( @@ -290,11 +296,14 @@ public class PrimitiveMemTableTest { memTable.writeAlignedRow(deviceID, measurementSchemas, i, new Object[] {i, i, i}); } memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 1, 10)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s1", TSDataType.INT32), 1, 10), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 1, 10)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s2", TSDataType.INT32), 1, 10), + true); memTable.delete( - new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 1, 10)); + new TreeDeletionEntry(new MeasurementPath("root.test.d1.s3", TSDataType.INT32), 1, 10), + true); Assert.assertFalse(originalWorkingList.isSorted()); AlignedReadOnlyMemChunk flushingQueryMemChunk = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java index a1ba4cee074..d22e41c35f5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkRegionScanTest.java @@ -266,7 +266,8 @@ public class WritableMemChunkRegionScanTest { memTable.delete( new TreeDeletionEntry( new MeasurementPath(new StringArrayDeviceID("root.test.d1"), "s1"), - new TimeRange(1, 1500))); + new TimeRange(1, 1500)), + true); writableMemChunk = (AlignedWritableMemChunk) memTable.getWritableMemChunk(new StringArrayDeviceID("root.test.d1"), ""); @@ -296,7 +297,7 @@ public class WritableMemChunkRegionScanTest { new MeasurementSchema("s2", TSDataType.INT32), new MeasurementSchema("s3", TSDataType.INT32)); AlignedWritableMemChunk writableMemChunk = - new AlignedWritableMemChunk(measurementSchemas, true); + new AlignedWritableMemChunk(measurementSchemas); int size = 100000; for (int i = 0; i < size; i++) { if (i <= 10000) { @@ -429,7 +430,8 @@ public class WritableMemChunkRegionScanTest { memTable.delete( new TreeDeletionEntry( new MeasurementPath(new StringArrayDeviceID("root.test.d1"), "s1"), - new TimeRange(1, 1500))); + new TimeRange(1, 1500)), + true); WritableMemChunk writableMemChunk = (WritableMemChunk) memTable.getWritableMemChunk(new StringArrayDeviceID("root.test.d1"), "s1"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java index aa71595d443..bcc5ee2df45 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java @@ -304,6 +304,49 @@ public class ImportWALTest { assertTrue(databases.values().stream().allMatch(database -> database == null)); } + /** + * Every directory must declare a database before any replay starts, otherwise the data model of a + * WAL snapshot could only be guessed. + */ + @Test + public void testUnresolvedDirectoryRequiresExplicitDatabase() { + final List<Path> files = + Arrays.asList( + Paths.get("factory-3/one.wal"), + Paths.get("0/two.wal"), + Paths.get("root.sg-4/three.wal")); + ImportWAL.requireResolvedDatabases( + ImportWAL.resolveDirectoryDatabases(files, "target", true, null)); + + final Map<Path, String> unresolved = + ImportWAL.resolveDirectoryDatabases(files, null, true, null); + final IllegalArgumentException failure = + assertThrows( + IllegalArgumentException.class, + () -> ImportWAL.requireResolvedDatabases(unresolved)); + // The inferable directory is not reported, the others are listed in directory order. + assertTrue(failure.getMessage().contains("0, root.sg-4")); + assertTrue(failure.getMessage().contains("-db/--database")); + } + + /** A directory whose database cannot be inferred must fail before any Session is opened. */ + @Test + public void testUnresolvedDatabaseFailsBeforeReplay() throws Exception { + final Path directory = temporaryFolder.newFolder("0").toPath(); + createWALFile(directory, 0); + final ByteArrayOutputStream error = new ByteArrayOutputStream(); + + assertEquals( + 1, + ImportWAL.run( + new String[] {"-f", directory.toString(), "-pw", "password"}, + new PrintStream(error), + new PrintStream(error))); + + assertTrue(error.toString().contains(directory.toString())); + assertTrue(error.toString().contains("-db/--database")); + } + /** * One thread may visit several databases; each directory needs its own Session and schema cache. */ @@ -1547,7 +1590,8 @@ public class ImportWALTest { new Object[] {10, null, DateUtils.parseDateExpressionToInt(firstDate)}); final Session treeSession = mock(Session.class); - new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + new ImportWAL.WALReplayer(treeSession, null, "root.sg") + .replay(new WALInfoEntry(1, memTable)); final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); verify(treeSession, times(3)).insertTablet(tabletCaptor.capture()); @@ -1593,7 +1637,8 @@ public class ImportWALTest { new Object[] {10, 100L, DateUtils.parseDateExpressionToInt(firstDate)}); final Session treeSession = mock(Session.class); - new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + new ImportWAL.WALReplayer(treeSession, null, "root.sg") + .replay(new WALInfoEntry(1, memTable)); final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); verify(treeSession).insertAlignedTablet(tabletCaptor.capture()); @@ -1623,7 +1668,7 @@ public class ImportWALTest { final ImportWAL.ReplayStatistics statistics = ImportWAL.replayWALFiles( Collections.singletonList(walFile.toPath()), - new ImportWAL.WALReplayer(treeSession, null, null)); + new ImportWAL.WALReplayer(treeSession, null, "root.sg")); assertEquals(1, statistics.getReplayedOperationCount()); final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); @@ -1644,7 +1689,8 @@ public class ImportWALTest { } final Session treeSession = mock(Session.class); - new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + new ImportWAL.WALReplayer(treeSession, null, "root.sg") + .replay(new WALInfoEntry(1, memTable)); final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); verify(treeSession, times(2)).insertTablet(tabletCaptor.capture());
