This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new fb0b2d9301f [Pipe] Support configurable TsFile parser (#18440) (#18449)
fb0b2d9301f is described below
commit fb0b2d9301f91ec5671de350e35f4d4dba59df42
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 12 14:10:51 2026 +0800
[Pipe] Support configurable TsFile parser (#18440) (#18449)
(cherry picked from commit 6ca8850a25bb333b5345d094ceea14511aab5a61)
---
.../tsfile/PipeCompactedTsFileInsertionEvent.java | 1 +
.../common/tsfile/PipeTsFileInsertionEvent.java | 41 +++++++++++------
.../TsFileInsertionDataContainerProvider.java | 53 ++++++++++++++++++++++
.../source/dataregion/IoTDBDataRegionSource.java | 17 +++++++
.../PipeHistoricalDataRegionTsFileSource.java | 6 +++
.../realtime/PipeRealtimeDataRegionSource.java | 9 ++++
.../realtime/assigner/PipeDataRegionAssigner.java | 1 +
.../event/TsFileInsertionDataContainerTest.java | 37 +++++++++++++++
.../db/pipe/source/IoTDBDataRegionSourceTest.java | 31 +++++++++++++
.../pipe/config/constant/PipeSourceConstant.java | 5 ++
10 files changed, 187 insertions(+), 14 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
index 25beecfbea6..20f345dceed 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
@@ -81,6 +81,7 @@ public class PipeCompactedTsFileInsertionEvent extends
PipeTsFileInsertionEvent
// init fields of PipeTsFileInsertionEvent
flushPointCount = bindFlushPointCount(originalEvents);
overridingProgressIndex = bindOverridingProgressIndex(originalEvents);
+ setTsFileParser(anyOfOriginalEvents.getTsFileParser());
}
private static boolean bindIsWithMod(Set<PipeTsFileInsertionEvent>
originalEvents) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index a4859c33579..b6938890cf0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -104,6 +104,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
protected volatile ProgressIndex overridingProgressIndex;
private Set<String> tableNames;
+ private String tsFileParser;
public PipeTsFileInsertionEvent(final TsFileResource resource, final boolean
isLoaded) {
// The modFile must be copied before the event is assigned to the
listening pipes
@@ -418,6 +419,14 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
}
}
+ public String getTsFileParser() {
+ return tsFileParser;
+ }
+
+ public void setTsFileParser(final String tsFileParser) {
+ this.tsFileParser = tsFileParser;
+ }
+
@Override
public PipeTsFileInsertionEvent
shallowCopySelfAndBindPipeTaskMetaForProgressReport(
final String pipeName,
@@ -426,19 +435,22 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
final PipePattern pattern,
final long startTime,
final long endTime) {
- return new PipeTsFileInsertionEvent(
- resource,
- tsFile,
- isWithMod,
- isLoaded,
- isGeneratedByHistoricalExtractor,
- pipeName,
- creationTime,
- pipeTaskMeta,
- pattern,
- startTime,
- endTime,
- isTsFileSealed);
+ final PipeTsFileInsertionEvent copiedEvent =
+ new PipeTsFileInsertionEvent(
+ resource,
+ tsFile,
+ isWithMod,
+ isLoaded,
+ isGeneratedByHistoricalExtractor,
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ pattern,
+ startTime,
+ endTime,
+ isTsFileSealed);
+ copiedEvent.setTsFileParser(tsFileParser);
+ return copiedEvent;
}
@Override
@@ -786,7 +798,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
startTime,
endTime,
pipeTaskMeta,
- this)
+ this,
+ tsFileParser)
.provide(isWithMod));
return dataContainer.get();
} catch (final IOException e) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
index 59d630a90ce..563de66f0ec 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
@@ -40,6 +40,9 @@ import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_QUERY_VALUE;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_SCAN_VALUE;
+
public class TsFileInsertionDataContainerProvider {
private final String pipeName;
@@ -52,6 +55,7 @@ public class TsFileInsertionDataContainerProvider {
protected final PipeTaskMeta pipeTaskMeta;
protected final PipeTsFileInsertionEvent sourceEvent;
+ private final String tsFileParser;
public TsFileInsertionDataContainerProvider(
final String pipeName,
@@ -62,6 +66,28 @@ public class TsFileInsertionDataContainerProvider {
final long endTime,
final PipeTaskMeta pipeTaskMeta,
final PipeTsFileInsertionEvent sourceEvent) {
+ this(
+ pipeName,
+ creationTime,
+ tsFile,
+ pipePattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ null);
+ }
+
+ public TsFileInsertionDataContainerProvider(
+ final String pipeName,
+ final long creationTime,
+ final File tsFile,
+ final PipePattern pipePattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final PipeTsFileInsertionEvent sourceEvent,
+ final String tsFileParser) {
this.pipeName = pipeName;
this.creationTime = creationTime;
this.tsFile = tsFile;
@@ -70,6 +96,7 @@ public class TsFileInsertionDataContainerProvider {
this.endTime = endTime;
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
+ this.tsFileParser = tsFileParser;
}
public TsFileInsertionDataContainer provide(final boolean isWithMod) throws
IOException {
@@ -78,6 +105,32 @@ public class TsFileInsertionDataContainerProvider {
.markTsFileToTabletInvocation(pipeName + "_" + creationTime);
}
+ if (EXTRACTOR_TSFILE_PARSER_QUERY_VALUE.equals(tsFileParser)) {
+ return new TsFileInsertionQueryDataContainer(
+ pipeName,
+ creationTime,
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod);
+ }
+
+ if (EXTRACTOR_TSFILE_PARSER_SCAN_VALUE.equals(tsFileParser)) {
+ return new TsFileInsertionScanDataContainer(
+ pipeName,
+ creationTime,
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod);
+ }
+
// Use scan container to save memory
if ((double)
PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes()
/ PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes()
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
index 9ec1e2c34d2..5c82af0d459 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
@@ -70,6 +70,9 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.E
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_MODE_LOG_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_QUERY_VALUE;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_SCAN_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_WATERMARK_INTERVAL_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_WATERMARK_INTERVAL_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_END_TIME_KEY;
@@ -80,6 +83,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.S
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_MODE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_WATERMARK_INTERVAL_KEY;
public class IoTDBDataRegionSource extends IoTDBSource {
@@ -121,6 +125,19 @@ public class IoTDBDataRegionSource extends IoTDBSource {
EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE,
EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE);
+ // If unset, the parser is selected automatically.
+ validator
+ .validateAttributeValueRange(
+ EXTRACTOR_TSFILE_PARSER_KEY,
+ true,
+ EXTRACTOR_TSFILE_PARSER_QUERY_VALUE,
+ EXTRACTOR_TSFILE_PARSER_SCAN_VALUE)
+ .validateAttributeValueRange(
+ SOURCE_TSFILE_PARSER_KEY,
+ true,
+ EXTRACTOR_TSFILE_PARSER_QUERY_VALUE,
+ EXTRACTOR_TSFILE_PARSER_SCAN_VALUE);
+
// Get the pattern format to check whether the pattern is legal
final PipePattern pattern =
PipePattern.parsePipePatternFromSourceParameters(validator.getParameters());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
index 25e65ed8be9..25ace9387b1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
@@ -82,6 +82,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.E
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_END_TIME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY;
@@ -90,6 +91,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.S
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY;
public class PipeHistoricalDataRegionTsFileSource implements
PipeHistoricalDataRegionSource {
@@ -117,6 +119,7 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
private Pair<Boolean, Boolean> listeningOptionPair;
private boolean shouldExtractInsertion;
private boolean shouldTransferModFile; // Whether to transfer mods
+ private String tsFileParser;
private boolean shouldTerminatePipeOnAllHistoricalEventsConsumed;
private boolean isTerminateSignalSent = false;
@@ -266,6 +269,8 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
dataRegionId = environment.getRegionId();
pipePattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
+ tsFileParser =
+ parameters.getStringByKeys(EXTRACTOR_TSFILE_PARSER_KEY,
SOURCE_TSFILE_PARSER_KEY);
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new
DataRegionId(environment.getRegionId()));
@@ -622,6 +627,7 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
pipePattern,
historicalDataExtractionStartTime,
historicalDataExtractionEndTime);
+ event.setTsFileParser(tsFileParser);
if (sloppyPattern || isDbNameCoveredByPattern ||
isTsFileResourceCoveredByPattern(resource)) {
event.skipParsingPattern();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index fa716d5dcca..f72a4f06cfb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -74,10 +74,12 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.E
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_TSFILE_PARSER_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_END_TIME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_ENABLE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_LOOSE_RANGE_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_START_TIME_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY;
public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
@@ -110,6 +112,7 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
protected boolean isForwardingPipeRequests;
private boolean shouldTransferModFile; // Whether to transfer mods
+ private String tsFileParser;
private boolean sloppyTimeRange; // true to disable time range filter after
extraction
private boolean sloppyPattern; // true to disable pattern filter after
extraction
@@ -215,6 +218,8 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
taskID = pipeName + "_" + dataRegionId + "_" + creationTime;
pipePattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
+ tsFileParser =
+ parameters.getStringByKeys(EXTRACTOR_TSFILE_PARSER_KEY,
SOURCE_TSFILE_PARSER_KEY);
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new
DataRegionId(environment.getRegionId()));
@@ -508,6 +513,10 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
return shouldTransferModFile;
}
+ public final String getTsFileParser() {
+ return tsFileParser;
+ }
+
private void maySkipProgressIndexForRealtimeEvent(final PipeRealtimeEvent
event) {
if (PipeTsFileEpochProgressIndexKeeper.getInstance()
.isProgressIndexAfterOrEquals(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 01dcbb89265..dfce67865d3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -171,6 +171,7 @@ public class PipeDataRegionAssigner implements Closeable {
if (innerEvent instanceof PipeTsFileInsertionEvent) {
final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) innerEvent;
+
tsFileInsertionEvent.setTsFileParser(extractor.getTsFileParser());
tsFileInsertionEvent.disableMod4NonTransferPipes(
extractor.isShouldTransferModFile());
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
index dae32e56979..1d67bf68732 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainerProvider;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.AlignedSinglePageWholeChunkReader;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.SinglePageWholeChunkReader;
@@ -157,6 +158,42 @@ public class TsFileInsertionDataContainerTest {
System.out.println(System.currentTimeMillis() - startTime);
}
+ @Test
+ public void testConfiguredTsFileParserSelection() throws Exception {
+ final PipeTsFileInsertionEvent sourceEvent =
+
createPipeTsFileInsertionEventForRetryTest("configured-parser-selection.tsfile");
+
+ try (final TsFileInsertionDataContainer container =
+ new TsFileInsertionDataContainerProvider(
+ null,
+ 0,
+ nonalignedTsFile,
+ new PrefixPipePattern("root"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE,
+ null,
+ sourceEvent,
+ "query")
+ .provide(false)) {
+ Assert.assertTrue(container instanceof
TsFileInsertionQueryDataContainer);
+ }
+
+ try (final TsFileInsertionDataContainer container =
+ new TsFileInsertionDataContainerProvider(
+ null,
+ 0,
+ nonalignedTsFile,
+ new PrefixPipePattern("root"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE,
+ null,
+ sourceEvent,
+ "scan")
+ .provide(false)) {
+ Assert.assertTrue(container instanceof TsFileInsertionScanDataContainer);
+ }
+ }
+
@Test
public void testScanContainerReleasesTabletMemoryAfterRawTabletGenerated()
throws Exception {
nonalignedTsFile =
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java
index feca8d54fed..c5667624c5e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java
@@ -23,6 +23,7 @@ import
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
import org.junit.Assert;
import org.junit.Ignore;
@@ -53,6 +54,36 @@ public class IoTDBDataRegionSourceTest {
}
}
+ @Test
+ public void testTsFileParserParameter() throws Exception {
+ for (final String parser : new String[] {"query", "scan"}) {
+ try (final IoTDBDataRegionSource extractor = new
IoTDBDataRegionSource()) {
+ extractor.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY,
parser);
+ }
+ })));
+ }
+ }
+
+ try (final IoTDBDataRegionSource extractor = new IoTDBDataRegionSource()) {
+ Assert.assertThrows(
+ PipeParameterNotValidException.class,
+ () ->
+ extractor.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY,
"invalid");
+ }
+ }))));
+ }
+ }
+
@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testIoTDBDataRegionExtractorWithPattern() {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
index 4a7ca7534e8..fe53e228350 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
@@ -93,6 +93,11 @@ public class PipeSourceConstant {
public static final String SOURCE_MODS_ENABLE_KEY = "source.mods.enable";
public static final boolean EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE = false;
+ public static final String EXTRACTOR_TSFILE_PARSER_KEY =
"extractor.tsfile.parser";
+ public static final String SOURCE_TSFILE_PARSER_KEY = "source.tsfile.parser";
+ public static final String EXTRACTOR_TSFILE_PARSER_QUERY_VALUE = "query";
+ public static final String EXTRACTOR_TSFILE_PARSER_SCAN_VALUE = "scan";
+
public static final String EXTRACTOR_REALTIME_ENABLE_KEY =
"extractor.realtime.enable";
public static final String SOURCE_REALTIME_ENABLE_KEY =
"source.realtime.enable";
public static final boolean EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE = true;
