This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch IOTDB-6141 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 810b9a88c7ec1f06f49483bb6f091b443ea4f0a3 Author: Jackie Tien <[email protected]> AuthorDate: Tue Oct 17 12:18:06 2023 +0800 [To rel/1.2] [IOTDB-6141] Optimize the large time range raw query performance --- .../execution/operator/AbstractOperator.java | 3 + .../operator/source/AlignedSeriesScanOperator.java | 12 +- .../tsfile/read/common/block/TsBlockBuilder.java | 2 +- .../block/column/Int32ArrayColumnEncoder.java | 20 +- .../read/common/block/column/TsBlockSerde.java | 7 +- .../tsfile/read/reader/page/AlignedPageReader.java | 201 ++++++++++++++------- .../tsfile/read/reader/page/ValuePageReader.java | 131 ++++++++++++++ .../read/reader/series/PaginationController.java | 8 + 8 files changed, 311 insertions(+), 73 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AbstractOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AbstractOperator.java index 9526f07339f..b062998368a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AbstractOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AbstractOperator.java @@ -75,6 +75,9 @@ public abstract class AbstractOperator implements Operator { public TsBlock getResultFromRetainedTsBlock() { TsBlock res; + if (maxTupleSizeOfTsBlock == -1) { + initializeMaxTsBlockLength(retainedTsBlock); + } if (retainedTsBlock.getPositionCount() - startOffset <= maxTupleSizeOfTsBlock) { res = retainedTsBlock.subTsBlock(startOffset); retainedTsBlock = null; 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 06e7953b715..c80bed6963d 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 @@ -35,6 +35,8 @@ import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder; import java.io.IOException; import java.util.concurrent.TimeUnit; +import static org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder.MAX_LINE_NUMBER; + public class AlignedSeriesScanOperator extends AbstractDataSourceOperator { private final TsBlockBuilder builder; @@ -100,9 +102,11 @@ public class AlignedSeriesScanOperator extends AbstractDataSourceOperator { break; } - } while (System.nanoTime() - start < maxRuntime && !builder.isFull()); + } while (System.nanoTime() - start < maxRuntime + && !builder.isFull() + && retainedTsBlock == null); - finished = builder.isEmpty(); + finished = (builder.isEmpty() && retainedTsBlock == null); return !finished; } catch (IOException e) { @@ -163,6 +167,10 @@ public class AlignedSeriesScanOperator extends AbstractDataSourceOperator { private void appendToBuilder(TsBlock tsBlock) { int size = tsBlock.getPositionCount(); + if (builder.isEmpty() && tsBlock.getPositionCount() >= MAX_LINE_NUMBER) { + retainedTsBlock = tsBlock; + return; + } TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder(); TimeColumn timeColumn = tsBlock.getTimeColumn(); for (int i = 0; i < size; i++) { diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java index 883e5a2133a..776ec1d55f5 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/TsBlockBuilder.java @@ -47,7 +47,7 @@ public class TsBlockBuilder { // This could be any other small number. private static final int DEFAULT_INITIAL_EXPECTED_ENTRIES = 8; - private static final int MAX_LINE_NUMBER = + public static final int MAX_LINE_NUMBER = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber(); private TimeColumnBuilder timeColumnBuilder; diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/Int32ArrayColumnEncoder.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/Int32ArrayColumnEncoder.java index 9abc4fca1e5..74486e7f92d 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/Int32ArrayColumnEncoder.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/Int32ArrayColumnEncoder.java @@ -80,14 +80,26 @@ public class Int32ArrayColumnEncoder implements ColumnEncoder { TSDataType dataType = column.getDataType(); int positionCount = column.getPositionCount(); if (TSDataType.INT32.equals(dataType)) { - for (int i = 0; i < positionCount; i++) { - if (!column.isNull(i)) { + if (column.mayHaveNull()) { + for (int i = 0; i < positionCount; i++) { + if (!column.isNull(i)) { + output.writeInt(column.getInt(i)); + } + } + } else { + for (int i = 0; i < positionCount; i++) { output.writeInt(column.getInt(i)); } } } else if (TSDataType.FLOAT.equals(dataType)) { - for (int i = 0; i < positionCount; i++) { - if (!column.isNull(i)) { + if (column.mayHaveNull()) { + for (int i = 0; i < positionCount; i++) { + if (!column.isNull(i)) { + output.writeInt(Float.floatToIntBits(column.getFloat(i))); + } + } + } else { + for (int i = 0; i < positionCount; i++) { output.writeInt(Float.floatToIntBits(column.getFloat(i))); } } diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TsBlockSerde.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TsBlockSerde.java index 16eaf040036..2faf4d0a81e 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TsBlockSerde.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TsBlockSerde.java @@ -87,7 +87,12 @@ public class TsBlockSerde { * @return Serialized tsblock. */ public ByteBuffer serialize(TsBlock tsBlock) throws IOException { - ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); + if (tsBlock.getRetainedSizeInBytes() > Integer.MAX_VALUE) { + throw new IllegalStateException( + "TsBlock should not be that large: " + tsBlock.getRetainedSizeInBytes()); + } + ByteArrayOutputStream byteArrayOutputStream = + new ByteArrayOutputStream((int) tsBlock.getRetainedSizeInBytes()); DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream); // Value column count. diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java index 67be092ada5..a2c8909dc19 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java @@ -165,95 +165,166 @@ public class AlignedPageReader implements IPageReader, IAlignedPageReader { return new LazyLoadAlignedPagePointReader(timePageReader, valuePageReaderList); } + // if any values of these queried measurements has the same value count as the time column + // and no mods file exist, we can go fast way + private boolean canGoFastWay() { + if (isModified) { + return false; + } + boolean res = getValueStatisticsList().isEmpty(); + long rowCount = getTimeStatistics().getCount(); + for (Statistics vStatistics : getValueStatisticsList()) { + if (vStatistics != null && !vStatistics.hasNullValue(rowCount)) { + res = true; + break; + } + } + return res; + } + @Override public TsBlock getAllSatisfiedData() throws IOException { - builder.reset(); if (!pageSatisfy()) { return builder.build(); } long[] timeBatch = timePageReader.getNextTimeBatch(); - // if all the sub sensors' value are null in current row, just discard it - // if !filter.satisfy, discard this row - boolean[] keepCurrentRow = new boolean[timeBatch.length]; - if (filter == null) { - Arrays.fill(keepCurrentRow, true); - } else { - for (int i = 0, n = timeBatch.length; i < n; i++) { - keepCurrentRow[i] = filter.satisfy(timeBatch[i], null); - } - } - - boolean[][] isDeleted = null; - if (valueCount != 0) { - // using bitMap in valuePageReaders to indicate whether columns of current row are all null. - byte[] bitmask = new byte[(timeBatch.length - 1) / 8 + 1]; - Arrays.fill(bitmask, (byte) 0x00); - isDeleted = new boolean[valueCount][timeBatch.length]; - for (int columnIndex = 0; columnIndex < valueCount; columnIndex++) { - ValuePageReader pageReader = valuePageReaderList.get(columnIndex); - if (pageReader != null) { - byte[] bitmap = pageReader.getBitmap(); - pageReader.fillIsDeleted(timeBatch, isDeleted[columnIndex]); + if (canGoFastWay()) { + // skip all the page + if (paginationController.hasCurOffset(timeBatch.length)) { + paginationController.consumeOffset(timeBatch.length); + } else { + int readStartIndex = + paginationController.hasCurOffset() ? (int) paginationController.getCurOffset() : 0; + // consume the remaining offset + paginationController.consumeOffset(readStartIndex); + // not included + int readEndIndex = + (paginationController.hasCurLimit() && paginationController.getCurLimit() > 0) + && (paginationController.getCurLimit() < timeBatch.length - readStartIndex + 1) + ? readStartIndex + (int) paginationController.getCurLimit() + : timeBatch.length; + if (paginationController.hasCurLimit() && paginationController.getCurLimit() > 0) { + paginationController.consumeLimit((long) readEndIndex - readStartIndex); + } - for (int i = 0, n = isDeleted[columnIndex].length; i < n; i++) { - if (isDeleted[columnIndex][i]) { - int shift = i % 8; - bitmap[i / 8] = (byte) (bitmap[i / 8] & (~(MASK >>> shift))); + boolean[] keepCurrentRow = new boolean[readEndIndex - readStartIndex]; + if (filter == null) { + Arrays.fill(keepCurrentRow, true); + // construct time column + for (int i = readStartIndex; i < readEndIndex; i++) { + builder.getTimeColumnBuilder().writeLong(timeBatch[i]); + builder.declarePosition(); + } + } else { + for (int i = readStartIndex; i < readEndIndex; i++) { + keepCurrentRow[i - readStartIndex] = filter.satisfy(timeBatch[i], null); + // construct time column + if (keepCurrentRow[i - readStartIndex]) { + builder.getTimeColumnBuilder().writeLong(timeBatch[i]); + builder.declarePosition(); } } - for (int i = 0, n = bitmask.length; i < n; i++) { - bitmask[i] = (byte) (bitmap[i] | bitmask[i]); + } + + // construct value columns + for (int i = 0; i < valueCount; i++) { + ValuePageReader pageReader = valuePageReaderList.get(i); + if (pageReader != null) { + pageReader.writeColumnBuilderWithNextBatch( + readStartIndex, readEndIndex, builder.getColumnBuilder(i), keepCurrentRow); + } else { + for (int j = readStartIndex; j < readEndIndex; j++) { + if (keepCurrentRow[j - readStartIndex]) { + builder.getColumnBuilder(i).appendNull(); + } + } } } } + } else { + // if all the sub sensors' value are null in current row, just discard it + // if !filter.satisfy, discard this row + boolean[] keepCurrentRow = new boolean[timeBatch.length]; + if (filter == null) { + Arrays.fill(keepCurrentRow, true); + } else { + for (int i = 0, n = timeBatch.length; i < n; i++) { + keepCurrentRow[i] = filter.satisfy(timeBatch[i], null); + } + } + + boolean[][] isDeleted = null; + if (valueCount != 0) { + // using bitMap in valuePageReaders to indicate whether columns of current row are all null. + byte[] bitmask = new byte[(timeBatch.length - 1) / 8 + 1]; + Arrays.fill(bitmask, (byte) 0x00); + isDeleted = new boolean[valueCount][timeBatch.length]; + for (int columnIndex = 0; columnIndex < valueCount; columnIndex++) { + ValuePageReader pageReader = valuePageReaderList.get(columnIndex); + if (pageReader != null) { + byte[] bitmap = pageReader.getBitmap(); + pageReader.fillIsDeleted(timeBatch, isDeleted[columnIndex]); - for (int i = 0, n = bitmask.length; i < n; i++) { - if (bitmask[i] == (byte) 0xFF) { - // 8 rows are not all null, do nothing - } else if (bitmask[i] == (byte) 0x00) { - for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) { - keepCurrentRow[i * 8 + j] = false; + for (int i = 0, n = isDeleted[columnIndex].length; i < n; i++) { + if (isDeleted[columnIndex][i]) { + int shift = i % 8; + bitmap[i / 8] = (byte) (bitmap[i / 8] & (~(MASK >>> shift))); + } + } + for (int i = 0, n = bitmask.length; i < n; i++) { + bitmask[i] = (byte) (bitmap[i] | bitmask[i]); + } } - } else { - for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) { - if (((bitmask[i] & 0xFF) & (MASK >>> j)) == 0) { + } + + for (int i = 0, n = bitmask.length; i < n; i++) { + if (bitmask[i] == (byte) 0xFF) { + // 8 rows are not all null, do nothing + } else if (bitmask[i] == (byte) 0x00) { + for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) { keepCurrentRow[i * 8 + j] = false; } + } else { + for (int j = 0; j < 8 && (i * 8 + j < keepCurrentRow.length); j++) { + if (((bitmask[i] & 0xFF) & (MASK >>> j)) == 0) { + keepCurrentRow[i * 8 + j] = false; + } + } } } } - } - // construct time column - int readEndIndex = timeBatch.length; - for (int i = 0; i < timeBatch.length; i++) { - if (keepCurrentRow[i]) { - if (paginationController.hasCurOffset()) { - paginationController.consumeOffset(); - keepCurrentRow[i] = false; - } else if (paginationController.hasCurLimit()) { - builder.getTimeColumnBuilder().writeLong(timeBatch[i]); - builder.declarePosition(); - paginationController.consumeLimit(); - } else { - readEndIndex = i; - break; + // construct time column + int readEndIndex = timeBatch.length; + for (int i = 0; i < timeBatch.length; i++) { + if (keepCurrentRow[i]) { + if (paginationController.hasCurOffset()) { + paginationController.consumeOffset(); + keepCurrentRow[i] = false; + } else if (paginationController.hasCurLimit()) { + builder.getTimeColumnBuilder().writeLong(timeBatch[i]); + builder.declarePosition(); + paginationController.consumeLimit(); + } else { + readEndIndex = i; + break; + } } } - } - // construct value columns - for (int i = 0; i < valueCount; i++) { - ValuePageReader pageReader = valuePageReaderList.get(i); - if (pageReader != null) { - pageReader.writeColumnBuilderWithNextBatch( - readEndIndex, builder.getColumnBuilder(i), keepCurrentRow, isDeleted[i]); - } else { - for (int j = 0; j < readEndIndex; j++) { - if (keepCurrentRow[j]) { - builder.getColumnBuilder(i).appendNull(); + // construct value columns + for (int i = 0; i < valueCount; i++) { + ValuePageReader pageReader = valuePageReaderList.get(i); + if (pageReader != null) { + pageReader.writeColumnBuilderWithNextBatch( + readEndIndex, builder.getColumnBuilder(i), keepCurrentRow, isDeleted[i]); + } else { + for (int j = 0; j < readEndIndex; j++) { + if (keepCurrentRow[j]) { + builder.getColumnBuilder(i).appendNull(); + } } } } @@ -316,6 +387,6 @@ public class AlignedPageReader implements IPageReader, IAlignedPageReader { @Override public void initTsBlockBuilder(List<TSDataType> dataTypes) { - builder = new TsBlockBuilder(dataTypes); + builder = new TsBlockBuilder((int) timePageReader.getStatistics().getCount(), dataTypes); } } diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/ValuePageReader.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/ValuePageReader.java index 79e923c97f0..c158e3077eb 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/ValuePageReader.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/ValuePageReader.java @@ -329,6 +329,137 @@ public class ValuePageReader { } } + public void writeColumnBuilderWithNextBatch( + int readStartIndex, int readEndIndex, ColumnBuilder columnBuilder, boolean[] satisfied) { + if (valueBuffer == null) { + for (int i = readStartIndex; i < readEndIndex; i++) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + } + return; + } + + switch (dataType) { + case BOOLEAN: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readBoolean(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + boolean aBoolean = valueDecoder.readBoolean(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeBoolean(aBoolean); + } + } + break; + case INT32: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readInt(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + int aInt = valueDecoder.readInt(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeInt(aInt); + } + } + break; + case INT64: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readLong(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + long aLong = valueDecoder.readLong(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeLong(aLong); + } + } + break; + case FLOAT: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readFloat(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + float aFloat = valueDecoder.readFloat(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeFloat(aFloat); + } + } + break; + case DOUBLE: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readDouble(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + double aDouble = valueDecoder.readDouble(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeDouble(aDouble); + } + } + break; + case TEXT: + // skip useless data + for (int i = 0; i < readStartIndex; i++) { + valueDecoder.readBinary(valueBuffer); + } + + for (int i = readStartIndex; i < readEndIndex; i++) { + if (((bitmap[i / 8] & 0xFF) & (MASK >>> (i % 8))) == 0) { + if (satisfied[i - readStartIndex]) { + columnBuilder.appendNull(); + } + continue; + } + Binary aBinary = valueDecoder.readBinary(valueBuffer); + if (satisfied[i - readStartIndex]) { + columnBuilder.writeBinary(aBinary); + } + } + break; + default: + throw new UnSupportedDataTypeException(String.valueOf(dataType)); + } + } + public Statistics getStatistics() { return pageHeader.getStatistics(); } diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/series/PaginationController.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/series/PaginationController.java index 108fa0d0a5b..aa0f7971e47 100644 --- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/series/PaginationController.java +++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/series/PaginationController.java @@ -60,6 +60,14 @@ public class PaginationController { curOffset--; } + public long getCurOffset() { + return curOffset; + } + + public long getCurLimit() { + return curLimit; + } + public void consumeLimit() { if (hasLimit) { curLimit--;
