This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_buffered_tsfile_reader_develop in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 6b322007abdc33967e58547b62d6640609d57f88 Author: Tian Jiang <[email protected]> AuthorDate: Mon Jul 20 11:55:41 2026 +0800 Add buffered TsFile input --- .../tsfile/read/reader/BufferedTsFileInput.java | 203 +++++++++++++ .../tsfile/read/reader/LocalTsFileInput.java | 4 + .../read/TsFileSequenceReaderPerformanceTest.java | 331 +++++++++++++++++++++ .../read/reader/BufferedTsFileInputTest.java | 145 +++++++++ 4 files changed, 683 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/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); + } +}
