You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by sh...@apache.org on 2026/03/13 10:29:26 UTC
(iotdb) branch fetchSchemaWhenDataTypeNotMatchedInCompaction created (now ccd4b388293)
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a change to branch fetchSchemaWhenDataTypeNotMatchedInCompaction
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at ccd4b388293 Fix incorrect measurement schema during compaction
This branch includes the following new commits:
new ccd4b388293 Fix incorrect measurement schema during compaction
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Fix incorrect measurement schema during compaction
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch fetchSchemaWhenDataTypeNotMatchedInCompaction
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ccd4b388293b2eed376f111eb550affaff52acac
Author: shuwenwei <s1...@gmail.com>
AuthorDate: Fri Mar 13 18:29:04 2026 +0800
Fix incorrect measurement schema during compaction
---
.../performer/impl/FastCompactionPerformer.java | 15 +-
.../impl/ReadChunkCompactionPerformer.java | 64 +++-
.../execute/utils/CompactionSeriesContext.java | 13 +
.../compaction/execute/utils/CompactionUtils.java | 46 +++
.../execute/utils/MultiTsFileDeviceIterator.java | 216 ++++++++++--
...edReadChunkAlignedSeriesCompactionExecutor.java | 10 +-
.../FastNonAlignedSeriesCompactionExecutor.java | 11 +-
.../ReadChunkAlignedSeriesCompactionExecutor.java | 121 ++++++-
.../plan/analyze/FakeSchemaFetcherImpl.java | 4 +-
.../compaction/AbstractCompactionTest.java | 16 +
.../compaction/CompactionDataTypeAlterTest.java | 377 ---------------------
.../AbstractCompactionAlterDataTypeTest.java | 181 ++++++++++
.../CompactionDataTypeAlterTableTest.java | 241 +++++++++++++
.../alterDataType/CompactionDataTypeAlterTest.java | 230 +++++++++++++
...ctionDataTypeNotMatchAlterableDataTypeTest.java | 53 +--
.../CompactionDataTypeNotMatchTest.java | 105 ++----
.../TableModelCompactionWithTTLTest.java | 25 +-
.../utils/CompactionFakeSchemaFetcherImpl.java | 37 ++
.../apache/iotdb/commons/schema/table/TsTable.java | 3 +-
19 files changed, 1169 insertions(+), 599 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java
index 629493f184b..b2f0e8c1392 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java
@@ -52,7 +52,6 @@ import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.encrypt.EncryptParameter;
-import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.StopReadTsFileByInterruptException;
import org.apache.tsfile.exception.write.PageException;
import org.apache.tsfile.file.metadata.IDeviceID;
@@ -281,21 +280,9 @@ public class FastCompactionPerformer
// timeseries metadata, in order to facilitate the reading of chunkMetadata directly by this
// offset later. Here we don't need to deserialize chunk metadata, we can deserialize them and
// get their schema later.
- Map<String, Map<TsFileResource, Pair<Long, Long>>> timeseriesMetadataOffsetMap =
- new LinkedHashMap<>();
-
- Map<String, TSDataType> measurementDataTypeMap = new LinkedHashMap<>();
-
Map<String, CompactionSeriesContext> compactionSeriesContextMap =
deviceIterator.getCompactionSeriesContextOfCurrentDevice();
-
- for (Map.Entry<String, CompactionSeriesContext> entry : compactionSeriesContextMap.entrySet()) {
- timeseriesMetadataOffsetMap.put(
- entry.getKey(), entry.getValue().getFileTimeseriesMetdataOffsetMap());
- measurementDataTypeMap.put(entry.getKey(), entry.getValue().getFinalType());
- }
-
- List<String> allMeasurements = new ArrayList<>(timeseriesMetadataOffsetMap.keySet());
+ List<String> allMeasurements = new ArrayList<>(compactionSeriesContextMap.keySet());
allMeasurements.sort((String::compareTo));
int subTaskNums = Math.min(allMeasurements.size(), SUB_TASK_NUM);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java
index d406286e37f..64a940ae960 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ISeqCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchemaCollector;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.MultiTsFileDeviceIterator;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.BatchedReadChunkAlignedSeriesCompactionExecutor;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.SingleSeriesCompactionExecutor;
@@ -48,10 +49,12 @@ import org.apache.tsfile.file.metadata.ChunkMetadata;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.Schema;
import java.io.IOException;
import java.util.Collections;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
@@ -147,7 +150,11 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
if (aligned) {
compactAlignedSeries(
- device, targetResources.get(currentTargetFileIndex), currentWriter, deviceIterator);
+ deviceIterator.getDatabaseName(),
+ device,
+ targetResources.get(currentTargetFileIndex),
+ currentWriter,
+ deviceIterator);
} else {
compactNotAlignedSeries(
device, targetResources.get(currentTargetFileIndex), currentWriter, deviceIterator);
@@ -224,6 +231,7 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
}
private void compactAlignedSeries(
+ String database,
IDeviceID device,
TsFileResource targetResource,
CompactionTsFileWriter writer,
@@ -239,6 +247,7 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
writer.startChunkGroup(device);
BatchedReadChunkAlignedSeriesCompactionExecutor compactionExecutor =
new BatchedReadChunkAlignedSeriesCompactionExecutor(
+ database,
device,
targetResource,
readerAndChunkMetadataList,
@@ -293,7 +302,7 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
seriesIterator.getMetadataListForCurrentSeries();
// remove the chunk metadata whose data type not match the data type of last chunk
readerAndChunkMetadataList =
- filterDataTypeNotMatchedChunkMetadata(readerAndChunkMetadataList);
+ filterDataTypeNotMatchedChunkMetadata(device, measurement, readerAndChunkMetadataList);
SingleSeriesCompactionExecutor compactionExecutorOfCurrentTimeSeries =
new SingleSeriesCompactionExecutor(
device, measurement, readerAndChunkMetadataList, writer, targetResource, summary);
@@ -304,42 +313,63 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
private LinkedList<Pair<TsFileSequenceReader, List<ChunkMetadata>>>
filterDataTypeNotMatchedChunkMetadata(
+ IDeviceID deviceID,
+ String measurement,
LinkedList<Pair<TsFileSequenceReader, List<ChunkMetadata>>> readerAndChunkMetadataList) {
if (readerAndChunkMetadataList.isEmpty()) {
return readerAndChunkMetadataList;
}
- LinkedList<Pair<TsFileSequenceReader, List<ChunkMetadata>>> result = new LinkedList<>();
// find correct data type
TSDataType correctDataType = null;
- for (int i = readerAndChunkMetadataList.size() - 1; i >= 0 && correctDataType == null; i--) {
- List<ChunkMetadata> chunkMetadataList = readerAndChunkMetadataList.get(i).getRight();
- if (chunkMetadataList == null || chunkMetadataList.isEmpty()) {
- continue;
- }
+ boolean hasDifferentDataTypes = false;
+ Iterator<Pair<TsFileSequenceReader, List<ChunkMetadata>>> descIterator =
+ readerAndChunkMetadataList.descendingIterator();
+ while (descIterator.hasNext()) {
+ Pair<TsFileSequenceReader, List<ChunkMetadata>> pair = descIterator.next();
+ List<ChunkMetadata> chunkMetadataList = pair.right;
+ TSDataType dataTypeInCurrentFile = null;
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
- if (chunkMetadata == null) {
- continue;
+ if (chunkMetadata != null) {
+ dataTypeInCurrentFile = chunkMetadata.getDataType();
+ break;
}
- correctDataType = chunkMetadata.getDataType();
+ }
+ if (correctDataType == null) {
+ correctDataType = dataTypeInCurrentFile;
+ } else if (correctDataType != dataTypeInCurrentFile) {
+ hasDifferentDataTypes = true;
break;
}
}
- if (correctDataType == null) {
+ if (!hasDifferentDataTypes) {
return readerAndChunkMetadataList;
}
+
+ IMeasurementSchema schema =
+ CompactionUtils.getLatestMeasurementSchemasForTreeModel(
+ deviceID, Collections.singletonList(measurement))
+ .get(0);
+ if (schema != null) {
+ correctDataType = schema.getType();
+ }
+
+ LinkedList<Pair<TsFileSequenceReader, List<ChunkMetadata>>> result = new LinkedList<>();
// check data type consistent and skip compact files with wrong data type
for (Pair<TsFileSequenceReader, List<ChunkMetadata>> tsFileSequenceReaderListPair :
readerAndChunkMetadataList) {
boolean dataTypeConsistent = true;
for (ChunkMetadata chunkMetadata : tsFileSequenceReaderListPair.getRight()) {
- if (chunkMetadata != null
- && !MetadataUtils.canAlter(chunkMetadata.getDataType(), correctDataType)) {
- dataTypeConsistent = false;
+ if (chunkMetadata == null) {
+ continue;
+ }
+ if (chunkMetadata.getDataType() == correctDataType) {
break;
}
- if (chunkMetadata != null && chunkMetadata.getDataType() != correctDataType) {
- chunkMetadata.setNewType(correctDataType);
+ if (!MetadataUtils.canAlter(chunkMetadata.getDataType(), correctDataType)) {
+ dataTypeConsistent = false;
+ break;
}
+ chunkMetadata.setNewType(correctDataType);
}
if (!dataTypeConsistent) {
continue;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionSeriesContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionSeriesContext.java
index 8f6d017f235..fc208adfb26 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionSeriesContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionSeriesContext.java
@@ -29,6 +29,7 @@ import java.util.Map;
public class CompactionSeriesContext {
Map<TsFileResource, Pair<Long, Long>> fileTimeseriesMetdataOffsetMap;
TSDataType finalType;
+ boolean needUpdateDataType = false;
public CompactionSeriesContext() {
fileTimeseriesMetdataOffsetMap = new HashMap<>();
@@ -57,4 +58,16 @@ public class CompactionSeriesContext {
this.finalType = finalType;
}
}
+
+ public void setFinalType(TSDataType finalType) {
+ this.finalType = finalType;
+ }
+
+ public boolean isNeedUpdateDataType() {
+ return needUpdateDataType;
+ }
+
+ public void setNeedUpdateDataType(boolean needUpdateDataType) {
+ this.needUpdateDataType = needUpdateDataType;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
index 61943047e54..ac46cbb70bd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
@@ -24,12 +24,20 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.path.PatternTreeMap;
+import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.common.schematree.DeviceSchemaInfo;
+import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
import org.apache.iotdb.db.service.metrics.FileMetrics;
@@ -56,6 +64,7 @@ import org.apache.tsfile.file.metadata.ChunkMetadata;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.fileSystem.FSFactoryProducer;
import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -85,8 +94,15 @@ public class CompactionUtils {
LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
private static final String SYSTEM = "system";
+ private static ISchemaFetcher schemaFetcherForTest = null;
+
private CompactionUtils() {}
+ @TestOnly
+ public static void setSchemaFetcher(ISchemaFetcher schemaFetcher) {
+ CompactionUtils.schemaFetcherForTest = schemaFetcher;
+ }
+
/**
* Update the targetResource. Move tmp target file to target file and serialize
* xxx.tsfile.resource.
@@ -640,4 +656,34 @@ public class CompactionUtils {
}
rateLimiter.acquire((int) size);
}
+
+ public static List<IMeasurementSchema> getLatestMeasurementSchemasForTreeModel(
+ IDeviceID deviceID, List<String> measurements) {
+ if (measurements.isEmpty()) {
+ return Collections.emptyList();
+ }
+ ISchemaFetcher schemaFetcher =
+ schemaFetcherForTest == null ? ClusterSchemaFetcher.getInstance() : schemaFetcherForTest;
+ PartialPath devicePath;
+ PathPatternTree patternTree = new PathPatternTree();
+ try {
+ devicePath = CompactionPathUtils.getPath(deviceID);
+ for (String measurement : measurements) {
+ patternTree.appendFullPath(devicePath, measurement);
+ }
+ } catch (IllegalPathException e) {
+ throw new RuntimeException(e);
+ }
+ ISchemaTree schemaTree =
+ schemaFetcher.fetchRawSchemaInMeasurementLevel(
+ patternTree,
+ SchemaConstant.ALL_MATCH_SCOPE,
+ new MPPQueryContext(new QueryId("compaction")),
+ true);
+ DeviceSchemaInfo deviceSchemaInfo = schemaTree.searchDeviceSchemaInfo(devicePath, measurements);
+ if (deviceSchemaInfo == null) {
+ return Collections.nCopies(measurements.size(), null);
+ }
+ return deviceSchemaInfo.getMeasurementSchemaList();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/MultiTsFileDeviceIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/MultiTsFileDeviceIterator.java
index 31296771090..f2c0a49575d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/MultiTsFileDeviceIterator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/MultiTsFileDeviceIterator.java
@@ -24,9 +24,11 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PatternTreeMap;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
-import org.apache.iotdb.commons.utils.MetadataUtils;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileReader;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
@@ -47,14 +49,17 @@ import org.apache.tsfile.file.metadata.TimeseriesMetadata;
import org.apache.tsfile.read.TsFileDeviceIterator;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -245,6 +250,10 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
return currentDevice;
}
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
public long getTTLForCurrentDevice() {
return ttlForCurrentDevice;
}
@@ -253,6 +262,12 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
return timeLowerBoundForCurrentDevice;
}
+ public Map<String, MeasurementSchema> getAllSchemasOfCurrentDevice() throws IOException {
+ return ignoreAllNullRows
+ ? getAllSchemasOfCurrentDeviceForTree()
+ : getAllSchemasOfCurrentDeviceForTable();
+ }
+
/**
* Get all measurements and schemas of the current device from source files. Traverse all the
* files from the newest to the oldest in turn and start traversing the index tree from the
@@ -260,8 +275,9 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
*
* @throws IOException if io errors occurred
*/
- public Map<String, MeasurementSchema> getAllSchemasOfCurrentDevice() throws IOException {
+ public Map<String, MeasurementSchema> getAllSchemasOfCurrentDeviceForTree() throws IOException {
Map<String, MeasurementSchema> schemaMap = new ConcurrentHashMap<>();
+ Set<String> seriesNeedToUpdateDataType = new HashSet<>();
// get schemas from the newest file to the oldest file
for (TsFileResource resource : tsFileResourcesSortedByDesc) {
if (!deviceIteratorMap.containsKey(resource)
@@ -275,16 +291,75 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
reader.getDeviceTimeseriesMetadata(
timeseriesMetadataList,
deviceIteratorMap.get(resource).getFirstMeasurementNodeOfCurrentDevice(),
+ seriesNeedToUpdateDataType,
+ true,
+ null);
+ for (TimeseriesMetadata timeseriesMetadata : timeseriesMetadataList) {
+ MeasurementSchema measurementSchema = schemaMap.get(timeseriesMetadata.getMeasurementId());
+ if (measurementSchema == null) {
+ if (!timeseriesMetadata.getChunkMetadataList().isEmpty()) {
+ schemaMap.put(
+ timeseriesMetadata.getMeasurementId(),
+ reader.getMeasurementSchema(timeseriesMetadata.getChunkMetadataList()));
+ }
+ continue;
+ }
+ if (measurementSchema.getType() != timeseriesMetadata.getTsDataType()) {
+ seriesNeedToUpdateDataType.add(timeseriesMetadata.getMeasurementId());
+ }
+ }
+ }
+ List<IMeasurementSchema> latestMeasurementSchemas =
+ CompactionUtils.getLatestMeasurementSchemasForTreeModel(
+ currentDevice.left, new ArrayList<>(seriesNeedToUpdateDataType));
+ for (IMeasurementSchema latestMeasurementSchema : latestMeasurementSchemas) {
+ if (latestMeasurementSchema != null) {
+ schemaMap.put(
+ latestMeasurementSchema.getMeasurementName(),
+ (MeasurementSchema) latestMeasurementSchema);
+ }
+ }
+ return schemaMap;
+ }
+
+ private Map<String, MeasurementSchema> getAllSchemasOfCurrentDeviceForTable() throws IOException {
+ Map<String, MeasurementSchema> schemaMap = new ConcurrentHashMap<>();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance().getTable(databaseName, currentDevice.left.getTableName());
+ // get schemas from the newest file to the oldest file
+ for (TsFileResource resource : tsFileResourcesSortedByDesc) {
+ TsFileDeviceIterator deviceIterator = deviceIteratorMap.get(resource);
+ if (!deviceIteratorMap.containsKey(resource)
+ || !deviceIterator.current().equals(currentDevice)) {
+ // if this tsfile has no more device or next device is not equals to the current device,
+ // which means this tsfile does not contain the current device, then skip it.
+ continue;
+ }
+ TsFileSequenceReader reader = readerMap.get(resource);
+ List<TimeseriesMetadata> timeseriesMetadataList = new ArrayList<>();
+ reader.getDeviceTimeseriesMetadata(
+ timeseriesMetadataList,
+ deviceIterator.getFirstMeasurementNodeOfCurrentDevice(),
schemaMap.keySet(),
true,
null);
for (TimeseriesMetadata timeseriesMetadata : timeseriesMetadataList) {
- if (!schemaMap.containsKey(timeseriesMetadata.getMeasurementId())
- && !timeseriesMetadata.getChunkMetadataList().isEmpty()) {
- schemaMap.put(
- timeseriesMetadata.getMeasurementId(),
- reader.getMeasurementSchema(timeseriesMetadata.getChunkMetadataList()));
+ MeasurementSchema measurementSchema = schemaMap.get(timeseriesMetadata.getMeasurementId());
+ if (measurementSchema != null) {
+ continue;
+ }
+ if (tsTable != null) {
+ TsTableColumnSchema columnSchema =
+ tsTable.getColumnSchema(timeseriesMetadata.getMeasurementId());
+ if (columnSchema != null) {
+ measurementSchema = (MeasurementSchema) columnSchema.getMeasurementSchema();
+ }
+ }
+ if (measurementSchema == null && !timeseriesMetadata.getChunkMetadataList().isEmpty()) {
+ measurementSchema =
+ reader.getMeasurementSchema(timeseriesMetadata.getChunkMetadataList());
}
+ schemaMap.put(timeseriesMetadata.getMeasurementId(), measurementSchema);
}
}
return schemaMap;
@@ -300,6 +375,7 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
public Map<String, CompactionSeriesContext> getCompactionSeriesContextOfCurrentDevice()
throws IOException {
Map<String, CompactionSeriesContext> compactionSeriesContextMap = new HashMap<>();
+ List<String> seriesNeedToUpdateDataType = new ArrayList<>();
for (TsFileResource resource : tsFileResourcesSortedByDesc) {
if (!deviceIteratorMap.containsKey(resource)
|| !deviceIteratorMap.get(resource).current().equals(currentDevice)) {
@@ -318,23 +394,37 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
String measurementId = entrySet.getKey();
TimeseriesMetadata timeseriesMetadata = entrySet.getValue().left;
Pair<Long, Long> offset = entrySet.getValue().right;
- TSDataType dataType = timeseriesMetadata.getTsDataType();
+ TSDataType dataTypeOfCurrentFile = timeseriesMetadata.getTsDataType();
CompactionSeriesContext compactionSeriesContext =
compactionSeriesContextMap.get(measurementId);
if (compactionSeriesContext != null
- && compactionSeriesContext.getFinalType() != null
- && !MetadataUtils.canAlter(dataType, compactionSeriesContext.getFinalType())) {
- continue;
+ && !compactionSeriesContext.isNeedUpdateDataType()
+ && compactionSeriesContext.getFinalType() != dataTypeOfCurrentFile) {
+ compactionSeriesContext.setNeedUpdateDataType(true);
+ seriesNeedToUpdateDataType.add(measurementId);
}
compactionSeriesContext =
compactionSeriesContextMap.computeIfAbsent(
measurementId, k -> new CompactionSeriesContext());
compactionSeriesContext.put(resource, offset);
- compactionSeriesContext.setFinalTypeIfAbsent(dataType);
+ compactionSeriesContext.setFinalTypeIfAbsent(dataTypeOfCurrentFile);
+ }
+ }
+
+ List<IMeasurementSchema> measurementSchema =
+ CompactionUtils.getLatestMeasurementSchemasForTreeModel(
+ currentDevice.left, seriesNeedToUpdateDataType);
+ for (IMeasurementSchema iMeasurementSchema : measurementSchema) {
+ if (iMeasurementSchema == null) {
+ continue;
}
+ compactionSeriesContextMap
+ .get(iMeasurementSchema.getMeasurementName())
+ .setFinalType(iMeasurementSchema.getType());
}
+
return compactionSeriesContextMap;
}
@@ -347,14 +437,80 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
* endOffset>
* @throws IOException if io errors occurred
*/
- @SuppressWarnings({"checkstyle:AtclauseOrderCheck", "squid:S3824"})
public Map<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>>
getTimeseriesSchemaAndMetadataOffsetOfCurrentDevice() throws IOException {
+ return ignoreAllNullRows
+ ? getTimeseriesSchemaAndMetadataOffsetOfCurrentDeviceForTree()
+ : getTimeseriesSchemaAndMetadataOffsetOfCurrentDeviceForTable();
+ }
+
+ @SuppressWarnings({"checkstyle:AtclauseOrderCheck", "squid:S3824"})
+ public Map<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>>
+ getTimeseriesSchemaAndMetadataOffsetOfCurrentDeviceForTree() throws IOException {
Map<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>>
timeseriesMetadataOffsetMap = new LinkedHashMap<>();
+ Set<MeasurementSchema> seriesNeedToUpdateDataType = new LinkedHashSet<>();
for (TsFileResource resource : tsFileResourcesSortedByDesc) {
- if (!deviceIteratorMap.containsKey(resource)
- || !deviceIteratorMap.get(resource).current().equals(currentDevice)) {
+ TsFileDeviceIterator iterator = deviceIteratorMap.get(resource);
+ if (iterator == null || !iterator.current().equals(currentDevice)) {
+ // if this tsfile has no more device or next device is not equals to the current device,
+ // which means this tsfile does not contain the current device, then skip it.
+ continue;
+ }
+
+ CompactionTsFileReader reader = (CompactionTsFileReader) readerMap.get(resource);
+
+ for (Map.Entry<String, Pair<TimeseriesMetadata, Pair<Long, Long>>> entrySet :
+ reader
+ .getTimeseriesMetadataAndOffsetByDevice(
+ deviceIteratorMap.get(resource).getFirstMeasurementNodeOfCurrentDevice(),
+ timeseriesMetadataOffsetMap.keySet(),
+ true)
+ .entrySet()) {
+ String measurementId = entrySet.getKey();
+ Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>> existedPair =
+ timeseriesMetadataOffsetMap.get(measurementId);
+ if (existedPair == null) {
+ MeasurementSchema schema =
+ reader.getMeasurementSchema(entrySet.getValue().left.getChunkMetadataList());
+ existedPair = new Pair<>(schema, new HashMap<>());
+ timeseriesMetadataOffsetMap.put(measurementId, existedPair);
+ } else if (!seriesNeedToUpdateDataType.contains(existedPair.getLeft())
+ && existedPair.left.getType() != entrySet.getValue().getLeft().getTsDataType()) {
+ seriesNeedToUpdateDataType.add(existedPair.getLeft());
+ }
+ existedPair.right.put(resource, entrySet.getValue().right);
+ }
+ }
+ List<IMeasurementSchema> correctMeasurementSchemas =
+ CompactionUtils.getLatestMeasurementSchemasForTreeModel(
+ currentDevice.left,
+ seriesNeedToUpdateDataType.stream()
+ .map(IMeasurementSchema::getMeasurementName)
+ .collect(Collectors.toList()));
+ int i = 0;
+ for (MeasurementSchema measurementSchema : seriesNeedToUpdateDataType) {
+ IMeasurementSchema correctSchema = correctMeasurementSchemas.get(i);
+ i++;
+ if (correctSchema == null) {
+ continue;
+ }
+ measurementSchema.setDataType(correctSchema.getType());
+ measurementSchema.setEncoding(correctSchema.getEncodingType());
+ measurementSchema.setCompressionType(correctSchema.getCompressor());
+ }
+ return timeseriesMetadataOffsetMap;
+ }
+
+ public Map<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>>
+ getTimeseriesSchemaAndMetadataOffsetOfCurrentDeviceForTable() throws IOException {
+ Map<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>>
+ timeseriesMetadataOffsetMap = new LinkedHashMap<>();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance().getTable(databaseName, currentDevice.left.getTableName());
+ for (TsFileResource resource : tsFileResourcesSortedByDesc) {
+ TsFileDeviceIterator iterator = deviceIteratorMap.get(resource);
+ if (iterator == null || !iterator.current().equals(currentDevice)) {
// if this tsfile has no more device or next device is not equals to the current device,
// which means this tsfile does not contain the current device, then skip it.
continue;
@@ -363,23 +519,33 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
continue;
}
- TsFileSequenceReader reader = readerMap.get(resource);
- for (Map.Entry<String, Pair<List<IChunkMetadata>, Pair<Long, Long>>> entrySet :
+ CompactionTsFileReader reader = (CompactionTsFileReader) readerMap.get(resource);
+
+ for (Map.Entry<String, Pair<TimeseriesMetadata, Pair<Long, Long>>> entrySet :
reader
- .getTimeseriesMetadataOffsetByDevice(
+ .getTimeseriesMetadataAndOffsetByDevice(
deviceIteratorMap.get(resource).getFirstMeasurementNodeOfCurrentDevice(),
timeseriesMetadataOffsetMap.keySet(),
true)
.entrySet()) {
String measurementId = entrySet.getKey();
- if (!timeseriesMetadataOffsetMap.containsKey(measurementId)) {
- MeasurementSchema schema = reader.getMeasurementSchema(entrySet.getValue().left);
- timeseriesMetadataOffsetMap.put(measurementId, new Pair<>(schema, new HashMap<>()));
+ Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>> existedPair =
+ timeseriesMetadataOffsetMap.get(measurementId);
+ if (existedPair == null) {
+ MeasurementSchema schema = null;
+ if (tsTable != null) {
+ TsTableColumnSchema columnSchema = tsTable.getColumnSchema(measurementId);
+ if (columnSchema != null) {
+ schema = (MeasurementSchema) columnSchema.getMeasurementSchema();
+ }
+ }
+ if (schema == null) {
+ schema = reader.getMeasurementSchema(entrySet.getValue().left.getChunkMetadataList());
+ }
+ existedPair = new Pair<>(schema, new HashMap<>());
+ timeseriesMetadataOffsetMap.put(measurementId, existedPair);
}
- timeseriesMetadataOffsetMap
- .get(measurementId)
- .right
- .put(resource, entrySet.getValue().right);
+ existedPair.right.put(resource, entrySet.getValue().right);
}
}
return timeseriesMetadataOffsetMap;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedReadChunkAlignedSeriesCompactionExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedReadChunkAlignedSeriesCompactionExecutor.java
index dc030787566..108c11fe4ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedReadChunkAlignedSeriesCompactionExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/batch/BatchedReadChunkAlignedSeriesCompactionExecutor.java
@@ -72,6 +72,7 @@ public class BatchedReadChunkAlignedSeriesCompactionExecutor
originReaderAndChunkMetadataList;
public BatchedReadChunkAlignedSeriesCompactionExecutor(
+ String database,
IDeviceID device,
TsFileResource targetResource,
LinkedList<Pair<TsFileSequenceReader, List<AbstractAlignedChunkMetadata>>>
@@ -80,7 +81,14 @@ public class BatchedReadChunkAlignedSeriesCompactionExecutor
CompactionTaskSummary summary,
boolean ignoreAllNullRows)
throws IOException {
- super(device, targetResource, readerAndChunkMetadataList, writer, summary, ignoreAllNullRows);
+ super(
+ database,
+ device,
+ targetResource,
+ readerAndChunkMetadataList,
+ writer,
+ summary,
+ ignoreAllNullRows);
this.originReaderAndChunkMetadataList = readerAndChunkMetadataList;
this.batchColumnSelection =
new AlignedSeriesBatchCompactionUtils.BatchColumnSelection(schemaList, batchSize);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/FastNonAlignedSeriesCompactionExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/FastNonAlignedSeriesCompactionExecutor.java
index 363554fb606..49cbcb14582 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/FastNonAlignedSeriesCompactionExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/FastNonAlignedSeriesCompactionExecutor.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.ex
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PatternTreeMap;
+import org.apache.iotdb.commons.utils.MetadataUtils;
import org.apache.iotdb.db.exception.ChunkTypeInconsistentException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.FastCompactionTaskSummary;
@@ -165,9 +166,17 @@ public class FastNonAlignedSeriesCompactionExecutor extends SeriesCompactionExec
removeFile(fileElement);
}
}
+ boolean checked = false;
for (int i = 0; i < iChunkMetadataList.size(); i++) {
IChunkMetadata chunkMetadata = iChunkMetadataList.get(i);
- if (dataType != null && chunkMetadata.getDataType() != dataType) {
+ if (!checked) {
+ if (!MetadataUtils.canAlter(chunkMetadata.getDataType(), dataType)) {
+ removeFile(fileElement);
+ break;
+ }
+ checked = true;
+ }
+ if (chunkMetadata.getDataType() != dataType) {
chunkMetadata.setNewType(dataType);
}
// add into queue
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/readchunk/ReadChunkAlignedSeriesCompactionExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/readchunk/ReadChunkAlignedSeriesCompactionExecutor.java
index 654f8f770e7..220e45dfb36 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/readchunk/ReadChunkAlignedSeriesCompactionExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/readchunk/ReadChunkAlignedSeriesCompactionExecutor.java
@@ -19,10 +19,14 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
import org.apache.iotdb.commons.utils.MetadataUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.ModifiedStatus;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.AlignedSeriesBatchCompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.CompactionAlignedPageLazyLoadPointReader;
@@ -34,6 +38,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFi
import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.encoding.decoder.Decoder;
import org.apache.tsfile.encrypt.EncryptUtils;
@@ -45,6 +50,7 @@ import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata;
import org.apache.tsfile.file.metadata.ChunkMetadata;
import org.apache.tsfile.file.metadata.IChunkMetadata;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.TsFileSequenceReader;
@@ -86,6 +92,7 @@ public class ReadChunkAlignedSeriesCompactionExecutor {
private boolean lastWriteTimestampSet = false;
public ReadChunkAlignedSeriesCompactionExecutor(
+ String database,
IDeviceID device,
TsFileResource targetResource,
LinkedList<Pair<TsFileSequenceReader, List<AbstractAlignedChunkMetadata>>>
@@ -99,13 +106,17 @@ public class ReadChunkAlignedSeriesCompactionExecutor {
this.writer = writer;
this.targetResource = targetResource;
this.summary = summary;
- collectValueColumnSchemaList();
+ this.ignoreAllNullRows = ignoreAllNullRows;
+ if (device.isTableModel()) {
+ collectValueColumnSchemaListForTable(database);
+ } else {
+ collectValueColumnSchemaListForTree();
+ }
fillAlignedChunkMetadataToMatchSchemaList();
int compactionFileLevel =
Integer.parseInt(this.targetResource.getTsFile().getName().split("-")[2]);
flushController = new ReadChunkAlignedSeriesCompactionFlushController(compactionFileLevel);
this.chunkWriter = constructAlignedChunkWriter();
- this.ignoreAllNullRows = ignoreAllNullRows;
}
// used for batched column compaction
@@ -133,8 +144,66 @@ public class ReadChunkAlignedSeriesCompactionExecutor {
this.ignoreAllNullRows = ignoreAllNullRows;
}
- private void collectValueColumnSchemaList() throws IOException {
+ private void collectValueColumnSchemaListForTable(String database) throws IOException {
+ TSFileConfig tsFileConfig = TSFileDescriptor.getInstance().getConfig();
+ TsTable tsTable = DataNodeTableCache.getInstance().getTable(database, device.getTableName());
Map<String, IMeasurementSchema> measurementSchemaMap = new HashMap<>();
+ for (int i = this.readerAndChunkMetadataList.size() - 1; i >= 0; i--) {
+ Pair<TsFileSequenceReader, List<AbstractAlignedChunkMetadata>> pair =
+ this.readerAndChunkMetadataList.get(i);
+ CompactionTsFileReader reader = (CompactionTsFileReader) pair.getLeft();
+ List<AbstractAlignedChunkMetadata> alignedChunkMetadataList = pair.getRight();
+ for (AbstractAlignedChunkMetadata alignedChunkMetadata : alignedChunkMetadataList) {
+ if (alignedChunkMetadata == null) {
+ continue;
+ }
+ if (timeSchema == null) {
+ timeSchema =
+ new MeasurementSchema(
+ "",
+ alignedChunkMetadata.getTimeChunkMetadata().getDataType(),
+ TSEncoding.valueOf(tsFileConfig.getTimeEncoder()),
+ tsFileConfig.getCompressor());
+ }
+ for (IChunkMetadata chunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
+ if (chunkMetadata == null
+ || measurementSchemaMap.containsKey(chunkMetadata.getMeasurementUid())) {
+ continue;
+ }
+ TsTableColumnSchema schemaInTsTable =
+ tsTable.getColumnSchema(chunkMetadata.getMeasurementUid());
+ IMeasurementSchema measurementSchema;
+ if (schemaInTsTable == null) {
+ ChunkHeader chunkHeader =
+ reader.readChunkHeader(chunkMetadata.getOffsetOfChunkHeader());
+ measurementSchema =
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType());
+ } else {
+ measurementSchema =
+ new MeasurementSchema(
+ chunkMetadata.getMeasurementUid(),
+ schemaInTsTable.getDataType(),
+ tsFileConfig.getValueEncoder(schemaInTsTable.getDataType()),
+ tsFileConfig.getCompressor(schemaInTsTable.getDataType()));
+ }
+ measurementSchemaMap.put(chunkMetadata.getMeasurementUid(), measurementSchema);
+ }
+ }
+ }
+ this.schemaList =
+ measurementSchemaMap.values().stream()
+ .sorted(Comparator.comparing(IMeasurementSchema::getMeasurementName))
+ .collect(Collectors.toList());
+ }
+
+ private void collectValueColumnSchemaListForTree() throws IOException {
+ Map<String, Pair<IMeasurementSchema, Boolean>> measurementSchemaMap = new HashMap<>();
+ List<Pair<IMeasurementSchema, Boolean>> measurementSchemasNeedToUpdate = new ArrayList<>();
+ List<String> measurementNamesNeedToUpdate = new ArrayList<>();
for (int i = this.readerAndChunkMetadataList.size() - 1; i >= 0; i--) {
Pair<TsFileSequenceReader, List<AbstractAlignedChunkMetadata>> pair =
this.readerAndChunkMetadataList.get(i);
@@ -158,24 +227,50 @@ public class ReadChunkAlignedSeriesCompactionExecutor {
}
for (IChunkMetadata chunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
- if (chunkMetadata == null
- || measurementSchemaMap.containsKey(chunkMetadata.getMeasurementUid())) {
+ if (chunkMetadata == null) {
continue;
}
- ChunkHeader chunkHeader = reader.readChunkHeader(chunkMetadata.getOffsetOfChunkHeader());
- IMeasurementSchema schema =
- new MeasurementSchema(
- chunkHeader.getMeasurementID(),
- chunkHeader.getDataType(),
- chunkHeader.getEncodingType(),
- chunkHeader.getCompressionType());
- measurementSchemaMap.put(chunkMetadata.getMeasurementUid(), schema);
+ Pair<IMeasurementSchema, Boolean> measurementSchema =
+ measurementSchemaMap.get(chunkMetadata.getMeasurementUid());
+ if (measurementSchema == null) {
+ ChunkHeader chunkHeader =
+ reader.readChunkHeader(chunkMetadata.getOffsetOfChunkHeader());
+ IMeasurementSchema schema =
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType());
+ measurementSchemaMap.put(chunkMetadata.getMeasurementUid(), new Pair<>(schema, false));
+ continue;
+ }
+ if (measurementSchema.getLeft().getType() != chunkMetadata.getDataType()
+ && !measurementSchema.getRight()) {
+ measurementSchema.setRight(true);
+ measurementSchemasNeedToUpdate.add(measurementSchema);
+ measurementNamesNeedToUpdate.add(chunkMetadata.getMeasurementUid());
+ }
+ }
+ }
+ }
+
+ if (ignoreAllNullRows) {
+ List<IMeasurementSchema> latestMeasurementSchemas =
+ CompactionUtils.getLatestMeasurementSchemasForTreeModel(
+ device, measurementNamesNeedToUpdate);
+ for (int i = 0; i < measurementSchemasNeedToUpdate.size(); i++) {
+ IMeasurementSchema latestMeasurementSchema = latestMeasurementSchemas.get(i);
+ if (latestMeasurementSchema == null) {
+ continue;
}
+ Pair<IMeasurementSchema, Boolean> pair = measurementSchemasNeedToUpdate.get(i);
+ pair.setLeft(latestMeasurementSchema);
}
}
this.schemaList =
measurementSchemaMap.values().stream()
+ .map(Pair::getLeft)
.sorted(Comparator.comparing(IMeasurementSchema::getMeasurementName))
.collect(Collectors.toList());
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakeSchemaFetcherImpl.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakeSchemaFetcherImpl.java
index 0aab3518945..e4d54a3f19e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakeSchemaFetcherImpl.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakeSchemaFetcherImpl.java
@@ -44,7 +44,7 @@ import java.util.Map;
public class FakeSchemaFetcherImpl implements ISchemaFetcher {
- private final ClusterSchemaTree schemaTree = new ClusterSchemaTree(generateSchemaTree());
+ protected final ClusterSchemaTree schemaTree = new ClusterSchemaTree(generateSchemaTree());
@Override
public ClusterSchemaTree fetchSchema(
@@ -106,7 +106,7 @@ public class FakeSchemaFetcherImpl implements ISchemaFetcher {
*
* @return the root node of the generated schemaTree
*/
- private SchemaNode generateSchemaTree() {
+ protected SchemaNode generateSchemaTree() {
SchemaNode root = new SchemaInternalNode("root");
SchemaNode sg = new SchemaInternalNode("sg");
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
index a5503bb9e64..25d9419abdb 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
@@ -33,6 +33,12 @@ import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContex
import org.apache.iotdb.db.storageengine.buffer.BloomFilterCache;
import org.apache.iotdb.db.storageengine.buffer.ChunkCache;
import org.apache.iotdb.db.storageengine.buffer.TimeSeriesMetadataCache;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerSeqCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerUnseqCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.MultiTsFileDeviceIterator;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.IDataBlockReader;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.SeriesDataBlockReader;
@@ -871,4 +877,14 @@ public class AbstractCompactionTest {
}
return new ArrayList<>(paths);
}
+
+ protected ICompactionPerformer getPerformer(String performerType) {
+ if (performerType.equalsIgnoreCase(InnerSeqCompactionPerformer.READ_CHUNK.toString())) {
+ return new ReadChunkCompactionPerformer();
+ } else if (performerType.equalsIgnoreCase(InnerUnseqCompactionPerformer.FAST.toString())) {
+ return new FastCompactionPerformer(false);
+ } else {
+ return new ReadPointCompactionPerformer();
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java
deleted file mode 100644
index 23d509f3dfe..00000000000
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java
+++ /dev/null
@@ -1,377 +0,0 @@
-/*
- * 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.storageengine.dataregion.compaction;
-
-import org.apache.iotdb.commons.exception.IllegalPathException;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.commons.path.MeasurementPath;
-import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
-import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
-import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
-import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
-
-import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.exception.write.WriteProcessException;
-import org.apache.tsfile.file.metadata.IDeviceID;
-import org.apache.tsfile.read.TsFileReader;
-import org.apache.tsfile.read.TsFileSequenceReader;
-import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
-import org.apache.tsfile.write.TsFileWriter;
-import org.apache.tsfile.write.record.TSRecord;
-import org.apache.tsfile.write.record.datapoint.DoubleDataPoint;
-import org.apache.tsfile.write.record.datapoint.IntDataPoint;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
-import org.apache.tsfile.write.schema.MeasurementSchema;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-public class CompactionDataTypeAlterTest extends AbstractCompactionTest {
- private final String oldThreadName = Thread.currentThread().getName();
- private final IDeviceID device =
- IDeviceID.Factory.DEFAULT_FACTORY.create(COMPACTION_TEST_SG + ".d1");
-
- @Before
- public void setUp()
- throws IOException, WriteProcessException, MetadataException, InterruptedException {
- super.setUp();
- Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1");
- }
-
- @After
- public void tearDown() throws IOException, StorageEngineException {
- super.tearDown();
- Thread.currentThread().setName(oldThreadName);
- }
-
- @Test
- public void testCompactNonAlignedSeriesWithReadChunkCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithNonAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- @Test
- public void testCompactNonAlignedSeriesWithFastCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithNonAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- @Test
- public void testCompactNonAlignedSeriesWithReadPointCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithNonAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadPointCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- @Test
- public void testCompactAlignedSeriesWithReadChunkCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- paths.add(new Path(device, "s2", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t0.0\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0\t3.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- @Test
- public void testCompactAlignedSeriesWithFastCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- paths.add(new Path(device, "s2", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t0.0\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0\t3.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- @Test
- public void testCompactAlignedSeriesWithReadPointCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadPointCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
- TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
- try (TsFileSequenceReader reader =
- new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
- TsFileReader readTsFile = new TsFileReader(reader)) {
- ArrayList<Path> paths = new ArrayList<>();
- paths.add(new Path(device, "s1", true));
- paths.add(new Path(device, "s2", true));
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet queryDataSet = readTsFile.query(queryExpression);
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("1\t0.0\t1.0", queryDataSet.next().toString());
- Assert.assertTrue(queryDataSet.hasNext());
- Assert.assertEquals("2\t2.0\t3.0", queryDataSet.next().toString());
- Assert.assertFalse(queryDataSet.hasNext());
- }
- }
-
- private void generateDataTypeNotMatchFilesWithNonAlignedSeries()
- throws IOException, WriteProcessException {
- MeasurementSchema measurementSchema1 = new MeasurementSchema("s1", TSDataType.INT32);
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerTimeseries(new Path(device), measurementSchema1);
- TSRecord record = new TSRecord(device, 1);
- record.addTuple(new IntDataPoint("s1", 1));
- writer.writeRecord(record);
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
- seqResources.add(resource1);
-
- MeasurementSchema measurementSchema2 = new MeasurementSchema("s1", TSDataType.DOUBLE);
- TsFileResource resource2 = createEmptyFileAndResource(true);
- resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
- writer.registerTimeseries(new Path(device), measurementSchema2);
- TSRecord record = new TSRecord(device, 2);
- record.addTuple(new DoubleDataPoint("s1", 2.0));
- writer.writeRecord(record);
- writer.flush();
- }
- resource2.updateStartTime(device, 2);
- resource2.updateEndTime(device, 2);
- resource2.serialize();
- seqResources.add(resource2);
- }
-
- private void generateDataTypeNotMatchFilesWithAlignedSeries()
- throws IOException, WriteProcessException {
- List<IMeasurementSchema> measurementSchemas1 = new ArrayList<>();
- measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
- measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
-
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
- TSRecord record = new TSRecord(device, 1);
- record.addTuple(new IntDataPoint("s1", 0));
- record.addTuple(new IntDataPoint("s2", 1));
- writer.writeRecord(record);
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
- seqResources.add(resource1);
-
- List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
- measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
- measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
- TsFileResource resource2 = createEmptyFileAndResource(true);
- resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas2);
- TSRecord record = new TSRecord(device, 2);
- record.addTuple(new DoubleDataPoint("s1", 2.0));
- record.addTuple(new DoubleDataPoint("s2", 3.0));
- writer.writeRecord(record);
- writer.flush();
- }
- resource2.updateStartTime(device, 2);
- resource2.updateEndTime(device, 2);
- resource2.serialize();
- seqResources.add(resource2);
- }
-
- @Test
- public void testAlterDataTypeWithAlignedSeriesWithTimeDeletion()
- throws IOException, WriteProcessException, IllegalPathException {
- List<IMeasurementSchema> measurementSchemas1 = new ArrayList<>();
- measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
- measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
-
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
- for (int i = 0; i < 100; i++) {
- TSRecord record = new TSRecord(device, i);
- record.addTuple(new IntDataPoint("s1", 0));
- record.addTuple(new IntDataPoint("s2", 1));
- writer.writeRecord(record);
- }
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
- ModificationFile modFile = resource1.getExclusiveModFile();
- modFile.write(new TreeDeletionEntry(new MeasurementPath(device + ".*"), 0, 40));
- modFile.close();
- seqResources.add(resource1);
-
- List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
- measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
- measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
- TsFileResource resource2 = createEmptyFileAndResource(true);
- resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas2);
- TSRecord record = new TSRecord(device, 200);
- record.addTuple(new DoubleDataPoint("s1", 2.0));
- record.addTuple(new DoubleDataPoint("s2", 3.0));
- writer.writeRecord(record);
- writer.flush();
- }
- resource2.updateStartTime(device, 2);
- resource2.updateEndTime(device, 2);
- resource2.serialize();
- seqResources.add(resource2);
-
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- }
-}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/AbstractCompactionAlterDataTypeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/AbstractCompactionAlterDataTypeTest.java
new file mode 100644
index 00000000000..d97b66b68b7
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/AbstractCompactionAlterDataTypeTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.storageengine.dataregion.compaction.alterDataType;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionFakeSchemaFetcherImpl;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.TSRecord;
+import org.apache.tsfile.write.record.datapoint.DoubleDataPoint;
+import org.apache.tsfile.write.record.datapoint.FloatDataPoint;
+import org.apache.tsfile.write.record.datapoint.IntDataPoint;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class AbstractCompactionAlterDataTypeTest extends AbstractCompactionTest {
+
+ protected final String oldThreadName = Thread.currentThread().getName();
+ protected final IDeviceID device =
+ IDeviceID.Factory.DEFAULT_FACTORY.create(COMPACTION_TEST_SG + ".d1");
+
+ protected CompactionFakeSchemaFetcherImpl schemaFetcher;
+
+ @Before
+ public void setUp()
+ throws IOException, WriteProcessException, MetadataException, InterruptedException {
+ super.setUp();
+ Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1");
+ this.schemaFetcher = new CompactionFakeSchemaFetcherImpl();
+ schemaFetcher.getSchemaTree().setDatabases(Collections.singleton(COMPACTION_TEST_SG));
+ CompactionUtils.setSchemaFetcher(schemaFetcher);
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ Thread.currentThread().setName(oldThreadName);
+ }
+
+ protected TsFileResource generateInt32AlignedSeriesFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ List<IMeasurementSchema> measurementSchemas1 = new ArrayList<>();
+ measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
+ measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource.getTsFile())) {
+ writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new IntDataPoint("s1", (int) i));
+ record.addTuple(new IntDataPoint("s2", (int) i));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource.updateStartTime(device, timeRange.getMin());
+ resource.updateEndTime(device, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+
+ protected TsFileResource generateDoubleAlignedSeriesFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+
+ List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
+ measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
+ measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource.getTsFile())) {
+ writer.registerAlignedTimeseries(new Path(device), measurementSchemas2);
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new DoubleDataPoint("s1", (double) i));
+ record.addTuple(new DoubleDataPoint("s2", (double) i));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource.updateStartTime(device, timeRange.getMin());
+ resource.updateEndTime(device, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+
+ protected TsFileResource generateInt32NonAlignedSeriesFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ MeasurementSchema measurementSchema = new MeasurementSchema("s1", TSDataType.INT32);
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource.getTsFile())) {
+ writer.registerTimeseries(new Path(device), measurementSchema);
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new IntDataPoint("s1", (int) i));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource.updateStartTime(device, timeRange.getMin());
+ resource.updateEndTime(device, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+
+ protected TsFileResource generateFloatNonAlignedSeriesFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ MeasurementSchema measurementSchema = new MeasurementSchema("s1", TSDataType.FLOAT);
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource.getTsFile())) {
+ writer.registerTimeseries(new Path(device), measurementSchema);
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new FloatDataPoint("s1", (float) i));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource.updateStartTime(device, timeRange.getMin());
+ resource.updateEndTime(device, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+
+ protected TsFileResource generateDoubleNonAlignedSeriesFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ MeasurementSchema measurementSchema2 = new MeasurementSchema("s1", TSDataType.DOUBLE);
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource.getTsFile())) {
+ writer.registerTimeseries(new Path(device), measurementSchema2);
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new DoubleDataPoint("s1", (double) i));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource.updateStartTime(device, timeRange.getMin());
+ resource.updateEndTime(device, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTableTest.java
new file mode 100644
index 00000000000..24551efa8d2
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTableTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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.storageengine.dataregion.compaction.alterDataType;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+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.IDeviceID;
+import org.apache.tsfile.file.metadata.StringArrayDeviceID;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.v4.ITsFileWriter;
+import org.apache.tsfile.write.v4.TsFileWriterBuilder;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+
+@RunWith(Parameterized.class)
+public class CompactionDataTypeAlterTableTest extends AbstractCompactionAlterDataTypeTest {
+
+ private IDeviceID tableDevice = new StringArrayDeviceID("table1.d1");
+
+ private boolean reverse;
+ private String performerType;
+
+ public CompactionDataTypeAlterTableTest(boolean reverse, String performerType) {
+ this.reverse = reverse;
+ this.performerType = performerType;
+ }
+
+ @Parameterized.Parameters(name = "reverse={0} performerType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(
+ new Object[][] {
+ {true, "read_chunk"},
+ {false, "read_chunk"},
+ {true, "fast"},
+ {false, "fast"},
+ {true, "read_point"},
+ {false, "read_point"},
+ });
+ }
+
+ @Before
+ @Override
+ public void setUp()
+ throws IOException, WriteProcessException, MetadataException, InterruptedException {
+ super.setUp();
+ DataNodeTableCache.getInstance().invalid(COMPACTION_TEST_SG);
+ }
+
+ @After
+ @Override
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ DataNodeTableCache.getInstance().invalid(COMPACTION_TEST_SG);
+ }
+
+ @Test
+ public void testAlter() throws IOException, WriteProcessException {
+ generateDataTypeNotMatchedFiles(true);
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
+ Assert.assertTrue(task.start());
+ TsFileResource target = tsFileManager.getTsFileList(true).get(0);
+
+ TimeseriesMetadata timeseriesMetadata = getTimeseriesMetadata(target, tableDevice, "s1");
+ Assert.assertEquals(1L, timeseriesMetadata.getStatistics().getStartTime());
+ Assert.assertEquals(2L, timeseriesMetadata.getStatistics().getEndTime());
+ }
+
+ @Test
+ public void testCannotAlter() throws IOException, WriteProcessException {
+ generateDataTypeNotMatchedFiles(false);
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
+ Assert.assertTrue(task.start());
+ TsFileResource target = tsFileManager.getTsFileList(true).get(0);
+ TimeseriesMetadata timeseriesMetadata = getTimeseriesMetadata(target, tableDevice, "s1");
+ if (!reverse) {
+ Assert.assertEquals(1L, timeseriesMetadata.getStatistics().getStartTime());
+ Assert.assertEquals(1L, timeseriesMetadata.getStatistics().getEndTime());
+ } else {
+ Assert.assertEquals(2L, timeseriesMetadata.getStatistics().getStartTime());
+ Assert.assertEquals(2L, timeseriesMetadata.getStatistics().getEndTime());
+ }
+ }
+
+ private TimeseriesMetadata getTimeseriesMetadata(
+ TsFileResource resource, IDeviceID deviceID, String measurement) throws IOException {
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(resource.getTsFilePath())) {
+ return reader.readTimeseriesMetadata(deviceID, measurement, true);
+ }
+ }
+
+ private void generateDataTypeNotMatchedFiles(boolean canAlter)
+ throws IOException, WriteProcessException {
+ if (!reverse) {
+ TsFileResource resource1 = generateInt32TableFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+ TsFileResource resource2 = generateDoubleTableFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ } else {
+ TsFileResource resource1 = generateDoubleTableFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+ TsFileResource resource2 = generateInt32TableFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ }
+ if (canAlter) {
+ createTable(tableDevice.getTableName(), TSDataType.DOUBLE);
+ } else {
+ createTable(tableDevice.getTableName(), TSDataType.INT32);
+ }
+ }
+
+ private void createTable(String tableName, TSDataType dataType) {
+ TsTable tsTable = new TsTable(tableName);
+ tsTable.addColumnSchema(new TagColumnSchema("id_column", TSDataType.STRING));
+ tsTable.addColumnSchema(
+ new FieldColumnSchema("s1", dataType, TSEncoding.PLAIN, CompressionType.LZ4));
+ DataNodeTableCache.getInstance().preUpdateTable(this.COMPACTION_TEST_SG, tsTable, null);
+ DataNodeTableCache.getInstance().commitUpdateTable(this.COMPACTION_TEST_SG, tableName, null);
+ }
+
+ private TsFileResource generateInt32TableFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ TableSchema tableSchema =
+ new TableSchema(
+ tableDevice.getTableName(),
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.INT32),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD));
+ try (ITsFileWriter writer =
+ new TsFileWriterBuilder().file(resource.getTsFile()).tableSchema(tableSchema).build()) {
+ Tablet tablet =
+ new Tablet(
+ "table1",
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.INT32),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD));
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
+ writer.write(tablet);
+ tablet.reset();
+ }
+ int row = tablet.getRowSize();
+ tablet.addTimestamp(row, i);
+ tablet.addValue(row, 0, "d1");
+ tablet.addValue(row, 1, 1);
+ }
+ if (tablet.getRowSize() > 0) {
+ writer.write(tablet);
+ }
+ }
+ resource.updateStartTime(tableDevice, timeRange.getMin());
+ resource.updateEndTime(tableDevice, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+
+ private TsFileResource generateDoubleTableFile(TimeRange timeRange, boolean seq)
+ throws IOException, WriteProcessException {
+ TsFileResource resource = createEmptyFileAndResource(seq);
+ resource.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ TableSchema tableSchema =
+ new TableSchema(
+ tableDevice.getTableName(),
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD));
+ try (ITsFileWriter writer =
+ new TsFileWriterBuilder().file(resource.getTsFile()).tableSchema(tableSchema).build()) {
+ Tablet tablet =
+ new Tablet(
+ "table1",
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD));
+ for (long i = timeRange.getMin(); i <= timeRange.getMax(); i++) {
+ if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
+ writer.write(tablet);
+ tablet.reset();
+ }
+ int row = tablet.getRowSize();
+ tablet.addTimestamp(row, i);
+ tablet.addValue(row, 0, "d1");
+ tablet.addValue(row, 1, (double) 1);
+ }
+ if (tablet.getRowSize() > 0) {
+ writer.write(tablet);
+ }
+ }
+ resource.updateStartTime(tableDevice, timeRange.getMin());
+ resource.updateEndTime(tableDevice, timeRange.getMax());
+ resource.serialize();
+ return resource;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTest.java
new file mode 100644
index 00000000000..d20e0307a32
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeAlterTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.storageengine.dataregion.compaction.alterDataType;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.read.TsFileReader;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
+import org.apache.tsfile.read.expression.QueryExpression;
+import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+@RunWith(Parameterized.class)
+public class CompactionDataTypeAlterTest extends AbstractCompactionAlterDataTypeTest {
+ private boolean reverse;
+ private String performerType;
+
+ public CompactionDataTypeAlterTest(boolean reverse, String performerType) {
+ this.reverse = reverse;
+ this.performerType = performerType;
+ }
+
+ @Parameterized.Parameters(name = "reverse={0} performerType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(
+ new Object[][] {
+ {true, "read_chunk"},
+ {false, "read_chunk"},
+ {true, "fast"},
+ {false, "fast"},
+ {true, "read_point"},
+ {false, "read_point"},
+ });
+ }
+
+ @Before
+ public void setUp()
+ throws IOException, WriteProcessException, MetadataException, InterruptedException {
+ super.setUp();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ }
+
+ @Test
+ public void testCompactNonAlignedSeries()
+ throws IOException, WriteProcessException, IllegalPathException {
+ generateDataTypeNotMatchFilesWithNonAlignedSeries();
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
+ Assert.assertTrue(task.start());
+ TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
+ Assert.assertEquals(
+ 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
+ Assert.assertEquals(
+ 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
+ TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
+ TsFileReader readTsFile = new TsFileReader(reader)) {
+ ArrayList<Path> paths = new ArrayList<>();
+ paths.add(new Path(device, "s1", true));
+ QueryExpression queryExpression = QueryExpression.create(paths, null);
+ QueryDataSet queryDataSet = readTsFile.query(queryExpression);
+ Assert.assertTrue(queryDataSet.hasNext());
+ Assert.assertEquals("1\t1.0", queryDataSet.next().toString());
+ Assert.assertTrue(queryDataSet.hasNext());
+ Assert.assertEquals("2\t2.0", queryDataSet.next().toString());
+ Assert.assertFalse(queryDataSet.hasNext());
+ }
+ }
+
+ @Test
+ public void testCompactAlignedSeries()
+ throws IOException, WriteProcessException, IllegalPathException {
+ generateDataTypeNotMatchFilesWithAlignedSeries();
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
+ Assert.assertTrue(task.start());
+ TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
+ Assert.assertEquals(
+ 1, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
+ Assert.assertEquals(
+ 2, ((long) tsFileManager.getTsFileList(true).get(0).getEndTime(device).get()));
+ TsFileResource tsFileResource = tsFileManager.getTsFileList(true).get(0);
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath());
+ TsFileReader readTsFile = new TsFileReader(reader)) {
+ ArrayList<Path> paths = new ArrayList<>();
+ paths.add(new Path(device, "s1", true));
+ paths.add(new Path(device, "s2", true));
+ QueryExpression queryExpression = QueryExpression.create(paths, null);
+ QueryDataSet queryDataSet = readTsFile.query(queryExpression);
+ Assert.assertTrue(queryDataSet.hasNext());
+ Assert.assertEquals("1\t1.0\t1.0", queryDataSet.next().toString());
+ Assert.assertTrue(queryDataSet.hasNext());
+ Assert.assertEquals("2\t2.0\t2.0", queryDataSet.next().toString());
+ Assert.assertFalse(queryDataSet.hasNext());
+ }
+ }
+
+ private void generateDataTypeNotMatchFilesWithNonAlignedSeries()
+ throws IOException, WriteProcessException, IllegalPathException {
+ if (!reverse) {
+ TsFileResource resource1 = generateInt32NonAlignedSeriesFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+
+ TsFileResource resource2 = generateDoubleNonAlignedSeriesFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ } else {
+ TsFileResource resource1 = generateDoubleNonAlignedSeriesFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+
+ TsFileResource resource2 = generateInt32NonAlignedSeriesFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ }
+
+ schemaFetcher
+ .getSchemaTree()
+ .appendSingleMeasurementPath(
+ new MeasurementPath(device, "s1", new MeasurementSchema("s1", TSDataType.DOUBLE)));
+ }
+
+ private void generateDataTypeNotMatchFilesWithAlignedSeries()
+ throws IOException, WriteProcessException, IllegalPathException {
+ if (!reverse) {
+ TsFileResource resource1 = generateInt32AlignedSeriesFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+ TsFileResource resource2 = generateDoubleAlignedSeriesFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ } else {
+ TsFileResource resource1 = generateDoubleAlignedSeriesFile(new TimeRange(1, 1), true);
+ seqResources.add(resource1);
+ TsFileResource resource2 = generateInt32AlignedSeriesFile(new TimeRange(2, 2), true);
+ seqResources.add(resource2);
+ }
+
+ List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
+ measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
+ measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
+
+ MeasurementPath s1Path = new MeasurementPath(device, "s1", measurementSchemas2.get(0));
+ s1Path.setUnderAlignedEntity(true);
+ MeasurementPath s2Path = new MeasurementPath(device, "s2", measurementSchemas2.get(1));
+ s2Path.setUnderAlignedEntity(true);
+ schemaFetcher.getSchemaTree().appendSingleMeasurementPath(s1Path);
+ schemaFetcher.getSchemaTree().appendSingleMeasurementPath(s2Path);
+ }
+
+ @Test
+ public void testAlterDataTypeWithAlignedSeriesWithTimeDeletion()
+ throws IOException, WriteProcessException, IllegalPathException {
+ List<IMeasurementSchema> measurementSchemas1 = new ArrayList<>();
+ measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
+ measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ TsFileResource resource1 = generateInt32AlignedSeriesFile(new TimeRange(0, 100), true);
+ ModificationFile modFile = resource1.getExclusiveModFile();
+ modFile.write(new TreeDeletionEntry(new MeasurementPath(device + ".*"), 0, 40));
+ modFile.close();
+ seqResources.add(resource1);
+
+ List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
+ measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
+ measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
+ TsFileResource resource2 = generateDoubleAlignedSeriesFile(new TimeRange(200, 200), true);
+ seqResources.add(resource2);
+
+ MeasurementPath s1Path = new MeasurementPath(device, "s1", measurementSchemas2.get(0));
+ s1Path.setUnderAlignedEntity(true);
+ MeasurementPath s2Path = new MeasurementPath(device, "s2", measurementSchemas2.get(1));
+ s2Path.setUnderAlignedEntity(true);
+ schemaFetcher.getSchemaTree().appendSingleMeasurementPath(s1Path);
+ schemaFetcher.getSchemaTree().appendSingleMeasurementPath(s2Path);
+
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
+ Assert.assertTrue(task.start());
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchAlterableDataTypeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchAlterableDataTypeTest.java
similarity index 77%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchAlterableDataTypeTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchAlterableDataTypeTest.java
index 66d3c97e6c6..c7782a411c0 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchAlterableDataTypeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchAlterableDataTypeTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.storageengine.dataregion.compaction;
+package org.apache.iotdb.db.storageengine.dataregion.compaction.alterDataType;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -31,13 +31,12 @@ import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.write.WriteProcessException;
-import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.write.TsFileWriter;
import org.apache.tsfile.write.record.TSRecord;
import org.apache.tsfile.write.record.datapoint.DoubleDataPoint;
import org.apache.tsfile.write.record.datapoint.FloatDataPoint;
-import org.apache.tsfile.write.record.datapoint.IntDataPoint;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
@@ -50,22 +49,18 @@ import java.util.ArrayList;
import java.util.List;
@SuppressWarnings("OptionalGetWithoutIsPresent")
-public class CompactionDataTypeNotMatchAlterableDataTypeTest extends AbstractCompactionTest {
- private final String oldThreadName = Thread.currentThread().getName();
- private final IDeviceID device =
- IDeviceID.Factory.DEFAULT_FACTORY.create(COMPACTION_TEST_SG + ".d1");
+public class CompactionDataTypeNotMatchAlterableDataTypeTest
+ extends AbstractCompactionAlterDataTypeTest {
@Before
public void setUp()
throws IOException, WriteProcessException, MetadataException, InterruptedException {
super.setUp();
- Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1");
}
@After
public void tearDown() throws IOException, StorageEngineException {
super.tearDown();
- Thread.currentThread().setName(oldThreadName);
}
@Test
@@ -149,33 +144,11 @@ public class CompactionDataTypeNotMatchAlterableDataTypeTest extends AbstractCom
private void generateDataTypeNotMatchFilesWithNonAlignedSeries()
throws IOException, WriteProcessException {
MeasurementSchema measurementSchema1 = new MeasurementSchema("s1", TSDataType.INT32);
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerTimeseries(new Path(device), measurementSchema1);
- TSRecord record = new TSRecord(device, 1);
- record.addTuple(new IntDataPoint("s1", 1));
- writer.writeRecord(record);
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
+ TsFileResource resource1 = generateInt32NonAlignedSeriesFile(new TimeRange(1, 1), true);
seqResources.add(resource1);
MeasurementSchema measurementSchema2 = new MeasurementSchema("s1", TSDataType.FLOAT);
- TsFileResource resource2 = createEmptyFileAndResource(true);
- resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
- writer.registerTimeseries(new Path(device), measurementSchema2);
- TSRecord record = new TSRecord(device, 2);
- record.addTuple(new FloatDataPoint("s1", 2));
- writer.writeRecord(record);
- writer.flush();
- }
- resource2.updateStartTime(device, 2);
- resource2.updateEndTime(device, 2);
- resource2.serialize();
+ TsFileResource resource2 = generateFloatNonAlignedSeriesFile(new TimeRange(2, 2), true);
seqResources.add(resource2);
}
@@ -185,19 +158,7 @@ public class CompactionDataTypeNotMatchAlterableDataTypeTest extends AbstractCom
measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
- TSRecord record = new TSRecord(device, 1);
- record.addTuple(new IntDataPoint("s1", 0));
- record.addTuple(new IntDataPoint("s2", 1));
- writer.writeRecord(record);
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
+ TsFileResource resource1 = generateInt32AlignedSeriesFile(new TimeRange(1, 1), true);
seqResources.add(resource1);
List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchTest.java
similarity index 57%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchTest.java
index ff12a13c6da..3b4dcacb8fa 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeNotMatchTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/alterDataType/CompactionDataTypeNotMatchTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.storageengine.dataregion.compaction;
+package org.apache.iotdb.db.storageengine.dataregion.compaction.alterDataType;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -33,73 +33,62 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.write.TsFileWriter;
import org.apache.tsfile.write.record.TSRecord;
import org.apache.tsfile.write.record.datapoint.BooleanDataPoint;
-import org.apache.tsfile.write.record.datapoint.IntDataPoint;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
@SuppressWarnings("OptionalGetWithoutIsPresent")
-public class CompactionDataTypeNotMatchTest extends AbstractCompactionTest {
- private final String oldThreadName = Thread.currentThread().getName();
+@RunWith(Parameterized.class)
+public class CompactionDataTypeNotMatchTest extends AbstractCompactionAlterDataTypeTest {
private final IDeviceID device =
IDeviceID.Factory.DEFAULT_FACTORY.create(COMPACTION_TEST_SG + ".d1");
+ private String performerType;
@Before
public void setUp()
throws IOException, WriteProcessException, MetadataException, InterruptedException {
super.setUp();
- Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1");
}
@After
public void tearDown() throws IOException, StorageEngineException {
super.tearDown();
- Thread.currentThread().setName(oldThreadName);
}
- @Test
- public void testCompactNonAlignedSeriesWithReadChunkCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithNonAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
+ @Parameterized.Parameters(name = "type={0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(
+ new Object[][] {
+ {"read_chunk"}, {"fast"}, {"read_point"},
+ });
}
- @Test
- public void testCompactNonAlignedSeriesWithFastCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithNonAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
+ public CompactionDataTypeNotMatchTest(String type) {
+ this.performerType = type;
}
@Test
- public void testCompactNonAlignedSeriesWithReadPointCompactionPerformer()
+ public void testCompactNonAlignedSeries()
throws IOException, WriteProcessException {
generateDataTypeNotMatchFilesWithNonAlignedSeries();
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadPointCompactionPerformer(), 0);
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
Assert.assertTrue(task.start());
TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
Assert.assertEquals(
@@ -107,43 +96,18 @@ public class CompactionDataTypeNotMatchTest extends AbstractCompactionTest {
}
@Test
- public void testCompactAlignedSeriesWithReadChunkCompactionPerformer()
+ public void testCompactAlignedSeries()
throws IOException, WriteProcessException {
generateDataTypeNotMatchFilesWithAlignedSeries();
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
Assert.assertTrue(task.start());
TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
Assert.assertEquals(
2, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
}
- @Test
- public void testCompactAlignedSeriesWithFastCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new FastCompactionPerformer(false), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- }
-
- @Test
- public void testCompactAlignedSeriesWithReadPointCompactionPerformer()
- throws IOException, WriteProcessException {
- generateDataTypeNotMatchFilesWithAlignedSeries();
- InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(
- 0, tsFileManager, seqResources, true, new ReadPointCompactionPerformer(), 0);
- Assert.assertTrue(task.start());
- TsFileResourceUtils.validateTsFileDataCorrectness(tsFileManager.getTsFileList(true).get(0));
- Assert.assertEquals(
- 2, ((long) tsFileManager.getTsFileList(true).get(0).getStartTime(device).get()));
- }
private void generateDataTypeNotMatchFilesWithNonAlignedSeries()
throws IOException, WriteProcessException {
@@ -163,18 +127,7 @@ public class CompactionDataTypeNotMatchTest extends AbstractCompactionTest {
seqResources.add(resource1);
MeasurementSchema measurementSchema2 = new MeasurementSchema("s1", TSDataType.INT32);
- TsFileResource resource2 = createEmptyFileAndResource(true);
- resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
- writer.registerTimeseries(new Path(device), measurementSchema2);
- TSRecord record = new TSRecord(device, 2);
- record.addTuple(new IntDataPoint("s1", 10));
- writer.writeRecord(record);
- writer.flush();
- }
- resource2.updateStartTime(device, 2);
- resource2.updateEndTime(device, 2);
- resource2.serialize();
+ TsFileResource resource2 = generateInt32NonAlignedSeriesFile(new TimeRange(2, 2), true);
seqResources.add(resource2);
}
@@ -184,19 +137,7 @@ public class CompactionDataTypeNotMatchTest extends AbstractCompactionTest {
measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
- TsFileResource resource1 = createEmptyFileAndResource(true);
- resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
- try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
- writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
- TSRecord record = new TSRecord(device, 1);
- record.addTuple(new IntDataPoint("s1", 0));
- record.addTuple(new IntDataPoint("s2", 1));
- writer.writeRecord(record);
- writer.flush();
- }
- resource1.updateStartTime(device, 1);
- resource1.updateEndTime(device, 1);
- resource1.serialize();
+ TsFileResource resource1 = generateInt32AlignedSeriesFile(new TimeRange(1, 1), true);
seqResources.add(resource1);
List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/TableModelCompactionWithTTLTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/TableModelCompactionWithTTLTest.java
index ef9e2fb796e..6746f0a2746 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/TableModelCompactionWithTTLTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/TableModelCompactionWithTTLTest.java
@@ -26,12 +26,6 @@ import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerSeqCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerUnseqCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -86,16 +80,6 @@ public class TableModelCompactionWithTTLTest extends AbstractCompactionTest {
});
}
- public ICompactionPerformer getPerformer() {
- if (performerType.equalsIgnoreCase(InnerSeqCompactionPerformer.READ_CHUNK.toString())) {
- return new ReadChunkCompactionPerformer();
- } else if (performerType.equalsIgnoreCase(InnerUnseqCompactionPerformer.FAST.toString())) {
- return new FastCompactionPerformer(false);
- } else {
- return new ReadPointCompactionPerformer();
- }
- }
-
@Test
public void testAllDataExpired() throws IOException {
createTable("t1", 1);
@@ -115,7 +99,8 @@ public class TableModelCompactionWithTTLTest extends AbstractCompactionTest {
}
seqResources.add(resource1);
InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, getPerformer(), 0);
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
Assert.assertTrue(task.start());
Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
}
@@ -143,7 +128,8 @@ public class TableModelCompactionWithTTLTest extends AbstractCompactionTest {
}
seqResources.add(resource1);
InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, getPerformer(), 0);
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
Assert.assertTrue(task.start());
TsFileResource target = tsFileManager.getTsFileList(true).get(0);
Assert.assertTrue(target.getFileStartTime() > startTime && target.getFileEndTime() == endTime);
@@ -171,7 +157,8 @@ public class TableModelCompactionWithTTLTest extends AbstractCompactionTest {
}
seqResources.add(resource1);
InnerSpaceCompactionTask task =
- new InnerSpaceCompactionTask(0, tsFileManager, seqResources, true, getPerformer(), 0);
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, getPerformer(performerType), 0);
Assert.assertTrue(task.start());
TsFileResource target = tsFileManager.getTsFileList(true).get(0);
Assert.assertTrue(target.getFileStartTime() == startTime && target.getFileEndTime() == endTime);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFakeSchemaFetcherImpl.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFakeSchemaFetcherImpl.java
new file mode 100644
index 00000000000..173c1519897
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionFakeSchemaFetcherImpl.java
@@ -0,0 +1,37 @@
+/*
+ * 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.storageengine.dataregion.compaction.utils;
+
+import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.common.schematree.node.SchemaInternalNode;
+import org.apache.iotdb.db.queryengine.common.schematree.node.SchemaNode;
+import org.apache.iotdb.db.queryengine.plan.analyze.FakeSchemaFetcherImpl;
+
+public class CompactionFakeSchemaFetcherImpl extends FakeSchemaFetcherImpl {
+
+ @Override
+ protected SchemaNode generateSchemaTree() {
+ return new SchemaInternalNode("root");
+ }
+
+ public ClusterSchemaTree getSchemaTree() {
+ return schemaTree;
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
index 8f484f4e231..48cd81b39e1 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
@@ -70,7 +70,6 @@ public class TsTable {
private final Map<String, TsTableColumnSchema> columnSchemaMap = new LinkedHashMap<>();
private final Map<String, Integer> tagColumnIndexMap = new HashMap<>();
- private final Map<String, Integer> idColumnIndexMap = new HashMap<>();
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
@@ -108,7 +107,7 @@ public class TsTable {
public TsTable(TsTable origin) {
this.tableName = origin.tableName;
origin.columnSchemaMap.forEach((col, schema) -> this.columnSchemaMap.put(col, schema.copy()));
- this.idColumnIndexMap.putAll(origin.idColumnIndexMap);
+ this.tagColumnIndexMap.putAll(origin.tagColumnIndexMap);
this.props = origin.props == null ? null : new HashMap<>(origin.props);
this.ttlValue = origin.ttlValue;
this.tagNums = origin.tagNums;