This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch load_v2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/load_v2 by this push:
new cd3890a55de add BatchedTsFileExtractor
cd3890a55de is described below
commit cd3890a55de1865f34a1fde8b35e8bb48c5fcce0
Author: Tian Jiang <[email protected]>
AuthorDate: Mon Oct 16 10:42:10 2023 +0800
add BatchedTsFileExtractor
---
.../dml/insertion/TsFileBatchInsertionEvent.java | 37 +++
.../tsfile/PipeBatchTsFileInsertionEvent.java | 240 +++++++++++++++++++
.../tsfile/TsFileListInsertionDataContainer.java | 261 +++++++++++++++++++++
.../historical/BatchedTsFileExtractor.java | 85 +++++++
.../PipeHistoricalDataRegionTsFileExtractor.java | 18 +-
5 files changed, 632 insertions(+), 9 deletions(-)
diff --git
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileBatchInsertionEvent.java
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileBatchInsertionEvent.java
new file mode 100644
index 00000000000..bc2b682d82f
--- /dev/null
+++
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileBatchInsertionEvent.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.pipe.api.event.dml.insertion;
+
+import org.apache.iotdb.pipe.api.event.Event;
+
+/**
+ * {@link TsFileBatchInsertionEvent} is used to define the event of writing a
batch of TsFiles.
+ * Event data stores in disks, which is compressed and encoded, and requires
IO cost for
+ * computational processing.
+ */
+public interface TsFileBatchInsertionEvent extends Event, AutoCloseable {
+
+ /**
+ * The method is used to convert the TsFileBatchInsertionEvent into several
TabletInsertionEvents.
+ *
+ * @return {@code Iterable<TabletInsertionEvent>} the list of
TabletInsertionEvent
+ */
+ Iterable<TabletInsertionEvent> toTabletInsertionEvents();
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
new file mode 100644
index 00000000000..284c84096ce
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileBatchInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PipeBatchTsFileInsertionEvent extends EnrichedEvent implements
+ TsFileBatchInsertionEvent {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeBatchTsFileInsertionEvent.class);
+
+ // used to filter data
+ private final long startTime;
+ private final long endTime;
+ private final boolean needParseTime;
+
+ private final List<TsFileResource> resources;
+ private List<File> tsFiles;
+
+ private final boolean isGeneratedByPipe;
+
+ private final AtomicBoolean[] isClosed;
+
+ private TsFileListInsertionDataContainer dataContainer;
+
+ public PipeBatchTsFileInsertionEvent(List<TsFileResource> resources, boolean
isGeneratedByPipe) {
+ this(resources, isGeneratedByPipe, null, null, Long.MIN_VALUE,
Long.MAX_VALUE, false);
+ }
+
+ public PipeBatchTsFileInsertionEvent(
+ List<TsFileResource> resources,
+ boolean isGeneratedByPipe,
+ PipeTaskMeta pipeTaskMeta,
+ String pattern,
+ long startTime,
+ long endTime,
+ boolean needParseTime) {
+ super(pipeTaskMeta, pattern);
+
+ this.startTime = startTime;
+ this.endTime = endTime;
+ this.needParseTime = needParseTime;
+
+ if (needParseTime) {
+ this.isPatternAndTimeParsed = false;
+ }
+
+ this.resources = resources;
+ tsFiles =
resources.stream().map(TsFileResource::getTsFile).collect(Collectors.toList());
+ isClosed = new AtomicBoolean[resources.size()];
+
+ this.isGeneratedByPipe = isGeneratedByPipe;
+
+ for (int i = 0; i < isClosed.length; i++) {
+ TsFileResource resource = resources.get(i);
+ isClosed[i] = new AtomicBoolean(resource.isClosed());
+ if (!isClosed[i].get()) {
+ final TsFileProcessor processor = resource.getProcessor();
+ if (processor != null) {
+ int finalI = i;
+ processor.addCloseFileListener(
+ o -> {
+ synchronized (isClosed[finalI]) {
+ isClosed[finalI].set(true);
+ isClosed[finalI].notifyAll();
+ }
+ });
+ }
+ }
+ // check again after register close listener in case TsFile is closed
during the process
+ isClosed[i].set(resource.isClosed());
+ }
+ }
+
+ public void waitForTsFileClose() throws InterruptedException {
+ for (AtomicBoolean signal : isClosed) {
+ if (!signal.get()) {
+ synchronized (signal) {
+ while (!signal.get()) {
+ isClosed.wait();
+ }
+ }
+ }
+ }
+ }
+
+ public List<File> getTsFiles() {
+ return tsFiles;
+ }
+
+ /////////////////////////// EnrichedEvent ///////////////////////////
+
+ @Override
+ public boolean internallyIncreaseResourceReferenceCount(String
holderMessage) {
+ for (int i = 0; i < tsFiles.size(); i++) {
+ File tsFile = tsFiles.get(i);
+ try {
+ tsFiles.set(i,
PipeResourceManager.tsfile().increaseFileReference(tsFile, true));
+ } catch (Exception e) {
+ LOGGER.warn(
+ String.format(
+ "Increase reference count for TsFile %s error. Holder Message:
%s",
+ tsFile.getPath(), holderMessage),
+ e);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean internallyDecreaseResourceReferenceCount(String
holderMessage) {
+ for (File tsFile : tsFiles) {
+ try {
+ PipeResourceManager.tsfile().decreaseFileReference(tsFile);
+ } catch (Exception e) {
+ LOGGER.warn(
+ String.format(
+ "Decrease reference count for TsFile %s error. Holder Message:
%s",
+ tsFile.getPath(), holderMessage),
+ e);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public ProgressIndex getProgressIndex() {
+ try {
+ waitForTsFileClose();
+ return resources.get(resources.size() -
1).getMaxProgressIndexAfterClose();
+ } catch (InterruptedException e) {
+ LOGGER.warn(
+ String.format(
+ "Interrupted when waiting for closing TsFiles %s.", resources));
+ Thread.currentThread().interrupt();
+ return MinimumProgressIndex.INSTANCE;
+ }
+ }
+
+ @Override
+ public PipeBatchTsFileInsertionEvent
shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ PipeTaskMeta pipeTaskMeta, String pattern) {
+ return new PipeBatchTsFileInsertionEvent(
+ resources, isGeneratedByPipe, pipeTaskMeta, pattern, startTime,
endTime, needParseTime);
+ }
+
+ @Override
+ public boolean isGeneratedByPipe() {
+ return isGeneratedByPipe;
+ }
+
+ /////////////////////////// TsFileInsertionEvent ///////////////////////////
+
+ @Override
+ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
+ try {
+ if (dataContainer == null) {
+ waitForTsFileClose();
+ dataContainer =
+ new TsFileListInsertionDataContainer(
+ tsFiles, getPattern(), startTime, endTime, pipeTaskMeta, this);
+ }
+ return dataContainer.toTabletInsertionEvents();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ close();
+
+ final String errorMsg =
+ String.format(
+ "Interrupted when waiting for closing TsFiles %s.", resources);
+ LOGGER.warn(errorMsg, e);
+ throw new PipeException(errorMsg);
+ } catch (IOException e) {
+ close();
+
+ final String errorMsg = String.format("Read TsFiles %s error.",
resources);
+ LOGGER.warn(errorMsg, e);
+ throw new PipeException(errorMsg);
+ }
+ }
+
+ /** Release the resource of data container. */
+ @Override
+ public void close() {
+ if (dataContainer != null) {
+ dataContainer.close();
+ dataContainer = null;
+ }
+ }
+
+ /////////////////////////// Object ///////////////////////////
+
+ @Override
+ public String toString() {
+ return "PipeTsFileInsertionEvent{"
+ + "resources="
+ + resources
+ + ", tsFiles="
+ + tsFiles
+ + ", isClosed="
+ + isClosed
+ + '}';
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
new file mode 100644
index 00000000000..6055f6f845d
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TsFileDeviceIterator;
+import org.apache.iotdb.tsfile.read.TsFileReader;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.expression.IExpression;
+import org.apache.iotdb.tsfile.read.expression.impl.BinaryExpression;
+import org.apache.iotdb.tsfile.read.expression.impl.GlobalTimeExpression;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TsFileListInsertionDataContainer implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(
+ TsFileListInsertionDataContainer.class);
+
+ private final String pattern; // used to filter data
+ private final IExpression timeFilterExpression; // used to filter data
+
+ private final PipeTaskMeta pipeTaskMeta; // used to report progress
+ private final EnrichedEvent sourceEvent; // used to report progress
+
+ private final List<TsFileSequenceReader> tsFileSequenceReaders;
+ private final List<TsFileReader> tsFileReaders;
+ private int currFileIndex = 0;
+
+ private final List<Iterator<Map.Entry<String, List<String>>>>
deviceMeasurementsMapIterators;
+ private final List<Map<String, Boolean>> deviceIsAlignedMaps;
+ private final List<Map<String, TSDataType>> measurementDataTypeMaps;
+
+ public TsFileListInsertionDataContainer(List<File> tsFiles, String pattern,
long startTime,
+ long endTime)
+ throws IOException {
+ this(tsFiles, pattern, startTime, endTime, null, null);
+ }
+
+ public TsFileListInsertionDataContainer(
+ List<File> tsFiles,
+ String pattern,
+ long startTime,
+ long endTime,
+ PipeTaskMeta pipeTaskMeta,
+ EnrichedEvent sourceEvent)
+ throws IOException {
+ this.pattern = pattern;
+ timeFilterExpression =
+ (startTime == Long.MIN_VALUE && endTime == Long.MAX_VALUE)
+ ? null
+ : BinaryExpression.and(
+ new GlobalTimeExpression(TimeFilter.gtEq(startTime)),
+ new GlobalTimeExpression(TimeFilter.ltEq(endTime)));
+
+ this.pipeTaskMeta = pipeTaskMeta;
+ this.sourceEvent = sourceEvent;
+
+ try {
+ tsFileSequenceReaders = new ArrayList<>();
+ tsFileReaders = new ArrayList<>();
+ for (File tsFile : tsFiles) {
+ TsFileSequenceReader tsFileSequenceReader = new TsFileSequenceReader(
+ tsFile.getAbsolutePath(), true, true);
+ tsFileSequenceReaders.add(tsFileSequenceReader);
+ tsFileReaders.add(new TsFileReader(tsFileSequenceReader));
+ }
+
+ deviceMeasurementsMapIterators = new ArrayList<>();
+ deviceIsAlignedMaps = new ArrayList<>();
+ for (int i = 0; i < tsFiles.size(); i++) {
+ deviceMeasurementsMapIterators.add(
+ filterDeviceMeasurementsMapByPattern(i).entrySet().iterator());
+ deviceIsAlignedMaps.add(readDeviceIsAlignedMap(i));
+ }
+ measurementDataTypeMaps = new ArrayList<>();
+ for (TsFileSequenceReader tsFileSequenceReader : tsFileSequenceReaders) {
+
measurementDataTypeMaps.add(tsFileSequenceReader.getFullPathDataTypeMap());
+ }
+ } catch (Exception e) {
+ close();
+ throw e;
+ }
+ }
+
+ private Map<String, List<String>> filterDeviceMeasurementsMapByPattern(int
i) throws IOException {
+ final Map<String, List<String>> filteredDeviceMeasurementsMap = new
HashMap<>();
+
+ for (Map.Entry<String, List<String>> entry :
+ tsFileSequenceReaders.get(i).getDeviceMeasurementsMap().entrySet()) {
+ final String deviceId = entry.getKey();
+
+ // case 1: for example, pattern is root.a.b or pattern is null and
device is root.a.b.c
+ // in this case, all data can be matched without checking the
measurements
+ if (pattern == null
+ || pattern.length() <= deviceId.length() &&
deviceId.startsWith(pattern)) {
+ if (!entry.getValue().isEmpty()) {
+ filteredDeviceMeasurementsMap.put(deviceId, entry.getValue());
+ }
+ }
+
+ // case 2: for example, pattern is root.a.b.c and device is root.a.b
+ // in this case, we need to check the full path
+ else if (pattern.length() > deviceId.length() &&
pattern.startsWith(deviceId)) {
+ final List<String> filteredMeasurements = new ArrayList<>();
+
+ for (final String measurement : entry.getValue()) {
+ // low cost check comes first
+ if (pattern.length() == deviceId.length() + measurement.length() + 1
+ // high cost check comes later
+ && pattern.endsWith(TsFileConstant.PATH_SEPARATOR +
measurement)) {
+ filteredMeasurements.add(measurement);
+ }
+ }
+
+ if (!filteredMeasurements.isEmpty()) {
+ filteredDeviceMeasurementsMap.put(deviceId, filteredMeasurements);
+ }
+ }
+ }
+
+ return filteredDeviceMeasurementsMap;
+ }
+
+ private Map<String, Boolean> readDeviceIsAlignedMap(int i) throws
IOException {
+ final Map<String, Boolean> deviceIsAlignedResultMap = new HashMap<>();
+ final TsFileDeviceIterator deviceIsAlignedIterator =
+ tsFileSequenceReaders.get(i).getAllDevicesIteratorWithIsAligned();
+ while (deviceIsAlignedIterator.hasNext()) {
+ final Pair<String, Boolean> deviceIsAlignedPair =
deviceIsAlignedIterator.next();
+ deviceIsAlignedResultMap.put(deviceIsAlignedPair.getLeft(),
deviceIsAlignedPair.getRight());
+ }
+ return deviceIsAlignedResultMap;
+ }
+
+ /**
+ * @return TabletInsertionEvent in a streaming way
+ */
+ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
+ return () ->
+ new Iterator<TabletInsertionEvent>() {
+
+ private TsFileInsertionDataTabletIterator tabletIterator = null;
+
+ @Override
+ public boolean hasNext() {
+ while (tabletIterator == null || !tabletIterator.hasNext()) {
+ if
(!deviceMeasurementsMapIterators.get(currFileIndex).hasNext()) {
+ if (currFileIndex >= tsFileReaders.size()) {
+ close();
+ return false;
+ } else {
+ currFileIndex++;
+ continue;
+ }
+ }
+
+ final Map.Entry<String, List<String>> entry =
deviceMeasurementsMapIterators.get(
+ currFileIndex).next();
+
+ try {
+ tabletIterator =
+ new TsFileInsertionDataTabletIterator(
+ tsFileReaders.get(currFileIndex),
+ measurementDataTypeMaps.get(currFileIndex),
+ entry.getKey(),
+ entry.getValue(),
+ timeFilterExpression);
+ } catch (IOException e) {
+ close();
+ throw new PipeException("failed to create
TsFileInsertionDataTabletIterator", e);
+ }
+ }
+
+ return true;
+ }
+
+ @Override
+ public TabletInsertionEvent next() {
+ if (!hasNext()) {
+ close();
+ throw new NoSuchElementException();
+ }
+
+ final Tablet tablet = tabletIterator.next();
+ final boolean isAligned = deviceIsAlignedMaps.get(currFileIndex)
+ .getOrDefault(tablet.deviceId, false);
+
+ final TabletInsertionEvent next;
+ if (!hasNext()) {
+ next =
+ new PipeRawTabletInsertionEvent(
+ tablet, isAligned, pipeTaskMeta, sourceEvent, true);
+ close();
+ } else {
+ next =
+ new PipeRawTabletInsertionEvent(
+ tablet, isAligned, pipeTaskMeta, sourceEvent, false);
+ }
+ return next;
+ }
+ };
+ }
+
+ @Override
+ public void close() {
+ for (TsFileReader tsFileReader : tsFileReaders) {
+ try {
+ if (tsFileReader != null) {
+ tsFileReader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close TsFileReader", e);
+ }
+ }
+
+ for (TsFileSequenceReader tsFileSequenceReader : tsFileSequenceReaders) {
+ try {
+ if (tsFileSequenceReader != null) {
+ tsFileSequenceReader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close TsFileSequenceReader", e);
+ }
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
new file mode 100644
index 00000000000..463c529fedb
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.pipe.extractor.historical;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.PipeBatchTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Similar to the base class, but it batches several files as an event to
enable further
+ * optimization during the latter transfer.
+ */
+public class BatchedTsFileExtractor extends
PipeHistoricalDataRegionTsFileExtractor {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(BatchedTsFileExtractor.class);
+ private int maxBatchSize;
+
+ public BatchedTsFileExtractor(int maxBatchSize) {
+ this.maxBatchSize = Math.max(1, maxBatchSize);
+ }
+
+ @Override
+ public synchronized Event supply() {
+ if (pendingQueue == null) {
+ return null;
+ }
+ TsFileResource resource = pendingQueue.poll();
+ if (resource == null) {
+ return null;
+ }
+ List<TsFileResource> tsFileResourceList = new ArrayList<>(maxBatchSize);
+ tsFileResourceList.add(resource);
+ while (!pendingQueue.isEmpty() && tsFileResourceList.size() <
maxBatchSize) {
+ tsFileResourceList.add(resource);
+ }
+
+ final PipeBatchTsFileInsertionEvent event = new
PipeBatchTsFileInsertionEvent(
+ tsFileResourceList,
+ false,
+ pipeTaskMeta,
+ pattern,
+ historicalDataExtractionStartTime,
+ historicalDataExtractionEndTime,
+ !(isTsFileResourceCoveredByTimeRange(tsFileResourceList.get(0))
+ && isTsFileResourceCoveredByTimeRange(
+ tsFileResourceList.get(tsFileResourceList.size() - 1))));
+
+ event.increaseReferenceCount(BatchedTsFileExtractor.class.getName());
+
+ for (TsFileResource res : tsFileResourceList) {
+ try {
+ PipeResourceManager.tsfile().unpinTsFileResource(res);
+ } catch (IOException e) {
+ LOGGER.warn(
+ "Pipe: failed to unpin TsFileResource after creating event,
original path: {}",
+ resource.getTsFilePath());
+ }
+ }
+
+ return event;
+ }
+
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
index 6094b9766a0..4227e5315c1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -62,19 +62,19 @@ public class PipeHistoricalDataRegionTsFileExtractor
implements PipeHistoricalDa
private static final Map<Integer, Long>
DATA_REGION_ID_TO_PIPE_FLUSHED_TIME_MAP = new HashMap<>();
private static final long PIPE_MIN_FLUSH_INTERVAL_IN_MS = 2000;
- private PipeTaskMeta pipeTaskMeta;
- private ProgressIndex startIndex;
+ protected PipeTaskMeta pipeTaskMeta;
+ protected ProgressIndex startIndex;
- private int dataRegionId;
+ protected int dataRegionId;
- private String pattern;
+ protected String pattern;
- private long historicalDataExtractionStartTime; // Event time
- private long historicalDataExtractionEndTime; // Event time
+ protected long historicalDataExtractionStartTime; // Event time
+ protected long historicalDataExtractionEndTime; // Event time
- private long historicalDataExtractionTimeLowerBound; // Arrival time
+ protected long historicalDataExtractionTimeLowerBound; // Arrival time
- private Queue<TsFileResource> pendingQueue;
+ protected Queue<TsFileResource> pendingQueue;
@Override
public void validate(PipeParameterValidator validator) {
@@ -249,7 +249,7 @@ public class PipeHistoricalDataRegionTsFileExtractor
implements PipeHistoricalDa
|| historicalDataExtractionEndTime < resource.getFileStartTime());
}
- private boolean isTsFileResourceCoveredByTimeRange(TsFileResource resource) {
+ protected boolean isTsFileResourceCoveredByTimeRange(TsFileResource
resource) {
return historicalDataExtractionStartTime <= resource.getFileStartTime()
&& historicalDataExtractionEndTime >= resource.getFileEndTime();
}
