This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new b8825b18c Add buffered TsFile input (#868)
b8825b18c is described below
commit b8825b18ce296b6af31aabd1c7727f9007b72652
Author: Jiang Tian <[email protected]>
AuthorDate: Mon Jul 20 12:26:54 2026 +0800
Add buffered TsFile input (#868)
* Add buffered TsFile input
* add a Constructor for PublicBAOS
---
.../tsfile/read/reader/BufferedTsFileInput.java | 203 +++++++++++++
.../tsfile/read/reader/LocalTsFileInput.java | 4 +
.../java/org/apache/tsfile/utils/PublicBAOS.java | 4 +
.../read/TsFileSequenceReaderPerformanceTest.java | 331 +++++++++++++++++++++
.../read/reader/BufferedTsFileInputTest.java | 145 +++++++++
5 files changed, 687 insertions(+)
diff --git
a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/BufferedTsFileInput.java
b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/BufferedTsFileInput.java
new file mode 100644
index 000000000..bea94fed6
--- /dev/null
+++
b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/BufferedTsFileInput.java
@@ -0,0 +1,203 @@
+/*
+ * 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.tsfile.read.reader;
+
+import org.apache.tsfile.i18n.Messages;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.file.Path;
+import java.util.Objects;
+
+/** A local TsFile input that caches file read results in a {@link
ByteBuffer}. */
+public class BufferedTsFileInput extends LocalTsFileInput {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8 * 1024;
+
+ private final ByteBuffer buffer;
+ private long bufferStartPosition = -1;
+ private long logicalPosition;
+
+ public BufferedTsFileInput(Path file) throws IOException {
+ this(file, DEFAULT_BUFFER_SIZE);
+ }
+
+ public BufferedTsFileInput(Path file, int bufferSize) throws IOException {
+ super(validateBufferSize(file, bufferSize));
+ buffer = ByteBuffer.allocate(bufferSize);
+ buffer.limit(0);
+ }
+
+ private static Path validateBufferSize(Path file, int bufferSize) {
+ if (bufferSize <= 0) {
+ throw new IllegalArgumentException(
+ Messages.get("error.utils.buffer_size_not_positive_input"));
+ }
+ return file;
+ }
+
+ @Override
+ public long position() throws IOException {
+ ensureOpen();
+ return logicalPosition;
+ }
+
+ @Override
+ public BufferedTsFileInput position(long newPosition) throws IOException {
+ ensureOpen();
+ if (newPosition < 0) {
+ throw new IllegalArgumentException();
+ }
+ logicalPosition = newPosition;
+ return this;
+ }
+
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ int readSize = read(dst, logicalPosition);
+ if (readSize > 0) {
+ logicalPosition += readSize;
+ }
+ return readSize;
+ }
+
+ @Override
+ public int read(ByteBuffer dst, long position) throws IOException {
+ Objects.requireNonNull(dst);
+ ensureOpen();
+ if (position < 0) {
+ throw new IllegalArgumentException();
+ }
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
+
+ int totalReadSize = 0;
+ long currentPosition = position;
+ while (dst.hasRemaining()) {
+ if (isInBuffer(currentPosition)) {
+ int copiedSize = copyFromBuffer(dst, currentPosition);
+ totalReadSize += copiedSize;
+ currentPosition += copiedSize;
+ continue;
+ }
+
+ if (dst.remaining() >= buffer.capacity()) {
+ int readSize = super.read(dst, currentPosition);
+ if (readSize <= 0) {
+ return totalReadSize == 0 ? readSize : totalReadSize;
+ }
+ return totalReadSize + readSize;
+ }
+
+ int readSize = fillBuffer(currentPosition);
+ if (readSize <= 0) {
+ return totalReadSize == 0 ? readSize : totalReadSize;
+ }
+ }
+ return totalReadSize;
+ }
+
+ private boolean isInBuffer(long position) {
+ return position >= bufferStartPosition && position - bufferStartPosition <
buffer.limit();
+ }
+
+ private int copyFromBuffer(ByteBuffer dst, long position) {
+ int bufferOffset = (int) (position - bufferStartPosition);
+ int copiedSize = Math.min(dst.remaining(), buffer.limit() - bufferOffset);
+ ByteBuffer source = buffer.asReadOnlyBuffer();
+ source.position(bufferOffset);
+ source.limit(bufferOffset + copiedSize);
+ dst.put(source);
+ return copiedSize;
+ }
+
+ private int fillBuffer(long position) throws IOException {
+ buffer.clear();
+ int readSize = super.read(buffer, position);
+ buffer.flip();
+ bufferStartPosition = position;
+ return readSize;
+ }
+
+ @Override
+ public InputStream wrapAsInputStream() {
+ return new BufferedTsFileInputStream();
+ }
+
+ @Override
+ public void close() throws IOException {
+ buffer.limit(0);
+ super.close();
+ }
+
+ private void ensureOpen() throws ClosedChannelException {
+ if (!isOpen()) {
+ throw new ClosedChannelException();
+ }
+ }
+
+ private class BufferedTsFileInputStream extends InputStream {
+
+ private final ByteBuffer oneByteBuffer = ByteBuffer.allocate(Byte.BYTES);
+
+ @Override
+ public int read() throws IOException {
+ oneByteBuffer.clear();
+ int readSize = BufferedTsFileInput.this.read(oneByteBuffer);
+ if (readSize < 0) {
+ return -1;
+ }
+ oneByteBuffer.flip();
+ return oneByteBuffer.get() & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] bytes, int offset, int length) throws IOException {
+ return BufferedTsFileInput.this.read(ByteBuffer.wrap(bytes, offset,
length));
+ }
+
+ @Override
+ public long skip(long skippedBytes) throws IOException {
+ if (skippedBytes <= 0) {
+ return 0;
+ }
+ long currentPosition = BufferedTsFileInput.this.position();
+ long remainingSize = Math.max(0, BufferedTsFileInput.this.size() -
currentPosition);
+ long actualSkippedBytes = Math.min(skippedBytes, remainingSize);
+ BufferedTsFileInput.this.position(currentPosition + actualSkippedBytes);
+ return actualSkippedBytes;
+ }
+
+ @Override
+ public int available() throws IOException {
+ long remainingSize =
+ Math.max(0, BufferedTsFileInput.this.size() -
BufferedTsFileInput.this.position());
+ return (int) Math.min(remainingSize, Integer.MAX_VALUE);
+ }
+
+ @Override
+ public void close() throws IOException {
+ BufferedTsFileInput.this.close();
+ }
+ }
+}
diff --git
a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/LocalTsFileInput.java
b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/LocalTsFileInput.java
index 9692c28cf..8ed39e3e9 100644
---
a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/LocalTsFileInput.java
+++
b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/LocalTsFileInput.java
@@ -45,6 +45,10 @@ public class LocalTsFileInput implements TsFileInput {
filePath = file.toString();
}
+ protected boolean isOpen() {
+ return channel.isOpen();
+ }
+
@Override
public long size() throws IOException {
try {
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/utils/PublicBAOS.java
b/java/tsfile/src/main/java/org/apache/tsfile/utils/PublicBAOS.java
index fc3928b52..3d97cb0e7 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/utils/PublicBAOS.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/PublicBAOS.java
@@ -43,6 +43,10 @@ public class PublicBAOS extends ByteArrayOutputStream {
super(size);
}
+ public PublicBAOS(byte[] buf) {
+ this.buf = buf;
+ }
+
private void ensureCapacity(int minCapacity) {
// overflow-conscious code
if (minCapacity - buf.length > 0) {
diff --git
a/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileSequenceReaderPerformanceTest.java
b/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileSequenceReaderPerformanceTest.java
new file mode 100644
index 000000000..a213daa9d
--- /dev/null
+++
b/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileSequenceReaderPerformanceTest.java
@@ -0,0 +1,331 @@
+/*
+ * 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.tsfile.read;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.common.conf.TSFileDescriptor;
+import org.apache.tsfile.encoding.decoder.Decoder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.external.commons.io.FileUtils;
+import org.apache.tsfile.file.MetaMarker;
+import org.apache.tsfile.file.header.ChunkHeader;
+import org.apache.tsfile.file.header.PageHeader;
+import org.apache.tsfile.file.metadata.ChunkMetadata;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.common.BatchData;
+import org.apache.tsfile.read.reader.BufferedTsFileInput;
+import org.apache.tsfile.read.reader.LocalTsFileInput;
+import org.apache.tsfile.read.reader.TsFileInput;
+import org.apache.tsfile.read.reader.page.PageReader;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class TsFileSequenceReaderPerformanceTest {
+
+ private static final int FILE_COUNT = 100;
+ private static final int DEVICE_COUNT = 100;
+ private static final int POINTS_PER_DEVICE = 100_000;
+ private static final int[] POINT_COUNTS_PER_CHUNK = {100, 1_000, 10_000};
+ private static final int LOCAL_BUFFER_SIZE = 0;
+ private static final int[] BUFFER_SIZES = {4 * 1024, 8 * 1024, 16 * 1024, 32
* 1024, 64 * 1024};
+ private static final String RUN_PERFORMANCE_TEST_PROPERTY =
"tsfile.runPerformanceTests";
+ private static final File TEST_DIRECTORY =
+ new File("target", "tsfile-sequence-reader-performance");
+
+ @Test
+ public void testReadTsFilesWithDifferentBufferAndChunkSizes() throws
Exception {
+ Assume.assumeTrue(
+ "Set -Dtsfile.runPerformanceTests=true to run the performance test",
+ Boolean.getBoolean(RUN_PERFORMANCE_TEST_PROPERTY));
+ TSFileConfig config = TSFileDescriptor.getInstance().getConfig();
+ int previousMaxPointsInPage = config.getMaxNumberOfPointsInPage();
+
+ try {
+ for (int pointsPerChunk : POINT_COUNTS_PER_CHUNK) {
+ assertEquals(0, POINTS_PER_DEVICE % pointsPerChunk);
+ int measurementCount = POINTS_PER_DEVICE / pointsPerChunk;
+ config.setMaxNumberOfPointsInPage(pointsPerChunk);
+ List<File> files = generateTsFiles(measurementCount, pointsPerChunk);
+
+ List<ReadConfiguration> configurations = createReadConfigurations();
+ for (int fileIndex = 0; fileIndex < files.size(); fileIndex++) {
+ File file = files.get(fileIndex);
+ int firstConfigurationIndex = fileIndex % configurations.size();
+ for (int offset = 0; offset < configurations.size(); offset++) {
+ ReadConfiguration configuration =
+ configurations.get((firstConfigurationIndex + offset) %
configurations.size());
+ configuration.elapsedNanos +=
+ readTsFile(
+ file,
+ configuration.bufferSize,
+ configuration.statistics,
+ measurementCount,
+ pointsPerChunk);
+ }
+ }
+
+ long expectedDeviceCount = (long) FILE_COUNT * DEVICE_COUNT;
+ long expectedChunkCount = expectedDeviceCount * measurementCount;
+ long expectedPointCount = expectedChunkCount * pointsPerChunk;
+ System.out.printf(
+ "Chunk scenario: points per chunk=%,d, measurements per
device=%,d%n",
+ pointsPerChunk, measurementCount);
+ for (ReadConfiguration configuration : configurations) {
+ assertStatistics(
+ configuration.statistics,
+ expectedDeviceCount,
+ expectedChunkCount,
+ expectedPointCount);
+ printStatistics(
+ configuration.inputName, configuration.statistics,
configuration.elapsedNanos);
+ }
+
+ ReadConfiguration localConfiguration = configurations.get(0);
+ for (int configurationIndex = 1;
+ configurationIndex < configurations.size();
+ configurationIndex++) {
+ ReadConfiguration bufferedConfiguration =
configurations.get(configurationIndex);
+ System.out.printf(
+ "%s vs LocalTsFileInput: speedup=%.3fx, time
reduction=%+.2f%%%n",
+ bufferedConfiguration.inputName,
+ localConfiguration.elapsedNanos / (double)
bufferedConfiguration.elapsedNanos,
+ (localConfiguration.elapsedNanos -
bufferedConfiguration.elapsedNanos)
+ * 100.0
+ / localConfiguration.elapsedNanos);
+ }
+ }
+ } finally {
+ config.setMaxNumberOfPointsInPage(previousMaxPointsInPage);
+ FileUtils.deleteDirectory(TEST_DIRECTORY);
+ }
+ }
+
+ private List<File> generateTsFiles(int measurementCount, int pointsPerChunk)
throws Exception {
+ FileUtils.deleteDirectory(TEST_DIRECTORY);
+ Files.createDirectories(TEST_DIRECTORY.toPath());
+
+ List<IMeasurementSchema> schemas =
createMeasurementSchemas(measurementCount);
+ Tablet tablet = createTablet(schemas, pointsPerChunk);
+ List<File> files = new ArrayList<>(FILE_COUNT);
+ for (int fileIndex = 0; fileIndex < FILE_COUNT; fileIndex++) {
+ File file = new File(TEST_DIRECTORY, "sequence-reader-" + fileIndex +
".tsfile");
+ generateTsFile(file, schemas, tablet);
+ files.add(file);
+ }
+ return files;
+ }
+
+ private List<ReadConfiguration> createReadConfigurations() {
+ List<ReadConfiguration> configurations = new
ArrayList<>(BUFFER_SIZES.length + 1);
+ configurations.add(new ReadConfiguration("LocalTsFileInput",
LOCAL_BUFFER_SIZE));
+ for (int bufferSize : BUFFER_SIZES) {
+ configurations.add(
+ new ReadConfiguration(
+ String.format("BufferedTsFileInput(%d KiB)", bufferSize / 1024),
bufferSize));
+ }
+ return configurations;
+ }
+
+ private List<IMeasurementSchema> createMeasurementSchemas(int
measurementCount) {
+ List<IMeasurementSchema> schemas = new ArrayList<>(measurementCount);
+ for (int measurementIndex = 0; measurementIndex < measurementCount;
measurementIndex++) {
+ schemas.add(
+ new MeasurementSchema(
+ "s" + measurementIndex, TSDataType.INT64, TSEncoding.TS_2DIFF,
CompressionType.LZ4));
+ }
+ return schemas;
+ }
+
+ private Tablet createTablet(List<IMeasurementSchema> schemas, int
pointsPerChunk) {
+ Tablet tablet = new Tablet(null, schemas, pointsPerChunk);
+ for (int pointIndex = 0; pointIndex < pointsPerChunk; pointIndex++) {
+ tablet.addTimestamp(pointIndex, pointIndex);
+ }
+ for (int measurementIndex = 0; measurementIndex < schemas.size();
measurementIndex++) {
+ for (int pointIndex = 0; pointIndex < pointsPerChunk; pointIndex++) {
+ tablet.addValue(pointIndex, measurementIndex, (long) pointIndex);
+ }
+ }
+ return tablet;
+ }
+
+ private void generateTsFile(File file, List<IMeasurementSchema> schemas,
Tablet tablet)
+ throws Exception {
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ for (int deviceIndex = 0; deviceIndex < DEVICE_COUNT; deviceIndex++) {
+ String device = "root.performance.d" + deviceIndex;
+ IDeviceID deviceID = IDeviceID.Factory.DEFAULT_FACTORY.create(device);
+ for (IMeasurementSchema schema : schemas) {
+ writer.registerTimeseries(deviceID, schema);
+ }
+ tablet.setDeviceId(device);
+ writer.writeTree(tablet);
+ }
+ }
+ }
+
+ private long readTsFile(
+ File file,
+ int bufferSize,
+ ReadStatistics statistics,
+ int measurementCount,
+ int pointsPerChunk)
+ throws IOException {
+ long startTime = System.nanoTime();
+ TsFileInput input =
+ bufferSize == LOCAL_BUFFER_SIZE
+ ? new LocalTsFileInput(file.toPath())
+ : new BufferedTsFileInput(file.toPath(), bufferSize);
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(input)) {
+ reader.position(TSFileConfig.MAGIC_STRING.getBytes().length + 1L);
+ byte marker;
+ while ((marker = reader.readMarker()) != MetaMarker.SEPARATOR) {
+ switch (marker) {
+ case MetaMarker.CHUNK_HEADER:
+ case MetaMarker.TIME_CHUNK_HEADER:
+ case MetaMarker.VALUE_CHUNK_HEADER:
+ case MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER:
+ case MetaMarker.ONLY_ONE_PAGE_TIME_CHUNK_HEADER:
+ case MetaMarker.ONLY_ONE_PAGE_VALUE_CHUNK_HEADER:
+ readChunk(reader, marker, statistics, pointsPerChunk);
+ break;
+ case MetaMarker.CHUNK_GROUP_HEADER:
+ reader.readChunkGroupHeader();
+ break;
+ case MetaMarker.OPERATION_INDEX_RANGE:
+ reader.readPlanIndex();
+ break;
+ default:
+ MetaMarker.handleUnexpectedMarker(marker);
+ }
+ }
+ verifyMetadata(reader, statistics, measurementCount, pointsPerChunk);
+ }
+ return System.nanoTime() - startTime;
+ }
+
+ private void assertStatistics(
+ ReadStatistics statistics,
+ long expectedDeviceCount,
+ long expectedChunkCount,
+ long expectedPointCount) {
+ assertEquals(expectedDeviceCount, statistics.deviceCount);
+ assertEquals(expectedChunkCount, statistics.chunkCount);
+ assertEquals(expectedPointCount, statistics.pointCount);
+ }
+
+ private void printStatistics(String inputName, ReadStatistics statistics,
long elapsedNanos) {
+ System.out.printf(
+ "%s sequentially read %d TsFiles: devices=%,d, chunks=%,d, points=%,d,
total time=%d ms (%.3f s)%n",
+ inputName,
+ FILE_COUNT,
+ statistics.deviceCount,
+ statistics.chunkCount,
+ statistics.pointCount,
+ TimeUnit.NANOSECONDS.toMillis(elapsedNanos),
+ elapsedNanos / 1_000_000_000.0);
+ }
+
+ private void readChunk(
+ TsFileSequenceReader reader, byte marker, ReadStatistics statistics, int
pointsPerChunk)
+ throws IOException {
+ ChunkHeader chunkHeader = reader.readChunkHeader(marker);
+ int remainingDataSize = chunkHeader.getDataSize();
+ long pointsInChunk = 0;
+ while (remainingDataSize > 0) {
+ boolean hasStatistics = (chunkHeader.getChunkType() & 0x3F) ==
MetaMarker.CHUNK_HEADER;
+ PageHeader pageHeader = reader.readPageHeader(chunkHeader.getDataType(),
hasStatistics);
+ ByteBuffer pageData = reader.readPage(pageHeader,
chunkHeader.getCompressionType());
+ Decoder valueDecoder =
+ Decoder.getDecoderByType(chunkHeader.getEncodingType(),
chunkHeader.getDataType());
+ Decoder timeDecoder =
+ Decoder.getDecoderByType(
+
TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder()),
+ TSDataType.INT64);
+ BatchData batchData =
+ new PageReader(pageData, chunkHeader.getDataType(), valueDecoder,
timeDecoder)
+ .getAllSatisfiedPageData();
+ pointsInChunk += batchData.length();
+ remainingDataSize -= pageHeader.getSerializedPageSize();
+ }
+
+ assertEquals(pointsPerChunk, pointsInChunk);
+ statistics.chunkCount++;
+ statistics.pointCount += pointsInChunk;
+ }
+
+ private void verifyMetadata(
+ TsFileSequenceReader reader,
+ ReadStatistics statistics,
+ int measurementCount,
+ int pointsPerChunk)
+ throws IOException {
+ List<IDeviceID> devices = reader.getAllDevices();
+ assertEquals(DEVICE_COUNT, devices.size());
+ for (IDeviceID device : devices) {
+ Map<String, List<ChunkMetadata>> metadataByMeasurement =
+ reader.readChunkMetadataInDevice(device);
+ assertEquals(measurementCount, metadataByMeasurement.size());
+ for (List<ChunkMetadata> chunkMetadata : metadataByMeasurement.values())
{
+ assertEquals(1, chunkMetadata.size());
+ assertEquals(pointsPerChunk, chunkMetadata.get(0).getNumOfPoints());
+ }
+ }
+ statistics.deviceCount += devices.size();
+ }
+
+ private static class ReadStatistics {
+
+ private long deviceCount;
+ private long chunkCount;
+ private long pointCount;
+ }
+
+ private static class ReadConfiguration {
+
+ private final String inputName;
+ private final int bufferSize;
+ private final ReadStatistics statistics = new ReadStatistics();
+ private long elapsedNanos;
+
+ private ReadConfiguration(String inputName, int bufferSize) {
+ this.inputName = inputName;
+ this.bufferSize = bufferSize;
+ }
+ }
+}
diff --git
a/java/tsfile/src/test/java/org/apache/tsfile/read/reader/BufferedTsFileInputTest.java
b/java/tsfile/src/test/java/org/apache/tsfile/read/reader/BufferedTsFileInputTest.java
new file mode 100644
index 000000000..679e10a91
--- /dev/null
+++
b/java/tsfile/src/test/java/org/apache/tsfile/read/reader/BufferedTsFileInputTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.tsfile.read.reader;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.file.Files;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+public class BufferedTsFileInputTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private File file;
+ private byte[] data;
+
+ @Before
+ public void setUp() throws Exception {
+ file = temporaryFolder.newFile("buffered-input.tsfile");
+ data = new byte[32];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ Files.write(file.toPath(), data);
+ }
+
+ @Test
+ public void testSequentialAndPositionedRead() throws Exception {
+ BufferedTsFileInput input = new BufferedTsFileInput(file.toPath(), 8);
+ try {
+ assertEquals(data.length, input.size());
+ assertEquals(file.toPath().toString(), input.getFilePath());
+ assertEquals(0, input.position());
+
+ ByteBuffer firstRead = ByteBuffer.allocate(5);
+ assertEquals(5, input.read(firstRead));
+ assertBufferEquals(new byte[] {0, 1, 2, 3, 4}, firstRead);
+ assertEquals(5, input.position());
+
+ ByteBuffer positionedRead = ByteBuffer.allocate(6);
+ assertEquals(6, input.read(positionedRead, 2));
+ assertBufferEquals(new byte[] {2, 3, 4, 5, 6, 7}, positionedRead);
+ assertEquals(5, input.position());
+
+ ByteBuffer crossBufferRead = ByteBuffer.allocate(10);
+ assertEquals(10, input.read(crossBufferRead));
+ assertBufferEquals(new byte[] {5, 6, 7, 8, 9, 10, 11, 12, 13, 14},
crossBufferRead);
+ assertEquals(15, input.position());
+
+ input.position(29);
+ ByteBuffer endRead = ByteBuffer.allocate(8);
+ assertEquals(3, input.read(endRead));
+ assertBufferEquals(new byte[] {29, 30, 31}, endRead);
+ assertEquals(32, input.position());
+ assertEquals(-1, input.read(ByteBuffer.allocate(1)));
+ assertEquals(0, input.read(ByteBuffer.allocate(0)));
+ } finally {
+ input.close();
+ }
+ }
+
+ @Test
+ public void testReadLargerThanBuffer() throws Exception {
+ BufferedTsFileInput input = new BufferedTsFileInput(file.toPath(), 4);
+ try {
+ ByteBuffer destination = ByteBuffer.allocate(20);
+ assertEquals(20, input.read(destination, 3));
+ byte[] expected = new byte[20];
+ System.arraycopy(data, 3, expected, 0, expected.length);
+ assertBufferEquals(expected, destination);
+ assertEquals(0, input.position());
+ } finally {
+ input.close();
+ }
+ }
+
+ @Test
+ public void testInputStreamSharesPosition() throws Exception {
+ BufferedTsFileInput input = new BufferedTsFileInput(file.toPath(), 4);
+ input.position(3);
+ try (InputStream stream = input.wrapAsInputStream()) {
+ assertEquals(3, stream.read());
+ assertEquals(4, input.position());
+
+ byte[] bytes = new byte[7];
+ assertEquals(7, stream.read(bytes));
+ assertArrayEquals(new byte[] {4, 5, 6, 7, 8, 9, 10}, bytes);
+ assertEquals(11, input.position());
+
+ assertEquals(5, stream.skip(5));
+ assertEquals(16, input.position());
+ assertEquals(16, stream.available());
+ assertEquals(16, stream.read());
+ }
+ assertThrows(ClosedChannelException.class, input::position);
+ }
+
+ @Test
+ public void testInvalidArguments() throws Exception {
+ assertThrows(IllegalArgumentException.class, () -> new
BufferedTsFileInput(file.toPath(), 0));
+ assertThrows(IllegalArgumentException.class, () -> new
BufferedTsFileInput(file.toPath(), -1));
+
+ BufferedTsFileInput input = new BufferedTsFileInput(file.toPath());
+ try {
+ assertThrows(IllegalArgumentException.class, () -> input.position(-1));
+ assertThrows(IllegalArgumentException.class, () ->
input.read(ByteBuffer.allocate(1), -1));
+ } finally {
+ input.close();
+ }
+ }
+
+ private void assertBufferEquals(byte[] expected, ByteBuffer actual) {
+ actual.flip();
+ byte[] actualBytes = new byte[actual.remaining()];
+ actual.get(actualBytes);
+ assertArrayEquals(expected, actualBytes);
+ }
+}
