tmp save
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/AlignedSeriesScanUtil.java similarity index 94% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/AlignedSeriesScanUtil.java index b5e7e24..7da01ab 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/AlignedSeriesScanUtil.java
@@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.db.queryengine.execution.operator.source; +package org.apache.iotdb.db.queryengine.execution.operator.scan; import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.PartialPath; @@ -84,31 +84,31 @@ @SuppressWarnings("squid:S3740") @Override - protected Statistics currentFileStatistics(int index) { + public Statistics currentFileStatistics(int index) { return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index); } @SuppressWarnings("squid:S3740") @Override - protected Statistics currentFileTimeStatistics() { + public Statistics currentFileTimeStatistics() { return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics(); } @SuppressWarnings("squid:S3740") @Override - protected Statistics currentChunkStatistics(int index) { + public Statistics currentChunkStatistics(int index) { return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index); } @SuppressWarnings("squid:S3740") @Override - protected Statistics currentChunkTimeStatistics() { + public Statistics currentChunkTimeStatistics() { return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics(); } @SuppressWarnings("squid:S3740") @Override - protected Statistics currentPageStatistics(int index) throws IOException { + public Statistics currentPageStatistics(int index) { if (firstPageReader == null) { return null; } @@ -117,7 +117,7 @@ @SuppressWarnings("squid:S3740") @Override - protected Statistics currentPageTimeStatistics() throws IOException { + public Statistics currentPageTimeStatistics() { if (firstPageReader == null) { return null; } @@ -147,7 +147,7 @@ } @Override - protected List<TSDataType> getTsDataTypeList() { + public List<TSDataType> getTsDataTypeList() { return dataTypes; }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/ChunkMetadataProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/ChunkMetadataProvider.java new file mode 100644 index 0000000..1b865bf --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/ChunkMetadataProvider.java
@@ -0,0 +1,158 @@ +/* + * 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.queryengine.execution.operator.scan; + +import org.apache.iotdb.db.utils.FileLoaderUtils; +import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata; +import org.apache.iotdb.tsfile.file.metadata.ITimeSeriesMetadata; +import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics; +import org.apache.iotdb.tsfile.read.filter.basic.Filter; + +import java.io.IOException; +import java.util.List; +import java.util.PriorityQueue; + +public class ChunkMetadataProvider { + + private final TimeSeriesMetadataProvider sourceProvider; + + private IChunkMetadata firstChunkMetadata; + private final PriorityQueue<IChunkMetadata> cachedChunkMetadata; + + private final SeriesScanUtil.TimeOrderUtils orderUtils; + + public ChunkMetadataProvider( + TimeSeriesMetadataProvider sourceProvider, SeriesScanUtil.TimeOrderUtils orderUtils) { + this.sourceProvider = sourceProvider; + this.cachedChunkMetadata = + new PriorityQueue<>( + orderUtils.comparingLong( + chunkMetadata -> orderUtils.getOrderTime(chunkMetadata.getStatistics()))); + this.orderUtils = orderUtils; + } + + public boolean isCachedChunkMetadataConsumed() { + return firstChunkMetadata == null && cachedChunkMetadata.isEmpty(); + } + + public boolean hasCurrentChunk() { + return firstChunkMetadata != null; + } + + public boolean hasNext() { + return !cachedChunkMetadata.isEmpty() + && (sourceProvider.hasCurrentFile() || sourceProvider.hasNext()); + } + + public void filterFirstChunkMetadata() throws IOException { + if (firstChunkMetadata != null && !isChunkOverlapped() && !firstChunkMetadata.isModified()) { + Filter queryFilter = scanOptions.getPushDownFilter(); + Statistics statistics = firstChunkMetadata.getStatistics(); + if (queryFilter == null || queryFilter.allSatisfy(statistics)) { + long rowCount = statistics.getCount(); + if (paginationController.hasCurOffset(rowCount)) { + skipCurrentChunk(); + paginationController.consumeOffset(rowCount); + } + } else if (!queryFilter.satisfy(statistics)) { + skipCurrentChunk(); + } + } + } + + public void skipCurrentChunk() { + firstChunkMetadata = null; + } + + public boolean currentChunkModified() { + return firstChunkMetadata.isModified(); + } + + /** construct first chunk metadata */ + public void initFirstChunkMetadata() throws IOException { + if (sourceProvider.hasCurrentFile()) { + /* + * try to unpack all overlapped TimeSeriesMetadata to cachedChunkMetadata + */ + sourceProvider.unpackAllOverlappedTsFilesToTimeSeriesMetadata( + orderUtils.getOverlapCheckTime( + timeSeriesMetadataProvider.getCurrentFile().getStatistics())); + unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( + orderUtils.getOverlapCheckTime( + timeSeriesMetadataProvider.getCurrentFile().getStatistics()), + true); + } else { + /* + * first time series metadata is already unpacked, consume cached ChunkMetadata + */ + while (!cachedChunkMetadata.isEmpty()) { + firstChunkMetadata = cachedChunkMetadata.peek(); + sourceProvider.unpackAllOverlappedTsFilesToTimeSeriesMetadata( + orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics())); + unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( + orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), false); + if (firstChunkMetadata.equals(cachedChunkMetadata.peek())) { + firstChunkMetadata = cachedChunkMetadata.poll(); + break; + } + } + } + } + + public void unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( + long endpointTime, boolean init) { + while (!seqTimeSeriesMetadata.isEmpty() + && orderUtils.isOverlapped(endpointTime, seqTimeSeriesMetadata.get(0).getStatistics())) { + unpackOneTimeSeriesMetadata(seqTimeSeriesMetadata.remove(0)); + } + while (!unSeqTimeSeriesMetadata.isEmpty() + && orderUtils.isOverlapped(endpointTime, unSeqTimeSeriesMetadata.peek().getStatistics())) { + unpackOneTimeSeriesMetadata(unSeqTimeSeriesMetadata.poll()); + } + + if (sourceProvider.hasCurrentFile() + && orderUtils.isOverlapped(endpointTime, sourceProvider.getCurrentFile().getStatistics())) { + unpackOneTimeSeriesMetadata(sourceProvider.getCurrentFile()); + sourceProvider.skipCurrentFile(); + } + + if (init && firstChunkMetadata == null && !cachedChunkMetadata.isEmpty()) { + firstChunkMetadata = cachedChunkMetadata.poll(); + } + } + + private void unpackOneTimeSeriesMetadata(ITimeSeriesMetadata timeSeriesMetadata) { + List<IChunkMetadata> chunkMetadataList = + FileLoaderUtils.loadChunkMetadataList(timeSeriesMetadata); + chunkMetadataList.forEach(chunkMetadata -> chunkMetadata.setSeq(timeSeriesMetadata.isSeq())); + + cachedChunkMetadata.addAll(chunkMetadataList); + } + + public boolean currentChunkOverlapped() { + Statistics chunkStatistics = firstChunkMetadata.getStatistics(); + return !cachedChunkMetadata.isEmpty() + && orderUtils.isOverlapped(chunkStatistics, cachedChunkMetadata.peek().getStatistics()); + } + + public IChunkMetadata getCurrentChunk() { + return firstChunkMetadata; + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/PageReaderProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/PageReaderProvider.java new file mode 100644 index 0000000..48c598b --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/PageReaderProvider.java
@@ -0,0 +1,554 @@ +/* + * 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.queryengine.execution.operator.scan; + +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.DescPriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; +import org.apache.iotdb.db.utils.FileLoaderUtils; +import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; +import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.read.common.block.TsBlock; +import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder; +import org.apache.iotdb.tsfile.read.filter.basic.Filter; +import org.apache.iotdb.tsfile.read.reader.IPageReader; +import org.apache.iotdb.tsfile.utils.TsPrimitiveType; + +import java.io.IOException; +import java.util.LinkedList; +import java.util.List; +import java.util.PriorityQueue; + +import static com.google.common.base.Preconditions.checkState; +import static org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil.SERIES_SCAN_COST_METRIC_SET; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_MERGE_READER_ALIGNED; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_MERGE_READER_NONALIGNED; + +public class PageReaderProvider { + + // page cache + private SeriesScanUtil.VersionPageReader firstPageReader; + private final List<SeriesScanUtil.VersionPageReader> seqPageReaders; + private final PriorityQueue<SeriesScanUtil.VersionPageReader> unSeqPageReaders; + + // point cache + private final PriorityMergeReader mergeReader; + + // result cache + private boolean hasCachedNextOverlappedPage; + private TsBlock cachedTsBlock; + + private ChunkMetadataProvider sourceProvider; + private SeriesScanUtil.TimeOrderUtils orderUtils; + + public PageReaderProvider( + ChunkMetadataProvider sourceProvider, SeriesScanUtil.TimeOrderUtils orderUtils) { + this.sourceProvider = sourceProvider; + this.seqPageReaders = new LinkedList<>(); + this.unSeqPageReaders = + new PriorityQueue<>( + orderUtils.comparingLong( + versionPageReader -> orderUtils.getOrderTime(versionPageReader.getStatistics()))); + + if (orderUtils.getAscending()) { + this.mergeReader = new PriorityMergeReader(); + } else { + this.mergeReader = new DescPriorityMergeReader(); + } + + this.orderUtils = orderUtils; + } + + public boolean isCachedPageReaderConsumed() { + return unSeqPageReaders.isEmpty() + && firstPageReader == null + && !mergeReader.hasNextTimeValuePair(); + } + + public boolean hasCurrentPage() { + return firstPageReader != null; + } + + public boolean hasNext() { + return sourceProvider.hasCurrentChunk() + || !seqPageReaders.isEmpty() + || !unSeqPageReaders.isEmpty(); + } + + public boolean currentPageOverlapped() throws IOException { + // has an overlapped page + if (hasCachedNextOverlappedPage) { + return true; + } + + // has a non-overlapped page in firstPageReader + checkState(!firstPageOverlappedWithMergeReader(), "overlapped data should be consumed first"); + + Statistics firstPageStatistics = firstPageReader.getStatistics(); + return !unSeqPageReaders.isEmpty() + && orderUtils.isOverlapped(firstPageStatistics, unSeqPageReaders.peek().getStatistics()); + } + + public void peekCurrentPage() throws IOException { + if (sourceProvider.hasCurrentChunk()) { + // try to unpack all overlapped ChunkMetadata to cachedPageReaders + unpackAllOverlappedChunkMetadataToPageReaders( + orderUtils.getOverlapCheckTime(sourceProvider.getCurrentChunk().getStatistics()), true); + } else { + // first chunk metadata is already unpacked, consume cached pages + initFirstPageReader(); + } + } + + private void unpackAllOverlappedChunkMetadataToPageReaders(long endpointTime, boolean init) + throws IOException { + if (sourceProvider.hasCurrentChunk() + && orderUtils.isOverlapped(endpointTime, firstChunkMetadata.getStatistics())) { + unpackOneChunkMetaData(firstChunkMetadata); + sourceProvider.skipCurrentChunk(); + } + // In case unpacking too many sequence chunks + boolean hasMeetSeq = false; + while (!cachedChunkMetadata.isEmpty() + && orderUtils.isOverlapped(endpointTime, cachedChunkMetadata.peek().getStatistics())) { + if (cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { + break; + } else if (cachedChunkMetadata.peek().isSeq()) { + hasMeetSeq = true; + } + unpackOneChunkMetaData(cachedChunkMetadata.poll()); + } + if (init + && firstPageReader == null + && (!seqPageReaders.isEmpty() || !unSeqPageReaders.isEmpty())) { + initFirstPageReader(); + } + } + + private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOException { + List<IPageReader> pageReaderList = + FileLoaderUtils.loadPageReaderList(chunkMetaData, getGlobalTimeFilter()); + + // init TsBlockBuilder for each page reader + pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); + + if (chunkMetaData.isSeq()) { + if (orderUtils.getAscending()) { + for (IPageReader iPageReader : pageReaderList) { + seqPageReaders.add( + new SeriesScanUtil.VersionPageReader( + chunkMetaData.getVersion(), + chunkMetaData.getOffsetOfChunkHeader(), + iPageReader, + true)); + } + } else { + for (int i = pageReaderList.size() - 1; i >= 0; i--) { + seqPageReaders.add( + new SeriesScanUtil.VersionPageReader( + chunkMetaData.getVersion(), + chunkMetaData.getOffsetOfChunkHeader(), + pageReaderList.get(i), + true)); + } + } + } else { + pageReaderList.forEach( + pageReader -> + unSeqPageReaders.add( + new SeriesScanUtil.VersionPageReader( + chunkMetaData.getVersion(), + chunkMetaData.getOffsetOfChunkHeader(), + pageReader, + false))); + } + } + + private void initFirstPageReader() throws IOException { + while (firstPageReader == null) { + SeriesScanUtil.VersionPageReader candidatePageReader = + getCandidateFirstPageReaderFromCachedReaders(); + if (candidatePageReader == null) { + break; + } + + // unpack overlapped page using current page reader + long overlapCheckTime = orderUtils.getOverlapCheckTime(candidatePageReader.getStatistics()); + unpackAllOverlappedTsFilesToTimeSeriesMetadata(overlapCheckTime); + unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(overlapCheckTime); + unpackAllOverlappedChunkMetadataToPageReaders(overlapCheckTime, false); + + // this page after unpacking must be the first page + if (candidatePageReader.equals(getCandidateFirstPageReaderFromCachedReaders())) { + // set first page reader + firstPageReader = candidatePageReader; + + // update seq and unseq page readers + if (!seqPageReaders.isEmpty() && candidatePageReader.equals(seqPageReaders.get(0))) { + seqPageReaders.remove(0); + } else if (!unSeqPageReaders.isEmpty() + && candidatePageReader.equals(unSeqPageReaders.peek())) { + unSeqPageReaders.poll(); + } + } + } + } + + // We use get() and peek() here in case it's not the first page reader before unpacking + private SeriesScanUtil.VersionPageReader getCandidateFirstPageReaderFromCachedReaders() { + SeriesScanUtil.VersionPageReader candidatePageReader = null; + if (!seqPageReaders.isEmpty() && !unSeqPageReaders.isEmpty()) { + if (orderUtils.isTakeSeqAsFirst( + seqPageReaders.get(0).getStatistics(), unSeqPageReaders.peek().getStatistics())) { + candidatePageReader = seqPageReaders.get(0); + } else { + candidatePageReader = unSeqPageReaders.peek(); + } + } else if (!seqPageReaders.isEmpty()) { + candidatePageReader = seqPageReaders.get(0); + } else if (!unSeqPageReaders.isEmpty()) { + candidatePageReader = unSeqPageReaders.peek(); + } + return candidatePageReader; + } + + public boolean isExistOverlappedPage() throws IOException { + if (firstPageOverlapped() && hasNextOverlappedPage()) { + // next page is overlapped, read overlapped data and cache it + cachedTsBlock = nextOverlappedPage(); + if (cachedTsBlock != null && !cachedTsBlock.isEmpty()) { + hasCachedNextOverlappedPage = true; + return true; + } + } + return false; + } + + private boolean firstPageOverlapped() throws IOException { + if (firstPageReader == null) { + return false; + } + + long endpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); + unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime); + unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(endpointTime); + unpackAllOverlappedChunkMetadataToPageReaders(endpointTime, false); + + return (!seqPageReaders.isEmpty() + && orderUtils.isOverlapped( + firstPageReader.getStatistics(), seqPageReaders.get(0).getStatistics())) + || (!unSeqPageReaders.isEmpty() + && orderUtils.isOverlapped( + firstPageReader.getStatistics(), unSeqPageReaders.peek().getStatistics()) + || (mergeReader.hasNextTimeValuePair() + && orderUtils.isOverlapped( + mergeReader.currentTimeValuePair().getTimestamp(), + firstPageReader.getStatistics()))); + } + + public boolean hasCachedNextOverlappedPage() { + return hasCachedNextOverlappedPage; + } + + private boolean firstPageOverlappedWithMergeReader() throws IOException { + return mergeReader.hasNextTimeValuePair() + && ((orderUtils.getAscending() + && mergeReader.currentTimeValuePair().getTimestamp() + <= firstPageReader.getStatistics().getEndTime()) + || (!orderUtils.getAscending() + && mergeReader.currentTimeValuePair().getTimestamp() + >= firstPageReader.getStatistics().getStartTime())); + } + + /** + * read overlapped data till currentLargestEndTime in mergeReader, if current batch does not + * contain data, read till next currentLargestEndTime again + */ + private boolean hasNextOverlappedPage() throws IOException { + long startTime = System.nanoTime(); + try { + if (hasCachedNextOverlappedPage) { + return true; + } + + tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader(); + + while (true) { + if (!mergeReader.hasNextTimeValuePair()) { + return false; + } + + // may has overlapped data + if (processOverlappedRecord()) { + return true; + } + } + } finally { + recordMergeReaderCost(System.nanoTime() - startTime); + } + } + + private long currentPageEndPointTime; + + private boolean processOverlappedRecord() throws IOException { + TsBlockBuilder builder = new TsBlockBuilder(getTsDataTypeList()); + currentPageEndPointTime = mergeReader.getCurrentReadStopTime(); + + while (mergeReader.hasNextTimeValuePair()) { + + // get current first point in mergeReader, this maybe overlapped later + TimeValuePair timeValuePair = mergeReader.currentTimeValuePair(); + + if (isExcessEndpoint(timeValuePair, builder)) { + return endMergeReaderScan(builder); + } + + // unpack all overlapped data for the first timeValuePair + unpackAllOverlappedData(timeValuePair); + + // update if there are unpacked unSeqPageReaders + timeValuePair = mergeReader.currentTimeValuePair(); + + // from now, the unsequence reader is all unpacked, so we don't need to consider it + + // if current timeValuePair excesses the first page reader's end time, we just use the + // cached data + if (isTimeValuePairBeyondFirstPageReader(timeValuePair)) { + return endMergeReaderScan(builder); + } + if (isTimeValuePairOverlappingWithFirstPageReader(timeValuePair)) { + // current timeValuePair is overlapped with firstPageReader, add it to merged reader + // and update endTime to the max end time + updateMergeReaderByPageReader(firstPageReader); + firstPageReader = null; + } + + timeValuePair = mergeReader.currentTimeValuePair(); + + if (isTimeValuePairBeyondSeqPageReader(timeValuePair)) { + return endMergeReaderScan(builder); + } + if (isTimeValuePairOverlappingWithFirstSeqPageReader(timeValuePair)) { + updateMergeReaderByPageReader(seqPageReaders.remove(0)); + } + + // get the latest first point in mergeReader + timeValuePair = mergeReader.nextTimeValuePair(); + if (processPaginationAndFilter(timeValuePair, builder)) { + return endMergeReaderScan(builder); + } + } + + return endMergeReaderScan(builder); + } + + private boolean isExcessEndpoint(TimeValuePair timeValuePair, TsBlockBuilder builder) { + // when the merged point excesses the currentPageEndPointTime, we have read all overlapped + // data before currentPageEndPointTime + // 1. has cached batch data, we don't need to read more data, just use the cached data later + // 2. has first page reader, which means first page reader last endTime < + // currentTimeValuePair.getTimestamp(), + // we could just use the first page reader later + // 3. sequence page reader is not empty, which means first page reader last endTime < + // currentTimeValuePair.getTimestamp(), + // we could use the first sequence page reader later + return orderUtils.isExcessEndpoint(timeValuePair.getTimestamp(), currentPageEndPointTime) + && (!builder.isEmpty() || firstPageReader != null || !seqPageReaders.isEmpty()); + } + + private boolean processPaginationAndFilter(TimeValuePair timeValuePair, TsBlockBuilder builder) { + if (!timeValuePairSatisfyFilter(timeValuePair)) { + return false; + } + if (paginationController.hasCurOffset()) { + paginationController.consumeOffset(); + return false; + } + if (paginationController.hasCurLimit()) { + addTimeValuePairToResult(timeValuePair, builder); + paginationController.consumeLimit(); + return false; + } else { + return true; + } + } + + private void unpackAllOverlappedData(TimeValuePair timeValuePair) throws IOException { + unpackAllOverlappedTsFilesToTimeSeriesMetadata(timeValuePair.getTimestamp()); + unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(timeValuePair.getTimestamp()); + unpackAllOverlappedChunkMetadataToPageReaders(timeValuePair.getTimestamp(), false); + unpackAllOverlappedUnseqPageReadersToMergeReader(timeValuePair.getTimestamp()); + } + + private boolean endMergeReaderScan(TsBlockBuilder builder) { + // if current overlapped page has valid data, return, otherwise read next overlapped page + hasCachedNextOverlappedPage = !builder.isEmpty(); + cachedTsBlock = builder.build(); + return hasCachedNextOverlappedPage; + } + + private boolean isTimeValuePairBeyondFirstPageReader(TimeValuePair timeValuePair) { + if (firstPageReader == null) { + return false; + } + return (orderUtils.getAscending() + && timeValuePair.getTimestamp() > firstPageReader.getStatistics().getEndTime()) + || (!orderUtils.getAscending() + && timeValuePair.getTimestamp() < firstPageReader.getStatistics().getStartTime()); + } + + private boolean isTimeValuePairOverlappingWithFirstPageReader(TimeValuePair timeValuePair) { + if (firstPageReader == null) { + return false; + } + return orderUtils.isOverlapped(timeValuePair.getTimestamp(), firstPageReader.getStatistics()); + } + + private boolean isTimeValuePairBeyondSeqPageReader(TimeValuePair timeValuePair) { + if (seqPageReaders.isEmpty()) { + return false; + } + return (orderUtils.getAscending() + && timeValuePair.getTimestamp() > seqPageReaders.get(0).getStatistics().getEndTime()) + || (!orderUtils.getAscending() + && timeValuePair.getTimestamp() < seqPageReaders.get(0).getStatistics().getStartTime()); + } + + private boolean isTimeValuePairOverlappingWithFirstSeqPageReader(TimeValuePair timeValuePair) { + if (seqPageReaders.isEmpty()) { + return false; + } + return orderUtils.isOverlapped( + timeValuePair.getTimestamp(), seqPageReaders.get(0).getStatistics()); + } + + private void updateMergeReaderByPageReader(SeriesScanUtil.VersionPageReader pageReader) + throws IOException { + mergeReader.addReader( + getPointReader(pageReader.getAllSatisfiedPageData(orderUtils.getAscending())), + pageReader.version, + orderUtils.getOverlapCheckTime(pageReader.getStatistics())); + currentPageEndPointTime = updateEndPointTime(currentPageEndPointTime, pageReader); + } + + private boolean timeValuePairSatisfyFilter(TimeValuePair timeValuePair) { + Filter queryFilter = scanOptions.getPushDownFilter(); + if (queryFilter != null) { + Object valueForFilter = timeValuePair.getValue().getValue(); + + if (timeValuePair.getValue().getDataType() == TSDataType.VECTOR) { + for (TsPrimitiveType tsPrimitiveType : timeValuePair.getValue().getVector()) { + if (tsPrimitiveType != null) { + valueForFilter = tsPrimitiveType.getValue(); + break; + } + } + } + + return queryFilter.satisfy(timeValuePair.getTimestamp(), valueForFilter); + } + return true; + } + + private void addTimeValuePairToResult(TimeValuePair timeValuePair, TsBlockBuilder builder) { + builder.getTimeColumnBuilder().writeLong(timeValuePair.getTimestamp()); + switch (dataType) { + case BOOLEAN: + builder.getColumnBuilder(0).writeBoolean(timeValuePair.getValue().getBoolean()); + break; + case INT32: + builder.getColumnBuilder(0).writeInt(timeValuePair.getValue().getInt()); + break; + case INT64: + builder.getColumnBuilder(0).writeLong(timeValuePair.getValue().getLong()); + break; + case FLOAT: + builder.getColumnBuilder(0).writeFloat(timeValuePair.getValue().getFloat()); + break; + case DOUBLE: + builder.getColumnBuilder(0).writeDouble(timeValuePair.getValue().getDouble()); + break; + case TEXT: + builder.getColumnBuilder(0).writeBinary(timeValuePair.getValue().getBinary()); + break; + case VECTOR: + TsPrimitiveType[] values = timeValuePair.getValue().getVector(); + for (int i = 0; i < values.length; i++) { + if (values[i] == null) { + builder.getColumnBuilder(i).appendNull(); + } else { + builder.getColumnBuilder(i).writeTsPrimitiveType(values[i]); + } + } + break; + default: + throw new UnSupportedDataTypeException(String.valueOf(dataType)); + } + builder.declarePosition(); + } + + private void recordMergeReaderCost(long duration) { + SERIES_SCAN_COST_METRIC_SET.recordSeriesScanCost( + isAligned + ? BUILD_TSBLOCK_FROM_MERGE_READER_ALIGNED + : BUILD_TSBLOCK_FROM_MERGE_READER_NONALIGNED, + duration); + } + + private long updateEndPointTime( + long currentPageEndPointTime, SeriesScanUtil.VersionPageReader pageReader) { + if (orderUtils.getAscending()) { + return Math.min(currentPageEndPointTime, pageReader.getStatistics().getEndTime()); + } else { + return Math.max(currentPageEndPointTime, pageReader.getStatistics().getStartTime()); + } + } + + private void tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader() throws IOException { + + /* + * no cached page readers + */ + if (firstPageReader == null && unSeqPageReaders.isEmpty() && seqPageReaders.isEmpty()) { + return; + } + + /* + * init firstPageReader + */ + if (firstPageReader == null) { + initFirstPageReader(); + } + + long currentPageEndpointTime; + if (mergeReader.hasNextTimeValuePair()) { + currentPageEndpointTime = mergeReader.getCurrentReadStopTime(); + } else { + currentPageEndpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); + } + + /* + * put all currently directly overlapped unseq page reader to merge reader + */ + unpackAllOverlappedUnseqPageReadersToMergeReader(currentPageEndpointTime); + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/SeriesScanUtil.java new file mode 100644 index 0000000..a016b2b --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/SeriesScanUtil.java
@@ -0,0 +1,806 @@ +/* + * 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.queryengine.execution.operator.scan; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet; +import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; +import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemAlignedPageReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemPageReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics; +import org.apache.iotdb.tsfile.read.common.block.TsBlock; +import org.apache.iotdb.tsfile.read.filter.basic.Filter; +import org.apache.iotdb.tsfile.read.reader.IAlignedPageReader; +import org.apache.iotdb.tsfile.read.reader.IPageReader; +import org.apache.iotdb.tsfile.read.reader.IPointReader; +import org.apache.iotdb.tsfile.read.reader.series.PaginationController; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.function.ToLongFunction; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkState; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_DISK; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_MEM; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_DISK; +import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_MEM; + +public class SeriesScanUtil { + + private static final String ONE_SENSOR_ERROR_MSG = + "Only one sensor in non-aligned SeriesScanUtil."; + + // The path of the target series which will be scanned. + private final PartialPath seriesPath; + protected boolean isAligned = false; + protected final TSDataType dataType; + + // inner class of SeriesReader for order purpose + private final TimeOrderUtils orderUtils; + + private QueryDataSource dataSource; + + // file index + protected int curSeqFileIndex; + protected int curUnseqFileIndex; + + // TimeSeriesMetadata cache + protected TimeSeriesMetadataProvider timeSeriesMetadataProvider; + + // chunk cache + protected ChunkMetadataProvider chunkMetadataProvider; + + // page cache + protected PageReaderProvider pageReaderProvider; + + protected SeriesScanOptions scanOptions; + protected PaginationController paginationController; + + private static final SeriesScanCostMetricSet SERIES_SCAN_COST_METRIC_SET = + SeriesScanCostMetricSet.getInstance(); + + public SeriesScanUtil(PartialPath seriesPath, Ordering scanOrder, SeriesScanOptions scanOptions) { + this.seriesPath = seriesPath; + dataType = seriesPath.getSeriesType(); + + this.scanOptions = scanOptions; + paginationController = scanOptions.getPaginationController(); + + if (scanOrder.isAscending()) { + orderUtils = new AscTimeOrderUtils(); + } else { + orderUtils = new DescTimeOrderUtils(); + } + + // init TimeSeriesMetadata materializer + timeSeriesMetadataProvider = new TimeSeriesMetadataProvider(orderUtils); + + // init ChunkMetadata materializer + chunkMetadataProvider = new ChunkMetadataProvider(timeSeriesMetadataProvider, orderUtils); + + // init PageReader materializer + pageReaderProvider = new PageReaderProvider(chunkMetadataProvider, orderUtils); + } + + public void initQueryDataSource(QueryDataSource dataSource) { + dataSource.fillOrderIndexes(seriesPath.getDevice(), orderUtils.getAscending()); + this.dataSource = dataSource; + + // updated filter concerning TTL + scanOptions.setTTL(dataSource.getDataTTL()); + + // init file index + orderUtils.setCurSeqFileIndex(dataSource); + curUnseqFileIndex = 0; + } + + public boolean hasNextFile() throws IOException { + if (!paginationController.hasCurLimit()) { + return false; + } + + checkState( + pageReaderProvider.isCachedPageReaderConsumed(), + "all cached pages should be consumed first"); + checkState( + chunkMetadataProvider.isCachedChunkMetadataConsumed(), + "all cached chunks should be consumed first"); + + if (timeSeriesMetadataProvider.hasCurrentFile()) { + return true; + } + + while (!timeSeriesMetadataProvider.hasCurrentFile() && timeSeriesMetadataProvider.hasNext()) { + // init first time series metadata whose startTime is minimum + timeSeriesMetadataProvider.tryToUnpackAllOverlappedFilesToTimeSeriesMetadata(); + // filter file based on push-down conditions + timeSeriesMetadataProvider.filterFirstTimeSeriesMetadata( + scanOptions.getPushDownFilter(), paginationController); + } + + return timeSeriesMetadataProvider.hasCurrentFile(); + } + + @SuppressWarnings("squid:S3740") + public boolean isFileOverlapped() { + return timeSeriesMetadataProvider.currentFileOverlapped(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentFileStatistics() { + return timeSeriesMetadataProvider.getCurrentFile().getStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentFileStatistics(int index) { + checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); + return currentFileStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentFileTimeStatistics() { + return currentFileStatistics(); + } + + public boolean currentFileModified() { + return timeSeriesMetadataProvider.currentFileModified(); + } + + public void skipCurrentFile() { + timeSeriesMetadataProvider.skipCurrentFile(); + } + + /** + * This method should be called after hasNextFile() until no next chunk, make sure that all + * overlapped chunks are consumed + */ + public boolean hasNextChunk() throws IOException { + if (!paginationController.hasCurLimit()) { + return false; + } + + checkState( + pageReaderProvider.isCachedPageReaderConsumed(), + "all cached pages should be consumed first"); + + if (chunkMetadataProvider.hasCurrentChunk()) { + return true; + } + + if (!timeSeriesMetadataProvider.hasCurrentFile() + && chunkMetadataProvider.isCachedChunkMetadataConsumed()) { + return false; + } + + while (!chunkMetadataProvider.hasCurrentChunk() && chunkMetadataProvider.hasNext()) { + chunkMetadataProvider.initFirstChunkMetadata(); + // filter chunk based on push-down conditions + chunkMetadataProvider.filterFirstChunkMetadata(); + } + return chunkMetadataProvider.hasCurrentChunk(); + } + + @SuppressWarnings("squid:S3740") + public boolean isChunkOverlapped() throws IOException { + if (!chunkMetadataProvider.hasCurrentChunk()) { + throw new IOException("no first chunk"); + } + return chunkMetadataProvider.currentChunkOverlapped(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentChunkStatistics() { + return chunkMetadataProvider.getCurrentChunk().getStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentChunkStatistics(int index) { + checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); + return currentChunkStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentChunkTimeStatistics() { + return currentChunkStatistics(); + } + + public boolean currentChunkModified() throws IOException { + if (!chunkMetadataProvider.hasCurrentChunk()) { + throw new IOException("no first chunk"); + } + return chunkMetadataProvider.currentChunkModified(); + } + + public void skipCurrentChunk() { + chunkMetadataProvider.skipCurrentChunk(); + } + + /** + * This method should be called after hasNextChunk() until no next page, make sure that all + * overlapped pages are consumed + */ + public boolean hasNextPage() throws IOException { + if (!paginationController.hasCurLimit()) { + return false; + } + + // has overlapped data before + if (pageReaderProvider.hasCachedNextOverlappedPage()) { + return true; + } + + if ((mergeReader.hasNextTimeValuePair() || firstPageOverlapped()) && hasNextOverlappedPage()) { + cachedTsBlock = nextOverlappedPage(); + if (cachedTsBlock != null && !cachedTsBlock.isEmpty()) { + hasCachedNextOverlappedPage = true; + return true; + } + } + + if (pageReaderProvider.hasCurrentPage()) { + return true; + } + + while (!pageReaderProvider.hasCurrentPage() && pageReaderProvider.hasNext()) { + pageReaderProvider.peekCurrentPage(); + if (pageReaderProvider.isExistOverlappedPage()) { + return true; + } + } + return pageReaderProvider.hasCurrentPage(); + } + + /** + * This method should be called after calling hasNextPage. + * + * <p>hasNextPage may cache firstPageReader if it is not overlapped or cached a BatchData if the + * first page is overlapped + */ + public boolean currentPageOverlapped() throws IOException { + return pageReaderProvider.currentPageOverlapped(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentPageStatistics() { + if (firstPageReader == null) { + return null; + } + return firstPageReader.getStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentPageStatistics(int index) { + checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); + return currentPageStatistics(); + } + + @SuppressWarnings("squid:S3740") + public Statistics currentPageTimeStatistics() { + return currentPageStatistics(); + } + + public boolean currentPageModified() throws IOException { + if (firstPageReader == null) { + throw new IOException("no first page"); + } + return firstPageReader.isModified(); + } + + public void skipCurrentPage() { + firstPageReader = null; + } + + /** This method should only be used when the method currentPageOverlapped() return true. */ + public TsBlock nextPage() throws IOException { + + if (hasCachedNextOverlappedPage) { + hasCachedNextOverlappedPage = false; + TsBlock res = cachedTsBlock; + cachedTsBlock = null; + return res; + } else { + // next page is not overlapped, push down value filter & limit offset + Filter queryFilter = scanOptions.getPushDownFilter(); + if (queryFilter != null) { + firstPageReader.setFilter(queryFilter); + } + TsBlock tsBlock; + if (orderUtils.getAscending()) { + firstPageReader.setLimitOffset(paginationController); + tsBlock = firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending()); + } else { + tsBlock = + paginationController.applyTsBlock( + firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending())); + } + + firstPageReader = null; + + return tsBlock; + } + } + + private void unpackAllOverlappedUnseqPageReadersToMergeReader(long endpointTime) + throws IOException { + while (!unSeqPageReaders.isEmpty() + && orderUtils.isOverlapped(endpointTime, unSeqPageReaders.peek().data.getStatistics())) { + putPageReaderToMergeReader(unSeqPageReaders.poll()); + } + if (firstPageReader != null + && !firstPageReader.isSeq() + && orderUtils.isOverlapped(endpointTime, firstPageReader.getStatistics())) { + putPageReaderToMergeReader(firstPageReader); + firstPageReader = null; + } + } + + private void putPageReaderToMergeReader(VersionPageReader pageReader) throws IOException { + mergeReader.addReader( + getPointReader(pageReader.getAllSatisfiedPageData(orderUtils.getAscending())), + pageReader.version, + orderUtils.getOverlapCheckTime(pageReader.getStatistics())); + } + + private TsBlock nextOverlappedPage() throws IOException { + if (hasCachedNextOverlappedPage || hasNextOverlappedPage()) { + hasCachedNextOverlappedPage = false; + return cachedTsBlock; + } + throw new IOException("No more batch data"); + } + + protected void unpackAllOverlappedTsFilesToTimeSeriesMetadata(long endpointTime) + throws IOException { + timeSeriesMetadataProvider.unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime); + } + + protected void unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(long endpointTime) { + chunkMetadataProvider.unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( + endpointTime, false); + } + + public List<TSDataType> getTsDataTypeList() { + return Collections.singletonList(dataType); + } + + protected IPointReader getPointReader(TsBlock tsBlock) { + return tsBlock.getTsBlockSingleColumnIterator(); + } + + public Filter getGlobalTimeFilter() { + return scanOptions.getGlobalTimeFilter(); + } + + @SuppressWarnings({"squid:S3740"}) + public boolean canUseCurrentFileStatistics() { + Statistics fileStatistics = currentFileTimeStatistics(); + return !isFileOverlapped() + && fileStatistics.containedByTimeFilter(getGlobalTimeFilter()) + && !currentFileModified(); + } + + @SuppressWarnings({"squid:S3740"}) + public boolean canUseCurrentChunkStatistics() throws IOException { + Statistics chunkStatistics = currentChunkTimeStatistics(); + return !isChunkOverlapped() + && chunkStatistics.containedByTimeFilter(getGlobalTimeFilter()) + && !currentChunkModified(); + } + + @SuppressWarnings({"squid:S3740"}) + public boolean canUseCurrentPageStatistics() throws IOException { + Statistics currentPageStatistics = currentPageTimeStatistics(); + if (currentPageStatistics == null) { + return false; + } + return !currentPageOverlapped() + && currentPageStatistics.containedByTimeFilter(getGlobalTimeFilter()) + && !currentPageModified(); + } + + protected static class VersionPageReader { + + private final PriorityMergeReader.MergeReaderPriority version; + private final IPageReader data; + + private final boolean isSeq; + private final boolean isAligned; + private final boolean isMem; + + VersionPageReader(long version, long offset, IPageReader data, boolean isSeq) { + this.version = new PriorityMergeReader.MergeReaderPriority(version, offset); + this.data = data; + this.isSeq = isSeq; + this.isAligned = data instanceof IAlignedPageReader; + this.isMem = data instanceof MemPageReader || data instanceof MemAlignedPageReader; + } + + @SuppressWarnings("squid:S3740") + Statistics getStatistics() { + return data.getStatistics(); + } + + @SuppressWarnings("squid:S3740") + Statistics getStatistics(int index) { + if (!(data instanceof IAlignedPageReader)) { + throw new IllegalArgumentException( + "Can only get statistics by index from AlignedPageReader"); + } + return ((IAlignedPageReader) data).getStatistics(index); + } + + @SuppressWarnings("squid:S3740") + Statistics getTimeStatistics() { + if (!(data instanceof IAlignedPageReader)) { + throw new IllegalArgumentException( + "Can only get statistics of time column from AlignedPageReader"); + } + return ((IAlignedPageReader) data).getTimeStatistics(); + } + + TsBlock getAllSatisfiedPageData(boolean ascending) throws IOException { + long startTime = System.nanoTime(); + try { + TsBlock tsBlock = data.getAllSatisfiedData(); + if (!ascending) { + tsBlock.reverse(); + } + return tsBlock; + } finally { + SERIES_SCAN_COST_METRIC_SET.recordSeriesScanCost( + getMetricType(), System.nanoTime() - startTime); + } + } + + private String metricType = null; + + private String getMetricType() { + if (metricType != null) { + return metricType; + } + if (isAligned) { + if (isMem) { + metricType = BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_MEM; + } else { + metricType = BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_DISK; + } + } else { + if (isMem) { + metricType = BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_MEM; + } else { + metricType = BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_DISK; + } + } + return metricType; + } + + void setFilter(Filter filter) { + data.setFilter(filter); + } + + boolean isModified() { + return data.isModified(); + } + + public boolean isSeq() { + return isSeq; + } + + public void setLimitOffset(PaginationController paginationController) { + data.setLimitOffset(paginationController); + } + } + + public interface TimeOrderUtils { + + long getOrderTime(Statistics<? extends Object> statistics); + + long getOrderTime(TsFileResource fileResource); + + long getOverlapCheckTime(Statistics<? extends Object> range); + + boolean isOverlapped(Statistics<? extends Object> left, Statistics<? extends Object> right); + + boolean isOverlapped(long time, Statistics<? extends Object> right); + + boolean isOverlapped(long time, TsFileResource right); + + <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor); + + long getCurrentEndPoint(long time, Statistics<? extends Object> statistics); + + long getCurrentEndPoint( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics); + + boolean isExcessEndpoint(long time, long endpointTime); + + /** Return true if taking first page reader from seq readers */ + boolean isTakeSeqAsFirst( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics); + + boolean getAscending(); + + boolean hasNextSeqResource(); + + boolean hasNextUnseqResource(); + + TsFileResource getNextSeqFileResource(boolean isDelete); + + TsFileResource getNextUnseqFileResource(boolean isDelete); + + void setCurSeqFileIndex(QueryDataSource dataSource); + } + + class DescTimeOrderUtils implements TimeOrderUtils { + + @SuppressWarnings("squid:S3740") + @Override + public long getOrderTime(Statistics statistics) { + return statistics.getEndTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public long getOrderTime(TsFileResource fileResource) { + return fileResource.getEndTime(seriesPath.getDevice()); + } + + @SuppressWarnings("squid:S3740") + @Override + public long getOverlapCheckTime(Statistics range) { + return range.getStartTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public boolean isOverlapped(Statistics left, Statistics right) { + return left.getStartTime() <= right.getEndTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public boolean isOverlapped(long time, Statistics right) { + return time <= right.getEndTime(); + } + + @Override + public boolean isOverlapped(long time, TsFileResource right) { + return time <= right.getEndTime(seriesPath.getDevice()); + } + + @Override + public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) { + Objects.requireNonNull(keyExtractor); + return (Comparator<T> & Serializable) + (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c2), keyExtractor.applyAsLong(c1)); + } + + @Override + public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) { + return Math.max(time, statistics.getStartTime()); + } + + @Override + public long getCurrentEndPoint( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { + return Math.max(seqStatistics.getStartTime(), unseqStatistics.getStartTime()); + } + + @Override + public boolean isExcessEndpoint(long time, long endpointTime) { + return time < endpointTime; + } + + @Override + public boolean isTakeSeqAsFirst( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { + return seqStatistics.getEndTime() > unseqStatistics.getEndTime(); + } + + @Override + public boolean getAscending() { + return false; + } + + @Override + public boolean hasNextSeqResource() { + while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) { + TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); + if (tsFileResource != null + && tsFileResource.isSatisfied( + seriesPath.getDevice(), getGlobalTimeFilter(), true, false)) { + break; + } + curSeqFileIndex--; + } + return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending()); + } + + @Override + public boolean hasNextUnseqResource() { + while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) { + TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); + if (tsFileResource != null + && tsFileResource.isSatisfied( + seriesPath.getDevice(), getGlobalTimeFilter(), false, false)) { + break; + } + curUnseqFileIndex++; + } + return dataSource.hasNextUnseqResource(curUnseqFileIndex); + } + + @Override + public TsFileResource getNextSeqFileResource(boolean isDelete) { + TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); + if (isDelete) { + curSeqFileIndex--; + } + return tsFileResource; + } + + @Override + public TsFileResource getNextUnseqFileResource(boolean isDelete) { + TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); + if (isDelete) { + curUnseqFileIndex++; + } + return tsFileResource; + } + + @Override + public void setCurSeqFileIndex(QueryDataSource dataSource) { + curSeqFileIndex = dataSource.getSeqResourcesSize() - 1; + } + } + + class AscTimeOrderUtils implements TimeOrderUtils { + + @SuppressWarnings("squid:S3740") + @Override + public long getOrderTime(Statistics statistics) { + return statistics.getStartTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public long getOrderTime(TsFileResource fileResource) { + return fileResource.getStartTime(seriesPath.getDevice()); + } + + @SuppressWarnings("squid:S3740") + @Override + public long getOverlapCheckTime(Statistics range) { + return range.getEndTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public boolean isOverlapped(Statistics left, Statistics right) { + return left.getEndTime() >= right.getStartTime(); + } + + @SuppressWarnings("squid:S3740") + @Override + public boolean isOverlapped(long time, Statistics right) { + return time >= right.getStartTime(); + } + + @Override + public boolean isOverlapped(long time, TsFileResource right) { + return time >= right.getStartTime(seriesPath.getDevice()); + } + + @Override + public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) { + Objects.requireNonNull(keyExtractor); + return (Comparator<T> & Serializable) + (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c1), keyExtractor.applyAsLong(c2)); + } + + @Override + public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) { + return Math.min(time, statistics.getEndTime()); + } + + @Override + public long getCurrentEndPoint( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { + return Math.min(seqStatistics.getEndTime(), unseqStatistics.getEndTime()); + } + + @Override + public boolean isExcessEndpoint(long time, long endpointTime) { + return time > endpointTime; + } + + @Override + public boolean isTakeSeqAsFirst( + Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { + return seqStatistics.getStartTime() < unseqStatistics.getStartTime(); + } + + @Override + public boolean getAscending() { + return true; + } + + @Override + public boolean hasNextSeqResource() { + while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) { + TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); + if (tsFileResource != null + && tsFileResource.isSatisfied( + seriesPath.getDevice(), getGlobalTimeFilter(), true, false)) { + break; + } + curSeqFileIndex++; + } + return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending()); + } + + @Override + public boolean hasNextUnseqResource() { + while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) { + TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); + if (tsFileResource != null + && tsFileResource.isSatisfied( + seriesPath.getDevice(), getGlobalTimeFilter(), false, false)) { + break; + } + curUnseqFileIndex++; + } + return dataSource.hasNextUnseqResource(curUnseqFileIndex); + } + + @Override + public TsFileResource getNextSeqFileResource(boolean isDelete) { + TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); + if (isDelete) { + curSeqFileIndex++; + } + return tsFileResource; + } + + @Override + public TsFileResource getNextUnseqFileResource(boolean isDelete) { + TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); + if (isDelete) { + curUnseqFileIndex++; + } + return tsFileResource; + } + + @Override + public void setCurSeqFileIndex(QueryDataSource dataSource) { + curSeqFileIndex = 0; + } + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/TimeSeriesMetadataProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/TimeSeriesMetadataProvider.java new file mode 100644 index 0000000..007e948 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/scan/TimeSeriesMetadataProvider.java
@@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.queryengine.execution.operator.scan; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.utils.FileLoaderUtils; +import org.apache.iotdb.tsfile.file.metadata.ITimeSeriesMetadata; +import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics; +import org.apache.iotdb.tsfile.read.filter.basic.Filter; +import org.apache.iotdb.tsfile.read.reader.series.PaginationController; + +import java.io.IOException; +import java.io.Serializable; +import java.util.LinkedList; +import java.util.List; +import java.util.PriorityQueue; +import java.util.Set; + +import static com.google.common.base.Preconditions.checkState; + +public class TimeSeriesMetadataProvider { + + private ITimeSeriesMetadata currentTimeSeriesMetadata; + + private final List<ITimeSeriesMetadata> seqTimeSeriesMetadata; + private final PriorityQueue<ITimeSeriesMetadata> unSeqTimeSeriesMetadata; + + private final SeriesScanUtil.TimeOrderUtils orderUtils; + + public TimeSeriesMetadataProvider(SeriesScanUtil.TimeOrderUtils orderUtils) { + this.orderUtils = orderUtils; + this.seqTimeSeriesMetadata = new LinkedList<>(); + this.unSeqTimeSeriesMetadata = + new PriorityQueue<>( + orderUtils.comparingLong( + timeSeriesMetadata -> orderUtils.getOrderTime(timeSeriesMetadata.getStatistics()))); + } + + public boolean hasCurrentFile() { + return currentTimeSeriesMetadata != null; + } + + public ITimeSeriesMetadata getCurrentFile() { + return currentTimeSeriesMetadata; + } + + public boolean hasNext() { + return orderUtils.hasNextSeqResource() + || orderUtils.hasNextUnseqResource() + || !seqTimeSeriesMetadata.isEmpty() + || !unSeqTimeSeriesMetadata.isEmpty(); + } + + @SuppressWarnings("unchecked") + public boolean currentFileOverlapped() { + checkState(hasCurrentFile(), "current file is null"); + + Statistics<? extends Serializable> fileStatistics = currentTimeSeriesMetadata.getStatistics(); + return !seqTimeSeriesMetadata.isEmpty() + && orderUtils.isOverlapped(fileStatistics, seqTimeSeriesMetadata.get(0).getStatistics()) + || !unSeqTimeSeriesMetadata.isEmpty() + && orderUtils.isOverlapped( + fileStatistics, unSeqTimeSeriesMetadata.peek().getStatistics()); + } + + public void skipCurrentFile() { + currentTimeSeriesMetadata = null; + } + + public boolean currentFileModified() { + checkState(hasCurrentFile(), "current file is null"); + + return currentTimeSeriesMetadata.isModified(); + } + + public void unpackAllOverlappedTsFilesToTimeSeriesMetadata(long endpointTime) throws IOException { + while (orderUtils.hasNextUnseqResource() + && orderUtils.isOverlapped(endpointTime, orderUtils.getNextUnseqFileResource(false))) { + unpackUnseqTsFileResource(); + } + while (orderUtils.hasNextSeqResource() + && orderUtils.isOverlapped(endpointTime, orderUtils.getNextSeqFileResource(false))) { + unpackSeqTsFileResource(); + } + } + + public void updateCurrentTimeSeriesMetadata() { + if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) { + // only has seq + currentTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0); + } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) { + // only has unseq + currentTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll(); + } else if (!seqTimeSeriesMetadata.isEmpty()) { + // has seq and unseq + if (orderUtils.isTakeSeqAsFirst( + seqTimeSeriesMetadata.get(0).getStatistics(), + unSeqTimeSeriesMetadata.peek().getStatistics())) { + currentTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0); + } else { + currentTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll(); + } + } + } + + public void filterFirstTimeSeriesMetadata( + Filter queryFilter, PaginationController paginationController) { + if (hasCurrentFile() && !currentFileOverlapped() && !currentFileModified()) { + Statistics statistics = getCurrentFile().getStatistics(); + if (queryFilter == null || queryFilter.allSatisfy(statistics)) { + long rowCount = statistics.getCount(); + if (paginationController.hasCurOffset(rowCount)) { + skipCurrentFile(); + paginationController.consumeOffset(rowCount); + } + } else if (!queryFilter.satisfy(statistics)) { + skipCurrentFile(); + } + } + } + + /** + * unpack all overlapped seq/unseq files and find the first TimeSeriesMetadata + * + * <p>Because there may be too many files in the scenario used by the user, we cannot open all the + * chunks at once, which may cause OOM, so we can only unpack one file at a time when needed. This + * approach is likely to be ubiquitous, but it keeps the system running smoothly + */ + public void tryToUnpackAllOverlappedFilesToTimeSeriesMetadata() throws IOException { + /* + * Fill sequence TimeSeriesMetadata List until it is not empty + */ + while (seqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextSeqResource()) { + unpackSeqTsFileResource(); + } + + /* + * Fill unSequence TimeSeriesMetadata Priority Queue until it is not empty + */ + while (unSeqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextUnseqResource()) { + unpackUnseqTsFileResource(); + } + + // find end time of the first TimeSeriesMetadata + long endTime = -1L; + if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) { + // only has seq + endTime = orderUtils.getOverlapCheckTime(seqTimeSeriesMetadata.get(0).getStatistics()); + } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) { + // only has unseq + endTime = orderUtils.getOverlapCheckTime(unSeqTimeSeriesMetadata.peek().getStatistics()); + } else if (!seqTimeSeriesMetadata.isEmpty()) { + // has seq and unseq + endTime = + orderUtils.getCurrentEndPoint( + seqTimeSeriesMetadata.get(0).getStatistics(), + unSeqTimeSeriesMetadata.peek().getStatistics()); + } + + // unpack all directly overlapped seq/unseq files with first TimeSeriesMetadata + if (endTime != -1) { + unpackAllOverlappedTsFilesToTimeSeriesMetadata(endTime); + } + + updateCurrentTimeSeriesMetadata(); + } + + private void unpackSeqTsFileResource() throws IOException { + ITimeSeriesMetadata timeseriesMetadata = + loadTimeSeriesMetadata( + orderUtils.getNextSeqFileResource(true), + seriesPath, + context, + getGlobalTimeFilter(), + scanOptions.getAllSensors()); + if (timeseriesMetadata != null) { + timeseriesMetadata.setSeq(true); + seqTimeSeriesMetadata.add(timeseriesMetadata); + } + } + + private void unpackUnseqTsFileResource() throws IOException { + ITimeSeriesMetadata timeseriesMetadata = + loadTimeSeriesMetadata( + orderUtils.getNextUnseqFileResource(true), + seriesPath, + context, + getGlobalTimeFilter(), + scanOptions.getAllSensors()); + if (timeseriesMetadata != null) { + timeseriesMetadata.setModified(true); + timeseriesMetadata.setSeq(false); + unSeqTimeSeriesMetadata.add(timeseriesMetadata); + } + } + + private ITimeSeriesMetadata loadTimeSeriesMetadata( + TsFileResource resource, + PartialPath seriesPath, + QueryContext context, + Filter filter, + Set<String> allSensors) + throws IOException { + return FileLoaderUtils.loadTimeSeriesMetadata( + resource, seriesPath, context, filter, allSensors); + } +}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractDataSourceOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractDataSourceOperator.java index 41fc28f..61ee26f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractDataSourceOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractDataSourceOperator.java
@@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.execution.operator.source; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java index d353bd9..8387113 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator; import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator; import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; @@ -225,7 +226,7 @@ // start stopwatch long start = System.nanoTime(); while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextFile()) { - if (canUseCurrentFileStatistics()) { + if (seriesScanUtil.canUseCurrentFileStatistics()) { Statistics fileTimeStatistics = seriesScanUtil.currentFileTimeStatistics(); if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) { if (ascending) { @@ -266,7 +267,7 @@ // start stopwatch long start = System.nanoTime(); while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextChunk()) { - if (canUseCurrentChunkStatistics()) { + if (seriesScanUtil.canUseCurrentChunkStatistics()) { Statistics chunkTimeStatistics = seriesScanUtil.currentChunkTimeStatistics(); if (chunkTimeStatistics.getStartTime() > curTimeRange.getMax()) { if (ascending) { @@ -308,7 +309,7 @@ long start = System.nanoTime(); try { while (System.nanoTime() - start < leftRuntimeOfOneNextCall && seriesScanUtil.hasNextPage()) { - if (canUseCurrentPageStatistics()) { + if (seriesScanUtil.canUseCurrentPageStatistics()) { Statistics pageTimeStatistics = seriesScanUtil.currentPageTimeStatistics(); // There is no more eligible points in current time range if (pageTimeStatistics.getStartTime() > curTimeRange.getMax()) { @@ -353,33 +354,6 @@ } } - @SuppressWarnings({"squid:S3740"}) - protected boolean canUseCurrentFileStatistics() throws IOException { - Statistics fileStatistics = seriesScanUtil.currentFileTimeStatistics(); - return !seriesScanUtil.isFileOverlapped() - && fileStatistics.containedByTimeFilter(seriesScanUtil.getGlobalTimeFilter()) - && !seriesScanUtil.currentFileModified(); - } - - @SuppressWarnings({"squid:S3740"}) - protected boolean canUseCurrentChunkStatistics() throws IOException { - Statistics chunkStatistics = seriesScanUtil.currentChunkTimeStatistics(); - return !seriesScanUtil.isChunkOverlapped() - && chunkStatistics.containedByTimeFilter(seriesScanUtil.getGlobalTimeFilter()) - && !seriesScanUtil.currentChunkModified(); - } - - @SuppressWarnings({"squid:S3740"}) - protected boolean canUseCurrentPageStatistics() throws IOException { - Statistics currentPageStatistics = seriesScanUtil.currentPageTimeStatistics(); - if (currentPageStatistics == null) { - return false; - } - return !seriesScanUtil.isPageOverlapped() - && currentPageStatistics.containedByTimeFilter(seriesScanUtil.getGlobalTimeFilter()) - && !seriesScanUtil.currentPageModified(); - } - @Override protected List<TSDataType> getResultDataTypes() { List<TSDataType> dataTypes = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesAggregationScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesAggregationScanOperator.java index 8690c5c..72d8e08 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesAggregationScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesAggregationScanOperator.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator; import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator; import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; +import org.apache.iotdb.db.queryengine.execution.operator.scan.AlignedSeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java index 344b194..830b104 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java
@@ -21,6 +21,7 @@ import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; +import org.apache.iotdb.db.queryengine.execution.operator.scan.AlignedSeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesAggregationScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesAggregationScanOperator.java index 56b33b0..8ab9a64 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesAggregationScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesAggregationScanOperator.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator; import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator; import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java index acdbdf5..6d076b5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java
@@ -21,6 +21,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java deleted file mode 100644 index 77f1c10..0000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ /dev/null
@@ -1,1493 +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.queryengine.execution.operator.source; - -import org.apache.iotdb.commons.path.PartialPath; -import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; -import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; -import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet; -import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; -import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; -import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; -import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemAlignedPageReader; -import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemPageReader; -import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.DescPriorityMergeReader; -import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; -import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; -import org.apache.iotdb.db.utils.FileLoaderUtils; -import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; -import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata; -import org.apache.iotdb.tsfile.file.metadata.ITimeSeriesMetadata; -import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; -import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics; -import org.apache.iotdb.tsfile.read.TimeValuePair; -import org.apache.iotdb.tsfile.read.common.block.TsBlock; -import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder; -import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder; -import org.apache.iotdb.tsfile.read.filter.basic.Filter; -import org.apache.iotdb.tsfile.read.reader.IAlignedPageReader; -import org.apache.iotdb.tsfile.read.reader.IPageReader; -import org.apache.iotdb.tsfile.read.reader.IPointReader; -import org.apache.iotdb.tsfile.read.reader.series.PaginationController; -import org.apache.iotdb.tsfile.utils.TsPrimitiveType; - -import java.io.IOException; -import java.io.Serializable; -import java.util.Collections; -import java.util.Comparator; -import java.util.LinkedList; -import java.util.List; -import java.util.Objects; -import java.util.PriorityQueue; -import java.util.Set; -import java.util.function.ToLongFunction; - -import static com.google.common.base.Preconditions.checkArgument; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_MERGE_READER_ALIGNED; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_MERGE_READER_NONALIGNED; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_DISK; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_MEM; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_DISK; -import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_MEM; - -public class SeriesScanUtil { - - private static final String ONE_SENSOR_ERROR_MSG = - "Only one sensor in non-aligned SeriesScanUtil."; - - private final QueryContext context; - - // The path of the target series which will be scanned. - private final PartialPath seriesPath; - protected boolean isAligned = false; - protected final TSDataType dataType; - - // inner class of SeriesReader for order purpose - private final TimeOrderUtils orderUtils; - - private QueryDataSource dataSource; - - // file index - protected int curSeqFileIndex; - protected int curUnseqFileIndex; - - // TimeSeriesMetadata cache - protected ITimeSeriesMetadata firstTimeSeriesMetadata; - protected final List<ITimeSeriesMetadata> seqTimeSeriesMetadata; - protected final PriorityQueue<ITimeSeriesMetadata> unSeqTimeSeriesMetadata; - - // chunk cache - protected IChunkMetadata firstChunkMetadata; - protected final PriorityQueue<IChunkMetadata> cachedChunkMetadata; - - // page cache - protected VersionPageReader firstPageReader; - protected final List<VersionPageReader> seqPageReaders; - protected final PriorityQueue<VersionPageReader> unSeqPageReaders; - - // point cache - protected final PriorityMergeReader mergeReader; - - // result cache - protected boolean hasCachedNextOverlappedPage; - protected TsBlock cachedTsBlock; - - protected SeriesScanOptions scanOptions; - protected PaginationController paginationController; - - private static final SeriesScanCostMetricSet SERIES_SCAN_COST_METRIC_SET = - SeriesScanCostMetricSet.getInstance(); - - public SeriesScanUtil( - PartialPath seriesPath, - Ordering scanOrder, - SeriesScanOptions scanOptions, - FragmentInstanceContext context) { - this.seriesPath = seriesPath; - dataType = seriesPath.getSeriesType(); - - this.scanOptions = scanOptions; - paginationController = scanOptions.getPaginationController(); - - this.context = context; - - if (scanOrder.isAscending()) { - orderUtils = new AscTimeOrderUtils(); - mergeReader = getPriorityMergeReader(); - } else { - orderUtils = new DescTimeOrderUtils(); - mergeReader = getDescPriorityMergeReader(); - } - - // init TimeSeriesMetadata materializer - seqTimeSeriesMetadata = new LinkedList<>(); - unSeqTimeSeriesMetadata = - new PriorityQueue<>( - orderUtils.comparingLong( - timeSeriesMetadata -> orderUtils.getOrderTime(timeSeriesMetadata.getStatistics()))); - - // init ChunkMetadata materializer - cachedChunkMetadata = - new PriorityQueue<>( - orderUtils.comparingLong( - chunkMetadata -> orderUtils.getOrderTime(chunkMetadata.getStatistics()))); - - // init PageReader materializer - seqPageReaders = new LinkedList<>(); - unSeqPageReaders = - new PriorityQueue<>( - orderUtils.comparingLong( - versionPageReader -> orderUtils.getOrderTime(versionPageReader.getStatistics()))); - } - - public void initQueryDataSource(QueryDataSource dataSource) { - dataSource.fillOrderIndexes(seriesPath.getDevice(), orderUtils.getAscending()); - this.dataSource = dataSource; - - // updated filter concerning TTL - scanOptions.setTTL(dataSource.getDataTTL()); - - // init file index - orderUtils.setCurSeqFileIndex(dataSource); - curUnseqFileIndex = 0; - } - - protected PriorityMergeReader getPriorityMergeReader() { - return new PriorityMergeReader(); - } - - protected DescPriorityMergeReader getDescPriorityMergeReader() { - return new DescPriorityMergeReader(); - } - - public boolean hasNextFile() throws IOException { - if (!paginationController.hasCurLimit()) { - return false; - } - - if (!unSeqPageReaders.isEmpty() - || firstPageReader != null - || mergeReader.hasNextTimeValuePair()) { - throw new IOException( - "all cached pages should be consumed first unSeqPageReaders.isEmpty() is " - + unSeqPageReaders.isEmpty() - + " firstPageReader != null is " - + (firstPageReader != null) - + " mergeReader.hasNextTimeValuePair() = " - + mergeReader.hasNextTimeValuePair()); - } - - if (firstChunkMetadata != null || !cachedChunkMetadata.isEmpty()) { - throw new IOException("all cached chunks should be consumed first"); - } - - if (firstTimeSeriesMetadata != null) { - return true; - } - - while (firstTimeSeriesMetadata == null - && (orderUtils.hasNextSeqResource() - || orderUtils.hasNextUnseqResource() - || !seqTimeSeriesMetadata.isEmpty() - || !unSeqTimeSeriesMetadata.isEmpty())) { - // init first time series metadata whose startTime is minimum - tryToUnpackAllOverlappedFilesToTimeSeriesMetadata(); - // filter file based on push-down conditions - filterFirstTimeSeriesMetadata(); - } - - return firstTimeSeriesMetadata != null; - } - - @SuppressWarnings("squid:S3740") - boolean isFileOverlapped() throws IOException { - if (firstTimeSeriesMetadata == null) { - throw new IOException("no first file"); - } - - Statistics fileStatistics = firstTimeSeriesMetadata.getStatistics(); - return !seqTimeSeriesMetadata.isEmpty() - && orderUtils.isOverlapped(fileStatistics, seqTimeSeriesMetadata.get(0).getStatistics()) - || !unSeqTimeSeriesMetadata.isEmpty() - && orderUtils.isOverlapped( - fileStatistics, unSeqTimeSeriesMetadata.peek().getStatistics()); - } - - @SuppressWarnings("squid:S3740") - Statistics currentFileStatistics() { - return firstTimeSeriesMetadata.getStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentFileStatistics(int index) { - checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); - return currentFileStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentFileTimeStatistics() { - return currentFileStatistics(); - } - - boolean currentFileModified() throws IOException { - if (firstTimeSeriesMetadata == null) { - throw new IOException("no first file"); - } - return firstTimeSeriesMetadata.isModified(); - } - - void skipCurrentFile() { - firstTimeSeriesMetadata = null; - } - - /** - * This method should be called after hasNextFile() until no next chunk, make sure that all - * overlapped chunks are consumed - */ - public boolean hasNextChunk() throws IOException { - if (!paginationController.hasCurLimit()) { - return false; - } - - if (!unSeqPageReaders.isEmpty() - || firstPageReader != null - || mergeReader.hasNextTimeValuePair()) { - throw new IOException( - "all cached pages should be consumed first unSeqPageReaders.isEmpty() is " - + unSeqPageReaders.isEmpty() - + " firstPageReader != null is " - + (firstPageReader != null) - + " mergeReader.hasNextTimeValuePair() = " - + mergeReader.hasNextTimeValuePair()); - } - - if (firstChunkMetadata != null) { - return true; - // hasNextFile() has not been invoked - } else if (firstTimeSeriesMetadata == null && cachedChunkMetadata.isEmpty()) { - return false; - } - - while (firstChunkMetadata == null && (!cachedChunkMetadata.isEmpty() || hasNextFile())) { - initFirstChunkMetadata(); - // filter chunk based on push-down conditions - filterFirstChunkMetadata(); - } - return firstChunkMetadata != null; - } - - @SuppressWarnings("squid:S3740") - protected void filterFirstChunkMetadata() throws IOException { - if (firstChunkMetadata != null && !isChunkOverlapped() && !firstChunkMetadata.isModified()) { - Filter queryFilter = scanOptions.getPushDownFilter(); - Statistics statistics = firstChunkMetadata.getStatistics(); - if (queryFilter == null || queryFilter.allSatisfy(statistics)) { - long rowCount = statistics.getCount(); - if (paginationController.hasCurOffset(rowCount)) { - skipCurrentChunk(); - paginationController.consumeOffset(rowCount); - } - } else if (!queryFilter.satisfy(statistics)) { - skipCurrentChunk(); - } - } - } - - /** construct first chunk metadata */ - private void initFirstChunkMetadata() throws IOException { - if (firstTimeSeriesMetadata != null) { - /* - * try to unpack all overlapped TimeSeriesMetadata to cachedChunkMetadata - */ - unpackAllOverlappedTsFilesToTimeSeriesMetadata( - orderUtils.getOverlapCheckTime(firstTimeSeriesMetadata.getStatistics())); - unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( - orderUtils.getOverlapCheckTime(firstTimeSeriesMetadata.getStatistics()), true); - } else { - /* - * first time series metadata is already unpacked, consume cached ChunkMetadata - */ - while (!cachedChunkMetadata.isEmpty()) { - firstChunkMetadata = cachedChunkMetadata.peek(); - unpackAllOverlappedTsFilesToTimeSeriesMetadata( - orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics())); - unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( - orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), false); - if (firstChunkMetadata.equals(cachedChunkMetadata.peek())) { - firstChunkMetadata = cachedChunkMetadata.poll(); - break; - } - } - } - } - - private void unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( - long endpointTime, boolean init) throws IOException { - while (!seqTimeSeriesMetadata.isEmpty() - && orderUtils.isOverlapped(endpointTime, seqTimeSeriesMetadata.get(0).getStatistics())) { - unpackOneTimeSeriesMetadata(seqTimeSeriesMetadata.remove(0)); - } - while (!unSeqTimeSeriesMetadata.isEmpty() - && orderUtils.isOverlapped(endpointTime, unSeqTimeSeriesMetadata.peek().getStatistics())) { - unpackOneTimeSeriesMetadata(unSeqTimeSeriesMetadata.poll()); - } - - if (firstTimeSeriesMetadata != null - && orderUtils.isOverlapped(endpointTime, firstTimeSeriesMetadata.getStatistics())) { - unpackOneTimeSeriesMetadata(firstTimeSeriesMetadata); - firstTimeSeriesMetadata = null; - } - - if (init && firstChunkMetadata == null && !cachedChunkMetadata.isEmpty()) { - firstChunkMetadata = cachedChunkMetadata.poll(); - } - } - - protected void unpackOneTimeSeriesMetadata(ITimeSeriesMetadata timeSeriesMetadata) { - List<IChunkMetadata> chunkMetadataList = - FileLoaderUtils.loadChunkMetadataList(timeSeriesMetadata); - chunkMetadataList.forEach(chunkMetadata -> chunkMetadata.setSeq(timeSeriesMetadata.isSeq())); - - cachedChunkMetadata.addAll(chunkMetadataList); - } - - @SuppressWarnings("squid:S3740") - boolean isChunkOverlapped() throws IOException { - if (firstChunkMetadata == null) { - throw new IOException("no first chunk"); - } - - Statistics chunkStatistics = firstChunkMetadata.getStatistics(); - return !cachedChunkMetadata.isEmpty() - && orderUtils.isOverlapped(chunkStatistics, cachedChunkMetadata.peek().getStatistics()); - } - - @SuppressWarnings("squid:S3740") - Statistics currentChunkStatistics() { - return firstChunkMetadata.getStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentChunkStatistics(int index) { - checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); - return currentChunkStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentChunkTimeStatistics() { - return currentChunkStatistics(); - } - - boolean currentChunkModified() throws IOException { - if (firstChunkMetadata == null) { - throw new IOException("no first chunk"); - } - return firstChunkMetadata.isModified(); - } - - void skipCurrentChunk() { - firstChunkMetadata = null; - } - - /** - * This method should be called after hasNextChunk() until no next page, make sure that all - * overlapped pages are consumed - */ - @SuppressWarnings("squid:S3776") - // Suppress high Cognitive Complexity warning - public boolean hasNextPage() throws IOException { - if (!paginationController.hasCurLimit()) { - return false; - } - - /* - * has overlapped data before - */ - if (hasCachedNextOverlappedPage) { - return true; - } else if (mergeReader.hasNextTimeValuePair() || firstPageOverlapped()) { - if (hasNextOverlappedPage()) { - cachedTsBlock = nextOverlappedPage(); - if (cachedTsBlock != null && !cachedTsBlock.isEmpty()) { - hasCachedNextOverlappedPage = true; - return true; - } - } - } - - if (firstPageReader != null) { - return true; - } - - /* - * construct first page reader - */ - if (firstChunkMetadata != null) { - /* - * try to unpack all overlapped ChunkMetadata to cachedPageReaders - */ - unpackAllOverlappedChunkMetadataToPageReaders( - orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), true); - } else { - /* - * first chunk metadata is already unpacked, consume cached pages - */ - initFirstPageReader(); - } - - if (isExistOverlappedPage()) { - return true; - } - - // make sure firstPageReader won't be null while the unSeqPageReaders has more cached page - // readers - while (firstPageReader == null && (!seqPageReaders.isEmpty() || !unSeqPageReaders.isEmpty())) { - - initFirstPageReader(); - - if (isExistOverlappedPage()) { - return true; - } - } - return firstPageReader != null; - } - - private boolean isExistOverlappedPage() throws IOException { - if (firstPageOverlapped()) { - /* - * next page is overlapped, read overlapped data and cache it - */ - if (hasNextOverlappedPage()) { - cachedTsBlock = nextOverlappedPage(); - if (cachedTsBlock != null && !cachedTsBlock.isEmpty()) { - hasCachedNextOverlappedPage = true; - return true; - } - } - } - return false; - } - - private boolean firstPageOverlapped() throws IOException { - if (firstPageReader == null) { - return false; - } - - long endpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); - unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime); - unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(endpointTime, false); - unpackAllOverlappedChunkMetadataToPageReaders(endpointTime, false); - - return (!seqPageReaders.isEmpty() - && orderUtils.isOverlapped( - firstPageReader.getStatistics(), seqPageReaders.get(0).getStatistics())) - || (!unSeqPageReaders.isEmpty() - && orderUtils.isOverlapped( - firstPageReader.getStatistics(), unSeqPageReaders.peek().getStatistics()) - || (mergeReader.hasNextTimeValuePair() - && orderUtils.isOverlapped( - mergeReader.currentTimeValuePair().getTimestamp(), - firstPageReader.getStatistics()))); - } - - private void unpackAllOverlappedChunkMetadataToPageReaders(long endpointTime, boolean init) - throws IOException { - if (firstChunkMetadata != null - && orderUtils.isOverlapped(endpointTime, firstChunkMetadata.getStatistics())) { - unpackOneChunkMetaData(firstChunkMetadata); - firstChunkMetadata = null; - } - // In case unpacking too many sequence chunks - boolean hasMeetSeq = false; - while (!cachedChunkMetadata.isEmpty() - && orderUtils.isOverlapped(endpointTime, cachedChunkMetadata.peek().getStatistics())) { - if (cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { - break; - } else if (cachedChunkMetadata.peek().isSeq()) { - hasMeetSeq = true; - } - unpackOneChunkMetaData(cachedChunkMetadata.poll()); - } - if (init - && firstPageReader == null - && (!seqPageReaders.isEmpty() || !unSeqPageReaders.isEmpty())) { - initFirstPageReader(); - } - } - - private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOException { - List<IPageReader> pageReaderList = - FileLoaderUtils.loadPageReaderList(chunkMetaData, getGlobalTimeFilter()); - - // init TsBlockBuilder for each page reader - pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); - - if (chunkMetaData.isSeq()) { - if (orderUtils.getAscending()) { - for (IPageReader iPageReader : pageReaderList) { - seqPageReaders.add( - new VersionPageReader( - chunkMetaData.getVersion(), - chunkMetaData.getOffsetOfChunkHeader(), - iPageReader, - true)); - } - } else { - for (int i = pageReaderList.size() - 1; i >= 0; i--) { - seqPageReaders.add( - new VersionPageReader( - chunkMetaData.getVersion(), - chunkMetaData.getOffsetOfChunkHeader(), - pageReaderList.get(i), - true)); - } - } - } else { - pageReaderList.forEach( - pageReader -> - unSeqPageReaders.add( - new VersionPageReader( - chunkMetaData.getVersion(), - chunkMetaData.getOffsetOfChunkHeader(), - pageReader, - false))); - } - } - - /** - * This method should be called after calling hasNextPage. - * - * <p>hasNextPage may cache firstPageReader if it is not overlapped or cached a BatchData if the - * first page is overlapped - */ - @SuppressWarnings("squid:S3740") - boolean isPageOverlapped() throws IOException { - - /* - * has an overlapped page - */ - if (hasCachedNextOverlappedPage) { - return true; - } - - /* - * has a non-overlapped page in firstPageReader - */ - if (mergeReader.hasNextTimeValuePair() - && ((orderUtils.getAscending() - && mergeReader.currentTimeValuePair().getTimestamp() - <= firstPageReader.getStatistics().getEndTime()) - || (!orderUtils.getAscending() - && mergeReader.currentTimeValuePair().getTimestamp() - >= firstPageReader.getStatistics().getStartTime()))) { - throw new IOException("overlapped data should be consumed first"); - } - - Statistics firstPageStatistics = firstPageReader.getStatistics(); - - return !unSeqPageReaders.isEmpty() - && orderUtils.isOverlapped(firstPageStatistics, unSeqPageReaders.peek().getStatistics()); - } - - @SuppressWarnings("squid:S3740") - Statistics currentPageStatistics() { - if (firstPageReader == null) { - return null; - } - return firstPageReader.getStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentPageStatistics(int index) throws IOException { - checkArgument(index == 0, ONE_SENSOR_ERROR_MSG); - return currentPageStatistics(); - } - - @SuppressWarnings("squid:S3740") - protected Statistics currentPageTimeStatistics() throws IOException { - return currentPageStatistics(); - } - - boolean currentPageModified() throws IOException { - if (firstPageReader == null) { - throw new IOException("no first page"); - } - return firstPageReader.isModified(); - } - - void skipCurrentPage() { - firstPageReader = null; - } - - /** This method should only be used when the method isPageOverlapped() return true. */ - public TsBlock nextPage() throws IOException { - - if (hasCachedNextOverlappedPage) { - hasCachedNextOverlappedPage = false; - TsBlock res = cachedTsBlock; - cachedTsBlock = null; - return res; - } else { - // next page is not overlapped, push down value filter & limit offset - Filter queryFilter = scanOptions.getPushDownFilter(); - if (queryFilter != null) { - firstPageReader.setFilter(queryFilter); - } - TsBlock tsBlock; - if (orderUtils.getAscending()) { - firstPageReader.setLimitOffset(paginationController); - tsBlock = firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending()); - } else { - tsBlock = - paginationController.applyTsBlock( - firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending())); - } - - firstPageReader = null; - - return tsBlock; - } - } - - /** - * read overlapped data till currentLargestEndTime in mergeReader, if current batch does not - * contain data, read till next currentLargestEndTime again - */ - @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning - private boolean hasNextOverlappedPage() throws IOException { - long startTime = System.nanoTime(); - try { - if (hasCachedNextOverlappedPage) { - return true; - } - - tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader(); - - while (true) { - - // may has overlapped data - if (mergeReader.hasNextTimeValuePair()) { - - TsBlockBuilder builder = new TsBlockBuilder(getTsDataTypeList()); - TimeColumnBuilder timeBuilder = builder.getTimeColumnBuilder(); - long currentPageEndPointTime = mergeReader.getCurrentReadStopTime(); - while (mergeReader.hasNextTimeValuePair()) { - - /* - * get current first point in mergeReader, this maybe overlapped later - */ - TimeValuePair timeValuePair = mergeReader.currentTimeValuePair(); - - if (orderUtils.isExcessEndpoint( - timeValuePair.getTimestamp(), currentPageEndPointTime)) { - /* - * when the merged point excesses the currentPageEndPointTime, we have read all overlapped data before currentPageEndPointTime - * 1. has cached batch data, we don't need to read more data, just use the cached data later - * 2. has first page reader, which means first page reader last endTime < currentTimeValuePair.getTimestamp(), - * we could just use the first page reader later - * 3. sequence page reader is not empty, which means first page reader last endTime < currentTimeValuePair.getTimestamp(), - * we could use the first sequence page reader later - */ - if (!builder.isEmpty() || firstPageReader != null || !seqPageReaders.isEmpty()) { - break; - } - // so, we don't have other data except mergeReader - currentPageEndPointTime = mergeReader.getCurrentReadStopTime(); - } - - // unpack all overlapped data for the first timeValuePair - unpackAllOverlappedTsFilesToTimeSeriesMetadata(timeValuePair.getTimestamp()); - unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata( - timeValuePair.getTimestamp(), false); - unpackAllOverlappedChunkMetadataToPageReaders(timeValuePair.getTimestamp(), false); - unpackAllOverlappedUnseqPageReadersToMergeReader(timeValuePair.getTimestamp()); - - // update if there are unpacked unSeqPageReaders - timeValuePair = mergeReader.currentTimeValuePair(); - - // from now, the unsequence reader is all unpacked, so we don't need to consider it - // we has first page reader now - if (firstPageReader != null) { - // if current timeValuePair excesses the first page reader's end time, we just use the - // cached data - if ((orderUtils.getAscending() - && timeValuePair.getTimestamp() - > firstPageReader.getStatistics().getEndTime()) - || (!orderUtils.getAscending() - && timeValuePair.getTimestamp() - < firstPageReader.getStatistics().getStartTime())) { - hasCachedNextOverlappedPage = !builder.isEmpty(); - cachedTsBlock = builder.build(); - return hasCachedNextOverlappedPage; - } else if (orderUtils.isOverlapped( - timeValuePair.getTimestamp(), firstPageReader.getStatistics())) { - // current timeValuePair is overlapped with firstPageReader, add it to merged reader - // and update endTime to the max end time - mergeReader.addReader( - getPointReader( - firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending())), - firstPageReader.version, - orderUtils.getOverlapCheckTime(firstPageReader.getStatistics())); - currentPageEndPointTime = - updateEndPointTime(currentPageEndPointTime, firstPageReader); - firstPageReader = null; - } - } - - // the seq page readers is not empty, just like first page reader - if (!seqPageReaders.isEmpty()) { - if ((orderUtils.getAscending() - && timeValuePair.getTimestamp() - > seqPageReaders.get(0).getStatistics().getEndTime()) - || (!orderUtils.getAscending() - && timeValuePair.getTimestamp() - < seqPageReaders.get(0).getStatistics().getStartTime())) { - hasCachedNextOverlappedPage = !builder.isEmpty(); - cachedTsBlock = builder.build(); - return hasCachedNextOverlappedPage; - } else if (orderUtils.isOverlapped( - timeValuePair.getTimestamp(), seqPageReaders.get(0).getStatistics())) { - VersionPageReader pageReader = seqPageReaders.remove(0); - mergeReader.addReader( - getPointReader(pageReader.getAllSatisfiedPageData(orderUtils.getAscending())), - pageReader.version, - orderUtils.getOverlapCheckTime(pageReader.getStatistics())); - currentPageEndPointTime = updateEndPointTime(currentPageEndPointTime, pageReader); - } - } - - /* - * get the latest first point in mergeReader - */ - timeValuePair = mergeReader.nextTimeValuePair(); - - Object valueForFilter = timeValuePair.getValue().getValue(); - - // TODO fix value filter firstNotNullObject, currently, if it's a value filter, it will - // only accept AlignedPath with only one sub sensor - if (timeValuePair.getValue().getDataType() == TSDataType.VECTOR) { - for (TsPrimitiveType tsPrimitiveType : timeValuePair.getValue().getVector()) { - if (tsPrimitiveType != null) { - valueForFilter = tsPrimitiveType.getValue(); - break; - } - } - } - - Filter queryFilter = scanOptions.getPushDownFilter(); - if (queryFilter != null - && !queryFilter.satisfy(timeValuePair.getTimestamp(), valueForFilter)) { - continue; - } - if (paginationController.hasCurOffset()) { - paginationController.consumeOffset(); - continue; - } - if (paginationController.hasCurLimit()) { - timeBuilder.writeLong(timeValuePair.getTimestamp()); - switch (dataType) { - case BOOLEAN: - builder.getColumnBuilder(0).writeBoolean(timeValuePair.getValue().getBoolean()); - break; - case INT32: - builder.getColumnBuilder(0).writeInt(timeValuePair.getValue().getInt()); - break; - case INT64: - builder.getColumnBuilder(0).writeLong(timeValuePair.getValue().getLong()); - break; - case FLOAT: - builder.getColumnBuilder(0).writeFloat(timeValuePair.getValue().getFloat()); - break; - case DOUBLE: - builder.getColumnBuilder(0).writeDouble(timeValuePair.getValue().getDouble()); - break; - case TEXT: - builder.getColumnBuilder(0).writeBinary(timeValuePair.getValue().getBinary()); - break; - case VECTOR: - TsPrimitiveType[] values = timeValuePair.getValue().getVector(); - for (int i = 0; i < values.length; i++) { - if (values[i] == null) { - builder.getColumnBuilder(i).appendNull(); - } else { - builder.getColumnBuilder(i).writeTsPrimitiveType(values[i]); - } - } - break; - default: - throw new UnSupportedDataTypeException(String.valueOf(dataType)); - } - builder.declarePosition(); - paginationController.consumeLimit(); - } else { - break; - } - } - hasCachedNextOverlappedPage = !builder.isEmpty(); - cachedTsBlock = builder.build(); - /* - * if current overlapped page has valid data, return, otherwise read next overlapped page - */ - if (hasCachedNextOverlappedPage) { - return true; - } else if (mergeReader.hasNextTimeValuePair()) { - // condition: seqPage.endTime < mergeReader.currentTime - return false; - } - } else { - return false; - } - } - } finally { - SERIES_SCAN_COST_METRIC_SET.recordSeriesScanCost( - isAligned - ? BUILD_TSBLOCK_FROM_MERGE_READER_ALIGNED - : BUILD_TSBLOCK_FROM_MERGE_READER_NONALIGNED, - System.nanoTime() - startTime); - } - } - - private long updateEndPointTime(long currentPageEndPointTime, VersionPageReader pageReader) { - if (orderUtils.getAscending()) { - return Math.min(currentPageEndPointTime, pageReader.getStatistics().getEndTime()); - } else { - return Math.max(currentPageEndPointTime, pageReader.getStatistics().getStartTime()); - } - } - - private void tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader() throws IOException { - - /* - * no cached page readers - */ - if (firstPageReader == null && unSeqPageReaders.isEmpty() && seqPageReaders.isEmpty()) { - return; - } - - /* - * init firstPageReader - */ - if (firstPageReader == null) { - initFirstPageReader(); - } - - long currentPageEndpointTime; - if (mergeReader.hasNextTimeValuePair()) { - currentPageEndpointTime = mergeReader.getCurrentReadStopTime(); - } else { - currentPageEndpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); - } - - /* - * put all currently directly overlapped unseq page reader to merge reader - */ - unpackAllOverlappedUnseqPageReadersToMergeReader(currentPageEndpointTime); - } - - private void initFirstPageReader() throws IOException { - while (this.firstPageReader == null) { - VersionPageReader firstPageReader = getFirstPageReaderFromCachedReaders(); - - // unpack overlapped page using current page reader - if (firstPageReader != null) { - long overlapCheckTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); - unpackAllOverlappedTsFilesToTimeSeriesMetadata(overlapCheckTime); - unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(overlapCheckTime, false); - unpackAllOverlappedChunkMetadataToPageReaders(overlapCheckTime, false); - - // this page after unpacking must be the first page - if (firstPageReader.equals(getFirstPageReaderFromCachedReaders())) { - this.firstPageReader = firstPageReader; - if (!seqPageReaders.isEmpty() && firstPageReader.equals(seqPageReaders.get(0))) { - seqPageReaders.remove(0); - break; - } else if (!unSeqPageReaders.isEmpty() - && firstPageReader.equals(unSeqPageReaders.peek())) { - unSeqPageReaders.poll(); - break; - } - } - } else { - return; - } - } - } - - // We use get() and peek() here in case it's not the first page reader before unpacking - private VersionPageReader getFirstPageReaderFromCachedReaders() { - VersionPageReader firstPageReader = null; - if (!seqPageReaders.isEmpty() && !unSeqPageReaders.isEmpty()) { - if (orderUtils.isTakeSeqAsFirst( - seqPageReaders.get(0).getStatistics(), unSeqPageReaders.peek().getStatistics())) { - firstPageReader = seqPageReaders.get(0); - } else { - firstPageReader = unSeqPageReaders.peek(); - } - } else if (!seqPageReaders.isEmpty()) { - firstPageReader = seqPageReaders.get(0); - } else if (!unSeqPageReaders.isEmpty()) { - firstPageReader = unSeqPageReaders.peek(); - } - return firstPageReader; - } - - private void unpackAllOverlappedUnseqPageReadersToMergeReader(long endpointTime) - throws IOException { - while (!unSeqPageReaders.isEmpty() - && orderUtils.isOverlapped(endpointTime, unSeqPageReaders.peek().data.getStatistics())) { - putPageReaderToMergeReader(unSeqPageReaders.poll()); - } - if (firstPageReader != null - && !firstPageReader.isSeq() - && orderUtils.isOverlapped(endpointTime, firstPageReader.getStatistics())) { - putPageReaderToMergeReader(firstPageReader); - firstPageReader = null; - } - } - - private void putPageReaderToMergeReader(VersionPageReader pageReader) throws IOException { - mergeReader.addReader( - getPointReader(pageReader.getAllSatisfiedPageData(orderUtils.getAscending())), - pageReader.version, - orderUtils.getOverlapCheckTime(pageReader.getStatistics())); - } - - private TsBlock nextOverlappedPage() throws IOException { - if (hasCachedNextOverlappedPage || hasNextOverlappedPage()) { - hasCachedNextOverlappedPage = false; - return cachedTsBlock; - } - throw new IOException("No more batch data"); - } - - /** - * unpack all overlapped seq/unseq files and find the first TimeSeriesMetadata - * - * <p>Because there may be too many files in the scenario used by the user, we cannot open all the - * chunks at once, which may cause OOM, so we can only unpack one file at a time when needed. This - * approach is likely to be ubiquitous, but it keeps the system running smoothly - */ - @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning - protected void tryToUnpackAllOverlappedFilesToTimeSeriesMetadata() throws IOException { - /* - * Fill sequence TimeSeriesMetadata List until it is not empty - */ - while (seqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextSeqResource()) { - unpackSeqTsFileResource(); - } - - /* - * Fill unSequence TimeSeriesMetadata Priority Queue until it is not empty - */ - while (unSeqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextUnseqResource()) { - unpackUnseqTsFileResource(); - } - - /* - * find end time of the first TimeSeriesMetadata - */ - long endTime = -1L; - if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) { - // only has seq - endTime = orderUtils.getOverlapCheckTime(seqTimeSeriesMetadata.get(0).getStatistics()); - } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) { - // only has unseq - endTime = orderUtils.getOverlapCheckTime(unSeqTimeSeriesMetadata.peek().getStatistics()); - } else if (!seqTimeSeriesMetadata.isEmpty()) { - // has seq and unseq - endTime = - orderUtils.getCurrentEndPoint( - seqTimeSeriesMetadata.get(0).getStatistics(), - unSeqTimeSeriesMetadata.peek().getStatistics()); - } - - /* - * unpack all directly overlapped seq/unseq files with first TimeSeriesMetadata - */ - if (endTime != -1) { - unpackAllOverlappedTsFilesToTimeSeriesMetadata(endTime); - } - - /* - * update the first TimeSeriesMetadata - */ - if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) { - // only has seq - firstTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0); - } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) { - // only has unseq - firstTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll(); - } else if (!seqTimeSeriesMetadata.isEmpty()) { - // has seq and unseq - if (orderUtils.isTakeSeqAsFirst( - seqTimeSeriesMetadata.get(0).getStatistics(), - unSeqTimeSeriesMetadata.peek().getStatistics())) { - firstTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0); - } else { - firstTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll(); - } - } - } - - @SuppressWarnings("squid:S3740") - protected void filterFirstTimeSeriesMetadata() throws IOException { - if (firstTimeSeriesMetadata != null - && !isFileOverlapped() - && !firstTimeSeriesMetadata.isModified()) { - Filter queryFilter = scanOptions.getPushDownFilter(); - Statistics statistics = firstTimeSeriesMetadata.getStatistics(); - if (queryFilter == null || queryFilter.allSatisfy(statistics)) { - long rowCount = statistics.getCount(); - if (paginationController.hasCurOffset(rowCount)) { - skipCurrentFile(); - paginationController.consumeOffset(rowCount); - } - } else if (!queryFilter.satisfy(statistics)) { - skipCurrentFile(); - } - } - } - - protected void unpackAllOverlappedTsFilesToTimeSeriesMetadata(long endpointTime) - throws IOException { - while (orderUtils.hasNextUnseqResource() - && orderUtils.isOverlapped(endpointTime, orderUtils.getNextUnseqFileResource(false))) { - unpackUnseqTsFileResource(); - } - while (orderUtils.hasNextSeqResource() - && orderUtils.isOverlapped(endpointTime, orderUtils.getNextSeqFileResource(false))) { - unpackSeqTsFileResource(); - } - } - - private void unpackSeqTsFileResource() throws IOException { - ITimeSeriesMetadata timeseriesMetadata = - loadTimeSeriesMetadata( - orderUtils.getNextSeqFileResource(true), - seriesPath, - context, - getGlobalTimeFilter(), - scanOptions.getAllSensors()); - if (timeseriesMetadata != null) { - timeseriesMetadata.setSeq(true); - seqTimeSeriesMetadata.add(timeseriesMetadata); - } - } - - private void unpackUnseqTsFileResource() throws IOException { - ITimeSeriesMetadata timeseriesMetadata = - loadTimeSeriesMetadata( - orderUtils.getNextUnseqFileResource(true), - seriesPath, - context, - getGlobalTimeFilter(), - scanOptions.getAllSensors()); - if (timeseriesMetadata != null) { - timeseriesMetadata.setModified(true); - timeseriesMetadata.setSeq(false); - unSeqTimeSeriesMetadata.add(timeseriesMetadata); - } - } - - protected ITimeSeriesMetadata loadTimeSeriesMetadata( - TsFileResource resource, - PartialPath seriesPath, - QueryContext context, - Filter filter, - Set<String> allSensors) - throws IOException { - return FileLoaderUtils.loadTimeSeriesMetadata( - resource, seriesPath, context, filter, allSensors); - } - - protected List<TSDataType> getTsDataTypeList() { - return Collections.singletonList(dataType); - } - - protected IPointReader getPointReader(TsBlock tsBlock) { - return tsBlock.getTsBlockSingleColumnIterator(); - } - - Filter getGlobalTimeFilter() { - return scanOptions.getGlobalTimeFilter(); - } - - protected static class VersionPageReader { - - private final PriorityMergeReader.MergeReaderPriority version; - private final IPageReader data; - - private final boolean isSeq; - private final boolean isAligned; - private final boolean isMem; - - VersionPageReader(long version, long offset, IPageReader data, boolean isSeq) { - this.version = new PriorityMergeReader.MergeReaderPriority(version, offset); - this.data = data; - this.isSeq = isSeq; - this.isAligned = data instanceof IAlignedPageReader; - this.isMem = data instanceof MemPageReader || data instanceof MemAlignedPageReader; - } - - @SuppressWarnings("squid:S3740") - Statistics getStatistics() { - return data.getStatistics(); - } - - @SuppressWarnings("squid:S3740") - Statistics getStatistics(int index) throws IOException { - if (!(data instanceof IAlignedPageReader)) { - throw new IOException("Can only get statistics by index from AlignedPageReader"); - } - return ((IAlignedPageReader) data).getStatistics(index); - } - - @SuppressWarnings("squid:S3740") - Statistics getTimeStatistics() throws IOException { - if (!(data instanceof IAlignedPageReader)) { - throw new IOException("Can only get statistics of time column from AlignedPageReader"); - } - return ((IAlignedPageReader) data).getTimeStatistics(); - } - - TsBlock getAllSatisfiedPageData(boolean ascending) throws IOException { - long startTime = System.nanoTime(); - try { - TsBlock tsBlock = data.getAllSatisfiedData(); - if (!ascending) { - tsBlock.reverse(); - } - return tsBlock; - } finally { - SERIES_SCAN_COST_METRIC_SET.recordSeriesScanCost( - isAligned - ? (isMem - ? BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_MEM - : BUILD_TSBLOCK_FROM_PAGE_READER_ALIGNED_DISK) - : (isMem - ? BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_MEM - : BUILD_TSBLOCK_FROM_PAGE_READER_NONALIGNED_DISK), - System.nanoTime() - startTime); - } - } - - void setFilter(Filter filter) { - data.setFilter(filter); - } - - boolean isModified() { - return data.isModified(); - } - - public boolean isSeq() { - return isSeq; - } - - public void setLimitOffset(PaginationController paginationController) { - data.setLimitOffset(paginationController); - } - } - - public interface TimeOrderUtils { - - long getOrderTime(Statistics<? extends Object> statistics); - - long getOrderTime(TsFileResource fileResource); - - long getOverlapCheckTime(Statistics<? extends Object> range); - - boolean isOverlapped(Statistics<? extends Object> left, Statistics<? extends Object> right); - - boolean isOverlapped(long time, Statistics<? extends Object> right); - - boolean isOverlapped(long time, TsFileResource right); - - <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor); - - long getCurrentEndPoint(long time, Statistics<? extends Object> statistics); - - long getCurrentEndPoint( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics); - - boolean isExcessEndpoint(long time, long endpointTime); - - /** Return true if taking first page reader from seq readers */ - boolean isTakeSeqAsFirst( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics); - - boolean getAscending(); - - boolean hasNextSeqResource(); - - boolean hasNextUnseqResource(); - - TsFileResource getNextSeqFileResource(boolean isDelete); - - TsFileResource getNextUnseqFileResource(boolean isDelete); - - void setCurSeqFileIndex(QueryDataSource dataSource); - } - - class DescTimeOrderUtils implements TimeOrderUtils { - - @SuppressWarnings("squid:S3740") - @Override - public long getOrderTime(Statistics statistics) { - return statistics.getEndTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public long getOrderTime(TsFileResource fileResource) { - return fileResource.getEndTime(seriesPath.getDevice()); - } - - @SuppressWarnings("squid:S3740") - @Override - public long getOverlapCheckTime(Statistics range) { - return range.getStartTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public boolean isOverlapped(Statistics left, Statistics right) { - return left.getStartTime() <= right.getEndTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public boolean isOverlapped(long time, Statistics right) { - return time <= right.getEndTime(); - } - - @Override - public boolean isOverlapped(long time, TsFileResource right) { - return time <= right.getEndTime(seriesPath.getDevice()); - } - - @Override - public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) { - Objects.requireNonNull(keyExtractor); - return (Comparator<T> & Serializable) - (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c2), keyExtractor.applyAsLong(c1)); - } - - @Override - public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) { - return Math.max(time, statistics.getStartTime()); - } - - @Override - public long getCurrentEndPoint( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { - return Math.max(seqStatistics.getStartTime(), unseqStatistics.getStartTime()); - } - - @Override - public boolean isExcessEndpoint(long time, long endpointTime) { - return time < endpointTime; - } - - @Override - public boolean isTakeSeqAsFirst( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { - return seqStatistics.getEndTime() > unseqStatistics.getEndTime(); - } - - @Override - public boolean getAscending() { - return false; - } - - @Override - public boolean hasNextSeqResource() { - while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) { - TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); - if (tsFileResource != null - && tsFileResource.isSatisfied( - seriesPath.getDevice(), getGlobalTimeFilter(), true, false)) { - break; - } - curSeqFileIndex--; - } - return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending()); - } - - @Override - public boolean hasNextUnseqResource() { - while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) { - TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); - if (tsFileResource != null - && tsFileResource.isSatisfied( - seriesPath.getDevice(), getGlobalTimeFilter(), false, false)) { - break; - } - curUnseqFileIndex++; - } - return dataSource.hasNextUnseqResource(curUnseqFileIndex); - } - - @Override - public TsFileResource getNextSeqFileResource(boolean isDelete) { - TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); - if (isDelete) { - curSeqFileIndex--; - } - return tsFileResource; - } - - @Override - public TsFileResource getNextUnseqFileResource(boolean isDelete) { - TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); - if (isDelete) { - curUnseqFileIndex++; - } - return tsFileResource; - } - - @Override - public void setCurSeqFileIndex(QueryDataSource dataSource) { - curSeqFileIndex = dataSource.getSeqResourcesSize() - 1; - } - } - - class AscTimeOrderUtils implements TimeOrderUtils { - - @SuppressWarnings("squid:S3740") - @Override - public long getOrderTime(Statistics statistics) { - return statistics.getStartTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public long getOrderTime(TsFileResource fileResource) { - return fileResource.getStartTime(seriesPath.getDevice()); - } - - @SuppressWarnings("squid:S3740") - @Override - public long getOverlapCheckTime(Statistics range) { - return range.getEndTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public boolean isOverlapped(Statistics left, Statistics right) { - return left.getEndTime() >= right.getStartTime(); - } - - @SuppressWarnings("squid:S3740") - @Override - public boolean isOverlapped(long time, Statistics right) { - return time >= right.getStartTime(); - } - - @Override - public boolean isOverlapped(long time, TsFileResource right) { - return time >= right.getStartTime(seriesPath.getDevice()); - } - - @Override - public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) { - Objects.requireNonNull(keyExtractor); - return (Comparator<T> & Serializable) - (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c1), keyExtractor.applyAsLong(c2)); - } - - @Override - public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) { - return Math.min(time, statistics.getEndTime()); - } - - @Override - public long getCurrentEndPoint( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { - return Math.min(seqStatistics.getEndTime(), unseqStatistics.getEndTime()); - } - - @Override - public boolean isExcessEndpoint(long time, long endpointTime) { - return time > endpointTime; - } - - @Override - public boolean isTakeSeqAsFirst( - Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) { - return seqStatistics.getStartTime() < unseqStatistics.getStartTime(); - } - - @Override - public boolean getAscending() { - return true; - } - - @Override - public boolean hasNextSeqResource() { - while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) { - TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); - if (tsFileResource != null - && tsFileResource.isSatisfied( - seriesPath.getDevice(), getGlobalTimeFilter(), true, false)) { - break; - } - curSeqFileIndex++; - } - return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending()); - } - - @Override - public boolean hasNextUnseqResource() { - while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) { - TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); - if (tsFileResource != null - && tsFileResource.isSatisfied( - seriesPath.getDevice(), getGlobalTimeFilter(), false, false)) { - break; - } - curUnseqFileIndex++; - } - return dataSource.hasNextUnseqResource(curUnseqFileIndex); - } - - @Override - public TsFileResource getNextSeqFileResource(boolean isDelete) { - TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex); - if (isDelete) { - curSeqFileIndex++; - } - return tsFileResource; - } - - @Override - public TsFileResource getNextUnseqFileResource(boolean isDelete) { - TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex); - if (isDelete) { - curUnseqFileIndex++; - } - return tsFileResource; - } - - @Override - public void setCurSeqFileIndex(QueryDataSource dataSource) { - curSeqFileIndex = 0; - } - } -}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java index 822f34e..148dd59 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/reader/SeriesDataBlockReader.java
@@ -24,8 +24,8 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; -import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; -import org.apache.iotdb.db.queryengine.execution.operator.source.SeriesScanUtil; +import org.apache.iotdb.db.queryengine.execution.operator.scan.AlignedSeriesScanUtil; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/AlignedSeriesScanLimitOffsetPushDownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/AlignedSeriesScanLimitOffsetPushDownTest.java index dc54256..78a61e7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/AlignedSeriesScanLimitOffsetPushDownTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/AlignedSeriesScanLimitOffsetPushDownTest.java
@@ -22,7 +22,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.PartialPath; -import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; +import org.apache.iotdb.db.queryengine.execution.operator.scan.AlignedSeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.buffer.BloomFilterCache;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/SeriesScanLimitOffsetPushDownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/SeriesScanLimitOffsetPushDownTest.java index b95f844..a6e00c0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/SeriesScanLimitOffsetPushDownTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/series/SeriesScanLimitOffsetPushDownTest.java
@@ -22,7 +22,7 @@ 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.db.queryengine.execution.operator.source.SeriesScanUtil; +import org.apache.iotdb.db.queryengine.execution.operator.scan.SeriesScanUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;