diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index 2b0dcec3f760..8cbc47c24343 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java @@ -84,6 +84,15 @@ default void setRuntimeContext(Map options) {} */ SeekableInputStream newInputStream(Path path) throws IOException; + /** + * Opens a stream for a file whose length is already known, for example from a manifest. + * Implementations may avoid a separate file-status request and optimize positioned reads. The + * caller must supply the length of the file being opened. + */ + default SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + return newInputStream(path); + } + /** * Opens an PositionOutputStream at the indicated Path. * diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java index 587c1f2d4423..ac10e0c2a972 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java @@ -58,6 +58,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException { return wrap(() -> fileIO(path).newInputStream(path)); } + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + return wrap(() -> fileIO(path).newInputStream(path, fileSize)); + } + @Override public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException { return wrap(() -> fileIO(path).newOutputStream(path, overwrite)); diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java index 5568ba896cb3..5f38c8070705 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java @@ -74,6 +74,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException { return wrap(() -> fileIO(path).newInputStream(path)); } + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + return wrap(() -> fileIO(path).newInputStream(path, fileSize)); + } + @Override public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException { return wrap(() -> fileIO(path).newOutputStream(path, overwrite)); diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java b/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java index e0933d76e17d..fc71ea0bece8 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/VectoredReadUtils.java @@ -222,7 +222,8 @@ private static void copyToFileRanges( } } - private static byte[] getOrCreateBuffer(FileRange range) { + /** Return a range's caller-provided buffer, allocating it lazily when necessary. */ + public static byte[] getOrCreateBuffer(FileRange range) { if (range instanceof FileRange.FileRangeImpl) { return ((FileRange.FileRangeImpl) range).getOrCreateBuffer(); } @@ -258,7 +259,8 @@ private static void copyMultiBytesToBytes( } } - private static List validateAndSortRanges( + /** Validate ranges before dispatching them to multiple readers or caches. */ + public static List validateAndSortRanges( final List input) throws EOFException { requireNonNull(input, "Null input list"); checkArgument(!input.isEmpty(), "Empty input list"); diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java index 65eeaa3ebfd7..0ca0d250ccfd 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java @@ -150,6 +150,15 @@ public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws return delegate.newOutputStream(path, overwrite); } + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + if (!whitelist.contains(FileType.classify(path)) || FileType.isMutable(path)) { + return delegate.newInputStream(path, fileSize); + } + // Keep the existing cache namespace and mutable-file validation semantics. + return newInputStream(path); + } + @Override public FileStatus getFileStatus(Path path) throws IOException { return delegate.getFileStatus(path); diff --git a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java index a4312470007b..ca4206249af6 100644 --- a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java @@ -131,6 +131,11 @@ public SeekableInputStream newInputStream(Path path) throws IOException { return fileIO().newInputStream(path); } + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + return fileIO().newInputStream(path, fileSize); + } + @Override public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException { return fileIO().newOutputStream(path, overwrite); diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java index d39174378587..4d94ebdda636 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java @@ -125,7 +125,8 @@ private DataEvolutionGlobalIndexScanner( partitionFilter, coverageIndexFiles, table.coreOptions().scalarIndexSearchMode()); - GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath()); + GlobalIndexFileReader indexFileReader = + meta -> fileIO.newInputStream(meta.filePath(), meta.fileSize()); Map indexMetas = new HashMap<>(); Map> extraIndexMetas = new HashMap<>(); for (IndexFileMeta indexFile : indexFiles) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index 919100ec1f8f..add9e25867cc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -138,6 +138,7 @@ public DataEvolutionSplitRead( FileStorePathFactory pathFactory) { this.fileIO = fileIO; final Map cache = new HashMap<>(); + cache.put(schema.id(), schema); this.schemaFetcher = schemaId -> cache.computeIfAbsent(schemaId, key -> schemaManager.schema(schemaId)); this.schema = schema; diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java index cd60d981643a..357ef7f51963 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java @@ -32,9 +32,11 @@ import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.manifest.FileSource; import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.source.DataSplit; @@ -45,12 +47,15 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.paimon.data.BinaryRow.EMPTY_ROW; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -63,6 +68,91 @@ class DataEvolutionSplitReadTest { @TempDir java.nio.file.Path tempDir; + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testPredicateReusesCurrentSchemaAndLoadsOlderSchema(boolean evolved) throws Exception { + LocalFileIO fileIO = new LocalFileIO(); + Path tableRoot = new Path(tempDir.toUri().toString()); + CoreOptions options = new CoreOptions(new Options()); + FileStorePathFactory paths = + new FileStorePathFactory( + tableRoot, + RowType.of(), + options.partitionDefaultName(), + "parquet", + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null, + null, + CoreOptions.ExternalPathStrategy.NONE, + null, + false, + null); + SchemaManager manager = new FileSystemSchemaManager(fileIO, tableRoot); + TableSchema original = + manager.createTable( + Schema.newBuilder() + .column("f0", DataTypes.INT()) + .column("f1", DataTypes.STRING()) + .build()); + TableSchema current = + evolved + ? manager.commitChanges(SchemaChange.addColumn("extra", DataTypes.INT())) + : original; + Path bucket = paths.bucketPath(EMPTY_ROW, 0); + fileIO.mkdirs(bucket); + Path file = new Path(bucket, "data-0.parquet"); + writeFormatFile(fileIO, file, original.logicalRowType(), 100, "parquet"); + AtomicInteger fetched = new AtomicInteger(); + SchemaManager readManager = + new FileSystemSchemaManager(fileIO, tableRoot) { + @Override + public TableSchema schema(long id) { + fetched.incrementAndGet(); + return super.schema(id); + } + }; + DataSplit split = + DataSplit.builder() + .withPartition(EMPTY_ROW) + .withBucket(0) + .withBucketPath(bucket.toString()) + .withDataFiles( + Collections.singletonList( + createFile( + "data-0.parquet", + fileIO.getFileSize(file), + 10, + 100, + 1))) + .rawConvertible(false) + .build(); + DataEvolutionSplitRead read = + new DataEvolutionSplitRead( + fileIO, readManager, current, current.logicalRowType(), options, paths); + read.withFilter(new PredicateBuilder(current.logicalRowType()).greaterOrEqual(0, 1000)); + List actual = new ArrayList<>(); + try (RecordReader reader = + read.createReader( + new IndexedSplit( + split, + Arrays.asList(new Range(12, 12), new Range(42, 42)), + null))) { + reader.forEachRemaining( + row -> { + actual.add(row.getInt(0)); + if (evolved) { + assertTrue(row.isNullAt(2)); + } + }); + } + assertEquals(Arrays.asList(1002, 1032), actual); + assertEquals(evolved ? 1 : 0, fetched.get()); + } + @Test public void testDifferentRowIdRange() { DataFileMeta file1 = createFile("file1.parquet", 1L, 100, 10); diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java index 5a3b48a5c54b..b13e0e3e9048 100644 --- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java @@ -23,6 +23,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.HadoopOptionsProvider; import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.TwoPhaseOutputStream; import org.apache.paimon.options.Options; import org.apache.paimon.utils.IOUtils; @@ -236,6 +237,24 @@ protected AliyunOSSFileSystem createFileSystem(org.apache.hadoop.fs.Path path) { } } + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) throws IOException { + URI uri = path.toUri(); + if (fileSize < 0 || !"oss".equals(uri.getScheme()) || uri.getHost() == null) { + return super.newInputStream(path); + } + try { + return new OSSRangeInputStream( + ossClient(path), + uri.getHost(), + uri.getPath().substring(1), + fileSize, + FileSystem.getStatistics("oss", AliyunOSSFileSystem.class)); + } catch (Exception e) { + throw new IOException("Failed to open OSS file " + path, e); + } + } + @Override public boolean tryToWriteAtomic(Path path, String content) throws IOException { URI uri = path.toUri(); diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSRangeInputStream.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSRangeInputStream.java new file mode 100644 index 000000000000..b79d055ecb6d --- /dev/null +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSRangeInputStream.java @@ -0,0 +1,270 @@ +/* + * 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.paimon.oss; + +import org.apache.paimon.fs.FileRange; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.VectoredReadUtils; +import org.apache.paimon.fs.VectoredReadable; + +import com.aliyun.oss.OSSClient; +import com.aliyun.oss.OSSException; +import com.aliyun.oss.model.GetObjectRequest; +import org.apache.hadoop.fs.FileSystem; + +import javax.annotation.Nullable; + +import java.io.EOFException; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import static org.apache.paimon.utils.ExceptionUtils.firstOrSuppressed; +import static org.apache.paimon.utils.ExceptionUtils.rethrowIOException; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Lazy, bounded OSS range reads for files whose length is already recorded in metadata. */ +class OSSRangeInputStream extends SeekableInputStream implements VectoredReadable { + + private static final int BUFFER_SIZE = 64 * 1024; + + private final OSSClient client; + private final String bucket; + private final String key; + private final long fileSize; + @Nullable private final FileSystem.Statistics statistics; + private final Set activeRequests = new HashSet<>(); + + private volatile boolean closed; + private long position; + private byte[] buffer; + private long bufferStart; + private int bufferLength; + + OSSRangeInputStream( + OSSClient client, + String bucket, + String key, + long fileSize, + @Nullable FileSystem.Statistics statistics) { + checkArgument(fileSize >= 0, "File size must be non-negative."); + this.client = client; + this.bucket = bucket; + this.key = key; + this.fileSize = fileSize; + this.statistics = statistics; + } + + @Override + public synchronized void seek(long desired) throws IOException { + ensureOpen(); + if (desired < 0 || desired > fileSize) { + throw new EOFException("Seek outside file: " + desired); + } + position = desired; + } + + @Override + public synchronized long getPos() throws IOException { + ensureOpen(); + return position; + } + + @Override + public synchronized int read() throws IOException { + ensureOpen(); + if (position == fileSize) { + return -1; + } + if (position < bufferStart || position >= bufferStart + bufferLength) { + fillBuffer(); + } + int value = buffer[(int) (position++ - bufferStart)] & 0xff; + if (statistics != null) { + statistics.incrementBytesRead(1); + } + return value; + } + + @Override + public synchronized int read(byte[] bytes, int offset, int length) throws IOException { + checkBounds(bytes, offset, length); + ensureOpen(); + if (length == 0) { + return 0; + } + if (position == fileSize) { + return -1; + } + final int count; + if (position >= bufferStart && position < bufferStart + bufferLength) { + count = (int) Math.min(length, bufferStart + bufferLength - position); + System.arraycopy(buffer, (int) (position - bufferStart), bytes, offset, count); + } else if (length >= BUFFER_SIZE) { + count = readRange(position, bytes, offset, length); + } else { + fillBuffer(); + count = Math.min(length, bufferLength); + System.arraycopy(buffer, 0, bytes, offset, count); + } + position += count; + if (statistics != null) { + statistics.incrementBytesRead(count); + } + return count; + } + + private void fillBuffer() throws IOException { + if (buffer == null) { + buffer = new byte[BUFFER_SIZE]; + } + bufferStart = position; + bufferLength = 0; + bufferLength = + readRange(position, buffer, 0, (int) Math.min(BUFFER_SIZE, fileSize - position)); + } + + @Override + public int pread(long start, byte[] bytes, int offset, int length) throws IOException { + int count = readRange(start, bytes, offset, length); + if (count > 0 && statistics != null) { + statistics.incrementBytesRead(count); + } + return count; + } + + private int readRange(long start, byte[] bytes, int offset, int length) throws IOException { + checkBounds(bytes, offset, length); + ensureOpen(); + checkArgument(start >= 0, "Read position must be non-negative."); + if (length == 0) { + return 0; + } + if (start >= fileSize) { + return -1; + } + int count = (int) Math.min(length, fileSize - start); + GetObjectRequest request = new GetObjectRequest(bucket, key); + request.setRange(start, start + count - 1); + final InputStream input; + try { + input = client.getObject(request).getObjectContent(); + if (statistics != null) { + statistics.incrementReadOps(1); + } + } catch (OSSException e) { + if ("NoSuchKey".equals(e.getErrorCode())) { + FileNotFoundException missing = new FileNotFoundException(key); + missing.initCause(e); + throw missing; + } + throw new IOException("Failed to open OSS range for " + key, e); + } catch (RuntimeException e) { + throw new IOException("Failed to open OSS range for " + key, e); + } + synchronized (activeRequests) { + if (closed) { + input.close(); + throw new IOException("Stream is closed."); + } + activeRequests.add(input); + } + try (InputStream in = input) { + int read = 0; + while (read < count) { + int n = in.read(bytes, offset + read, count - read); + if (n < 0) { + throw new EOFException("OSS range ended before the recorded file length."); + } + if (n == 0) { + int value = in.read(); + if (value < 0) { + throw new EOFException("OSS range ended before the recorded file length."); + } + bytes[offset + read++] = (byte) value; + } else { + read += n; + } + } + return count; + } finally { + synchronized (activeRequests) { + activeRequests.remove(input); + } + } + } + + @Override + public void readVectored(List ranges) throws IOException { + ensureOpen(); + for (FileRange range : ranges) { + if (range.getLength() > 0 + && (range.getOffset() < 0 + || range.getOffset() >= fileSize + || range.getLength() > fileSize - range.getOffset())) { + throw new EOFException("Range exceeds the recorded file length."); + } + } + // Even a single combined range must use its exact bounds, not the sequential buffer. + VectoredReadUtils.readVectored( + this, + ranges, + VectoredReadUtils.ReadOptions.from(this).withSequentialReadFallback(false)); + } + + @Override + public void close() throws IOException { + final List requests; + synchronized (activeRequests) { + if (closed) { + return; + } + closed = true; + requests = new ArrayList<>(activeRequests); + activeRequests.clear(); + } + Throwable failure = null; + for (InputStream input : requests) { + try { + input.close(); + } catch (Throwable e) { + failure = firstOrSuppressed(e, failure); + } + } + if (failure != null) { + rethrowIOException(failure); + } + } + + private void ensureOpen() throws IOException { + if (closed) { + throw new IOException("Stream is closed."); + } + } + + private static void checkBounds(byte[] bytes, int offset, int length) { + if (offset < 0 || length < 0 || offset > bytes.length - length) { + throw new IndexOutOfBoundsException(); + } + } +} diff --git a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSRangeInputStreamTest.java b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSRangeInputStreamTest.java new file mode 100644 index 000000000000..791229be6c03 --- /dev/null +++ b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSRangeInputStreamTest.java @@ -0,0 +1,383 @@ +/* + * 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.paimon.oss; + +import org.apache.paimon.fs.FileRange; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; + +import com.aliyun.oss.OSSClient; +import com.aliyun.oss.OSSException; +import com.aliyun.oss.model.GetObjectRequest; +import com.aliyun.oss.model.OSSObject; +import org.apache.hadoop.fs.FileSystem; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; + +import java.io.ByteArrayInputStream; +import java.io.EOFException; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** Tests for exact OSS range reads, independent of an OSS service. */ +class OSSRangeInputStreamTest { + + @Test + void testKnownLengthOpenAndSeekIssueNoRequests() throws Exception { + OSSClient client = mock(OSSClient.class); + OSSFileIO fileIO = + new OSSFileIO() { + @Override + OSSClient ossClient(Path path) { + return client; + } + }; + try (SeekableInputStream in = + fileIO.newInputStream(new Path("oss://bucket/a.parquet"), 99)) { + assertThat(in).isInstanceOf(OSSRangeInputStream.class); + in.seek(98); + assertThat(in.getPos()).isEqualTo(98); + } + verifyNoInteractions(client); + } + + @Test + void testExactPositionedReadAndEndOfFile() throws Exception { + byte[] data = data(100_000); + AtomicInteger closed = new AtomicInteger(); + OSSClient client = client(data, closed); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", data.length, null)) { + in.seek(123); + byte[] result = new byte[8]; + assertThat(in.pread(data.length - 3, result, 2, 6)).isEqualTo(3); + assertThat(Arrays.copyOfRange(result, 2, 5)) + .containsExactly(Arrays.copyOfRange(data, data.length - 3, data.length)); + assertThat(in.getPos()).isEqualTo(123); + assertThat(in.pread(data.length, result, 0, result.length)).isEqualTo(-1); + assertThat(in.pread(data.length, result, 0, 0)).isZero(); + in.seek(data.length); + assertThat(in.read()).isEqualTo(-1); + assertThat(in.read(result, 0, 0)).isZero(); + assertThatThrownBy(() -> in.seek(-1)).isInstanceOf(EOFException.class); + assertThatThrownBy(() -> in.seek(data.length + 1L)).isInstanceOf(EOFException.class); + } + ArgumentCaptor request = ArgumentCaptor.forClass(GetObjectRequest.class); + verify(client).getObject(request.capture()); + assertThat(request.getValue().getRange()).containsExactly(data.length - 3, data.length - 1); + assertThat(closed.get()).isEqualTo(1); + } + + @Test + void testBufferAndRetryAfterTruncatedRead() throws Exception { + byte[] data = data(100_000); + OSSClient client = client(data, new AtomicInteger()); + OSSObject shortObject = new OSSObject(); + shortObject.setObjectContent(new ByteArrayInputStream(new byte[3])); + OSSObject fullObject = new OSSObject(); + fullObject.setObjectContent(new ByteArrayInputStream(Arrays.copyOf(data, 64 * 1024))); + when(client.getObject(any(GetObjectRequest.class))).thenReturn(shortObject, fullObject); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", data.length, null)) { + assertThatThrownBy(in::read).isInstanceOf(EOFException.class); + assertThat(in.getPos()).isZero(); + assertThat(in.read()).isEqualTo(data[0] & 0xff); + in.seek(9); + assertThat(in.read()).isEqualTo(data[9] & 0xff); + } + verify(client, times(2)).getObject(any(GetObjectRequest.class)); + } + + @Test + void testVectoredReadsAreBoundedAndDoNotMovePosition() throws Exception { + byte[] data = data(2_000_000); + AtomicInteger closed = new AtomicInteger(); + OSSClient client = client(data, closed); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", data.length, null)) { + in.seek(17); + List ranges = + Arrays.asList( + FileRange.createFileRange(1_200_000, 128), + FileRange.createFileRange(100, 256), + FileRange.createFileRange(600_000, 192)); + in.readVectored(ranges); + for (FileRange range : ranges) { + assertThat(range.getData().get(10, TimeUnit.SECONDS)) + .containsExactly( + Arrays.copyOfRange( + data, + (int) range.getOffset(), + (int) range.getOffset() + range.getLength())); + } + assertThat(in.getPos()).isEqualTo(17); + in.readVectored(Collections.emptyList()); + } + assertThat(closed.get()).isEqualTo(3); + verify(client, times(3)).getObject(any(GetObjectRequest.class)); + } + + @Test + void testFailureCompletesRangeExceptionally() throws Exception { + OSSClient client = mock(OSSClient.class); + when(client.getObject(any(GetObjectRequest.class))) + .thenThrow(new IllegalStateException("failure")); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", 1024, null)) { + FileRange range = FileRange.createFileRange(0, 10); + in.readVectored(Collections.singletonList(range)); + assertThatThrownBy(() -> range.getData().get(10, TimeUnit.SECONDS)) + .hasCauseInstanceOf(IOException.class); + } + } + + @Test + void testMissingObjectKeepsFileNotFoundSemantics() throws Exception { + OSSClient client = mock(OSSClient.class); + when(client.getObject(any(GetObjectRequest.class))) + .thenThrow( + new OSSException( + "missing", "NoSuchKey", "request", "host", null, null, "GET")); + try (OSSRangeInputStream in = new OSSRangeInputStream(client, "bucket", "key", 10, null)) { + assertThatThrownBy(in::read).isInstanceOf(FileNotFoundException.class); + } + } + + @Test + void testCloseReleasesAnActiveRequest() throws Exception { + CountDownLatch reading = new CountDownLatch(1); + CountDownLatch closed = new CountDownLatch(1); + OSSObject object = new OSSObject(); + object.setObjectContent( + new InputStream() { + @Override + public int read() throws IOException { + reading.countDown(); + try { + if (!closed.await(10, TimeUnit.SECONDS)) { + throw new IOException("Close timed out"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Closed"); + } + + @Override + public void close() { + closed.countDown(); + } + }); + OSSClient client = mock(OSSClient.class); + when(client.getObject(any(GetObjectRequest.class))).thenReturn(object); + OSSRangeInputStream in = new OSSRangeInputStream(client, "bucket", "key", 10, null); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future read = executor.submit(() -> in.pread(0, new byte[1], 0, 1)); + assertThat(reading.await(10, TimeUnit.SECONDS)).isTrue(); + in.close(); + assertThatThrownBy(() -> read.get(10, TimeUnit.SECONDS)) + .hasCauseInstanceOf(IOException.class); + assertThatThrownBy(in::read).isInstanceOf(IOException.class); + in.close(); + } finally { + in.close(); + executor.shutdownNow(); + } + } + + private static byte[] data(int length) { + byte[] data = new byte[length]; + for (int i = 0; i < length; i++) { + data[i] = (byte) (i * 31); + } + return data; + } + + @ParameterizedTest + @ValueSource(strings = {"io", "runtime", "error"}) + void testCloseAttemptsAllRequestsWhenClosingThrows(String kind) throws Exception { + Throwable failure = + "io".equals(kind) + ? new IOException("close failure") + : "runtime".equals(kind) + ? new IllegalStateException("close failure") + : new AssertionError("close failure"); + CountDownLatch reading = new CountDownLatch(3); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger closed = new AtomicInteger(); + OSSClient client = mock(OSSClient.class); + when(client.getObject(any(GetObjectRequest.class))) + .thenAnswer( + invocation -> { + OSSObject object = new OSSObject(); + object.setObjectContent( + new InputStream() { + private final AtomicBoolean attempted = new AtomicBoolean(); + + @Override + public int read() throws IOException { + reading.countDown(); + try { + if (!release.await(10, TimeUnit.SECONDS)) { + throw new IOException("Read timed out"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Read released"); + } + + @Override + public void close() throws IOException { + if (attempted.compareAndSet(false, true)) { + closed.incrementAndGet(); + org.apache.paimon.utils.ExceptionUtils + .rethrowIOException(failure); + } + } + }); + return object; + }); + OSSRangeInputStream in = new OSSRangeInputStream(client, "bucket", "key", 10, null); + ExecutorService executor = Executors.newFixedThreadPool(3); + List> reads = new ArrayList<>(); + try { + for (int i = 0; i < 3; i++) { + reads.add(executor.submit(() -> in.pread(0, new byte[1], 0, 1))); + } + assertThat(reading.await(10, TimeUnit.SECONDS)).isTrue(); + assertThatThrownBy(in::close).isSameAs(failure); + assertThat(closed.get()).isEqualTo(3); + in.close(); + } finally { + release.countDown(); + executor.shutdown(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + in.close(); + } + for (Future read : reads) { + assertThatThrownBy(() -> read.get(10, TimeUnit.SECONDS)) + .hasCauseInstanceOf(IOException.class); + } + } + + @Test + void testRandomReadsAcrossBufferBoundaries() throws Exception { + byte[] data = data(180_000); + OSSClient client = client(data, new AtomicInteger()); + Random random = new Random(20260908); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", data.length, null)) { + for (int i = 0; i < 500; i++) { + int position = random.nextInt(data.length + 1); + in.seek(position); + byte[] bytes = new byte[random.nextInt(90_000)]; + int count = in.read(bytes, 0, bytes.length); + if (count > 0) { + assertThat(Arrays.copyOf(bytes, count)) + .containsExactly(Arrays.copyOfRange(data, position, position + count)); + assertThat(in.getPos()).isEqualTo(position + count); + } else { + assertThat(count).isEqualTo(bytes.length == 0 ? 0 : -1); + } + long saved = in.getPos(); + int offset = random.nextInt(data.length); + byte[] other = new byte[31]; + int n = in.pread(offset, other, 0, other.length); + assertThat(Arrays.copyOf(other, n)) + .containsExactly(Arrays.copyOfRange(data, offset, offset + n)); + assertThat(in.getPos()).isEqualTo(saved); + } + assertThatThrownBy( + () -> + in.readVectored( + Collections.singletonList( + FileRange.createFileRange(data.length - 1, 2)))) + .isInstanceOf(EOFException.class); + } + } + + private static OSSClient client(byte[] data, AtomicInteger closed) { + OSSClient client = mock(OSSClient.class); + when(client.getObject(any(GetObjectRequest.class))) + .thenAnswer( + invocation -> { + long[] range = + ((GetObjectRequest) invocation.getArgument(0)).getRange(); + byte[] bytes = + Arrays.copyOfRange(data, (int) range[0], (int) range[1] + 1); + OSSObject object = new OSSObject(); + object.setObjectContent( + new ByteArrayInputStream(bytes) { + @Override + public void close() { + closed.incrementAndGet(); + } + }); + return object; + }); + return client; + } + + @Test + void testHadoopStatisticsCountLogicalReadsWithoutDoubleCountingBufferFills() throws Exception { + byte[] data = data(100_000); + OSSClient client = client(data, new AtomicInteger()); + FileSystem.Statistics statistics = new FileSystem.Statistics("oss"); + try (OSSRangeInputStream in = + new OSSRangeInputStream(client, "bucket", "key", data.length, statistics)) { + in.read(); + in.read(new byte[10]); + assertThat(statistics.getReadOps()).isEqualTo(1); + assertThat(statistics.getBytesRead()).isEqualTo(11); + in.pread(5000, new byte[10], 0, 10); + in.seek(90_000); + in.read(new byte[10_000]); + assertThat(statistics.getReadOps()).isEqualTo(3); + assertThat(statistics.getBytesRead()).isEqualTo(10_021); + } + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetInputFile.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetInputFile.java index 0e68416ba20a..778a5765e08e 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetInputFile.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetInputFile.java @@ -20,6 +20,8 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.VectoredReadable; import org.apache.parquet.io.InputFile; @@ -54,7 +56,12 @@ public long getLength() { @Override public ParquetInputStream newStream() throws IOException { - return new ParquetInputStream(fileIO.newInputStream(path)); + boolean cacheTail = length >= 0 && fileIO.isObjectStore(); + SeekableInputStream stream = fileIO.newInputStream(path, length); + if (cacheTail && stream instanceof VectoredReadable) { + stream = new ParquetTailInputStream(stream, length); + } + return new ParquetInputStream(stream); } @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetTailInputStream.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetTailInputStream.java new file mode 100644 index 000000000000..27bc13a7d536 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetTailInputStream.java @@ -0,0 +1,188 @@ +/* + * 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.paimon.format.parquet; + +import org.apache.paimon.fs.FileRange; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.VectoredReadUtils; +import org.apache.paimon.fs.VectoredReadable; + +import java.io.EOFException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * A bounded, per-open-file tail buffer. Usually the footer and page indexes share one range GET; + * larger metadata falls through to the original reader without changing the Parquet format. + */ +final class ParquetTailInputStream extends SeekableInputStream implements VectoredReadable { + + private static final int TAIL_SIZE = 128 * 1024; + + private final SeekableInputStream delegate; + private final VectoredReadable positioned; + private final long fileSize; + private final long tailStart; + private volatile byte[] tail; + private volatile boolean closed; + private long position; + + ParquetTailInputStream(SeekableInputStream delegate, long fileSize) { + this.delegate = delegate; + this.positioned = (VectoredReadable) delegate; + this.fileSize = fileSize; + this.tailStart = Math.max(0, fileSize - TAIL_SIZE); + } + + @Override + public void seek(long desired) throws IOException { + ensureOpen(); + if (desired < 0 || desired > fileSize) { + throw new EOFException("Seek outside Parquet file: " + desired); + } + position = desired; + } + + @Override + public long getPos() throws IOException { + ensureOpen(); + return position; + } + + @Override + public int read() throws IOException { + ensureOpen(); + if (position == fileSize) { + return -1; + } + if (position >= tailStart) { + return tail()[(int) (position++ - tailStart)] & 0xff; + } + delegate.seek(position); + int value = delegate.read(); + if (value >= 0) { + position++; + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + if (offset < 0 || length < 0 || offset > bytes.length - length) { + throw new IndexOutOfBoundsException(); + } + ensureOpen(); + if (length == 0) { + return 0; + } + if (position == fileSize) { + return -1; + } + final int count; + if (position >= tailStart) { + count = (int) Math.min(length, fileSize - position); + System.arraycopy(tail(), (int) (position - tailStart), bytes, offset, count); + } else { + delegate.seek(position); + count = delegate.read(bytes, offset, (int) Math.min(length, tailStart - position)); + } + if (count > 0) { + position += count; + } + return count; + } + + private synchronized byte[] tail() throws IOException { + ensureOpen(); + if (tail == null) { + byte[] bytes = new byte[(int) (fileSize - tailStart)]; + positioned.preadFully(tailStart, bytes, 0, bytes.length); + ensureOpen(); + tail = bytes; + } + return tail; + } + + @Override + public int pread(long start, byte[] bytes, int offset, int length) throws IOException { + if (start < 0 || offset < 0 || length < 0 || offset > bytes.length - length) { + throw new IndexOutOfBoundsException(); + } + ensureOpen(); + if (length == 0) { + return 0; + } + if (start >= fileSize) { + return -1; + } + if (start >= tailStart) { + int count = (int) Math.min(length, fileSize - start); + System.arraycopy(tail(), (int) (start - tailStart), bytes, offset, count); + return count; + } + return positioned.pread(start, bytes, offset, length); + } + + @Override + public void readVectored(List ranges) throws IOException { + ensureOpen(); + if (ranges.isEmpty()) { + return; + } + List uncached = new ArrayList<>(); + for (FileRange range : VectoredReadUtils.validateAndSortRanges(ranges)) { + if (range.getLength() == 0) { + range.getData().complete(VectoredReadUtils.getOrCreateBuffer(range)); + continue; + } + if (range.getOffset() >= fileSize || range.getLength() > fileSize - range.getOffset()) { + throw new EOFException("Range exceeds the Parquet file length."); + } + if (range.getOffset() >= tailStart + && range.getOffset() <= fileSize + && range.getLength() <= fileSize - range.getOffset()) { + int start = (int) (range.getOffset() - tailStart); + byte[] buffer = VectoredReadUtils.getOrCreateBuffer(range); + System.arraycopy(tail(), start, buffer, 0, range.getLength()); + range.getData().complete(buffer); + } else { + uncached.add(range); + } + } + // Preserve native vectored implementations for all uncached data ranges. + if (!uncached.isEmpty()) { + positioned.readVectored(uncached); + } + } + + @Override + public void close() throws IOException { + if (!closed) { + closed = true; + delegate.close(); + } + } + + private void ensureOpen() throws IOException { + if (closed) { + throw new IOException("Stream is closed."); + } + } +} diff --git a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 9f56c2176958..77f7f3df30a9 100644 --- a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -403,6 +403,19 @@ private List filterRowGroups(List blocks) throws I } } + // Row IDs rule out groups without reading their dictionaries or bloom filters. Keep the + // missing-row-offset check above: selection is unsafe without those original offsets. + if (selection != null) { + blocks = + blocks.stream() + .filter( + it -> + selection.intersects( + it.getRowIndexOffset(), + it.getRowIndexOffset() + it.getRowCount())) + .collect(Collectors.toList()); + } + if (FilterCompat.isFilteringRequired(recordFilter)) { // set up data filters based on configured levels List levels = new ArrayList<>(); @@ -421,17 +434,6 @@ private List filterRowGroups(List blocks) throws I blocks = RowGroupFilter.filterRowGroups(levels, recordFilter, blocks, this); } - if (selection != null) { - blocks = - blocks.stream() - .filter( - it -> - selection.intersects( - it.getRowIndexOffset(), - it.getRowIndexOffset() + it.getRowCount())) - .collect(Collectors.toList()); - } - return blocks; } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTailInputStreamTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTailInputStreamTest.java new file mode 100644 index 000000000000..f738ddc5b81a --- /dev/null +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTailInputStreamTest.java @@ -0,0 +1,253 @@ +/* + * 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.paimon.format.parquet; + +import org.apache.paimon.fs.FileRange; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.VectoredReadable; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.utils.IOUtils; + +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.junit.jupiter.api.Test; + +import java.io.EOFException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for bounded per-reader tail caching and positioned I/O delegation. */ +class ParquetTailInputStreamTest { + + @Test + void testTailAndMetadataShareOneRead() throws Exception { + CountingInputStream delegate = new CountingInputStream(400_000); + try (ParquetTailInputStream in = + new ParquetTailInputStream(delegate, delegate.data.length)) { + assertThat(delegate.requests).isEmpty(); + in.seek(delegate.data.length - 8); + byte[] trailer = new byte[8]; + IOUtils.readFully(in, trailer); + assertThat(trailer) + .containsExactly( + Arrays.copyOfRange( + delegate.data, delegate.data.length - 8, delegate.data.length)); + in.seek(delegate.data.length - 100_000); + byte[] metadata = new byte[4096]; + IOUtils.readFully(in, metadata); + assertThat(metadata) + .containsExactly( + Arrays.copyOfRange( + delegate.data, + delegate.data.length - 100_000, + delegate.data.length - 100_000 + 4096)); + assertThat(delegate.requests).hasSize(1); + assertThat(delegate.requests.get(0)) + .containsExactly(delegate.data.length - 128 * 1024, 128 * 1024); + } + assertThat(delegate.closeCount).isEqualTo(1); + } + + @Test + void testLargeMetadataFallsThroughAndCrossesTailBoundary() throws Exception { + CountingInputStream delegate = new CountingInputStream(400_000); + try (ParquetTailInputStream in = + new ParquetTailInputStream(delegate, delegate.data.length)) { + in.seek(delegate.data.length - 8); + in.read(); + in.seek(0); + byte[] all = new byte[delegate.data.length]; + IOUtils.readFully(in, all); + assertThat(all).containsExactly(delegate.data); + assertThat(in.getPos()).isEqualTo(delegate.data.length); + assertThat(in.read()).isEqualTo(-1); + assertThat(in.read(all, 0, 0)).isZero(); + } + } + + @Test + void testVectoredReadsKeepNativeDelegateAndUseCachedTail() throws Exception { + CountingInputStream delegate = new CountingInputStream(400_000); + try (ParquetTailInputStream in = + new ParquetTailInputStream(delegate, delegate.data.length)) { + in.seek(399_999); + in.read(); + FileRange uncached = FileRange.createFileRange(100, 32); + FileRange cached = FileRange.createFileRange(399_000, 64); + in.readVectored(Arrays.asList(cached, uncached)); + assertThat(uncached.getData().get(5, TimeUnit.SECONDS)) + .containsExactly(Arrays.copyOfRange(delegate.data, 100, 132)); + assertThat(cached.getData().get(5, TimeUnit.SECONDS)) + .containsExactly(Arrays.copyOfRange(delegate.data, 399_000, 399_064)); + assertThat(delegate.vectoredCalls).isEqualTo(1); + assertThat(delegate.requests).hasSize(2); + assertThat(in.getPos()).isEqualTo(400_000); + in.readVectored(Collections.emptyList()); + FileRange empty = FileRange.createFileRange(400_000, 0); + in.readVectored(Collections.singletonList(empty)); + assertThat(empty.getData().get()).isEmpty(); + assertThatThrownBy( + () -> + in.readVectored( + Arrays.asList( + FileRange.createFileRange(399_000, 20), + FileRange.createFileRange(399_010, 20)))) + .isInstanceOf(IllegalArgumentException.class); + } + } + + @Test + void testPerStreamCacheAndClosedState() throws Exception { + CountingInputStream first = new CountingInputStream(10); + CountingInputStream second = new CountingInputStream(10); + second.data[9] = 99; + ParquetTailInputStream a = new ParquetTailInputStream(first, 10); + ParquetTailInputStream b = new ParquetTailInputStream(second, 10); + a.seek(9); + b.seek(9); + assertThat(a.read()).isNotEqualTo(b.read()); + a.close(); + a.close(); + assertThat(first.closeCount).isEqualTo(1); + assertThatThrownBy(a::read).isInstanceOf(IOException.class); + assertThatThrownBy(() -> a.pread(9, new byte[1], 0, 1)).isInstanceOf(IOException.class); + assertThatThrownBy(() -> b.seek(11)).isInstanceOf(EOFException.class); + b.close(); + } + + @Test + void testCachedRangesFillCallerProvidedBuffers() throws Exception { + CountingInputStream delegate = new CountingInputStream(400_000); + try (ParquetTailInputStream in = + new ParquetTailInputStream(delegate, delegate.data.length)) { + byte[] supplied = new byte[64]; + byte[] empty = new byte[0]; + FileRange cached = FileRange.createFileRange(399_000, supplied); + FileRange zeroLength = FileRange.createFileRange(400_000, empty); + in.readVectored(Arrays.asList(cached, zeroLength)); + assertThat(cached.getData().get(5, TimeUnit.SECONDS)).isSameAs(supplied); + assertThat(supplied) + .containsExactly(Arrays.copyOfRange(delegate.data, 399_000, 399_064)); + assertThat(zeroLength.getData().get(5, TimeUnit.SECONDS)).isSameAs(empty); + assertThat(delegate.requests).hasSize(1); + assertThat(delegate.vectoredCalls).isZero(); + } + } + + @Test + void testCorruptFooterStillClosesInput() throws Exception { + CountingInputStream delegate = new CountingInputStream(400_000); + LocalFileIO fileIO = + new LocalFileIO() { + @Override + public boolean isObjectStore() { + return true; + } + + @Override + public SeekableInputStream newInputStream(Path path, long fileSize) { + assertThat(fileSize).isEqualTo(delegate.data.length); + return delegate; + } + }; + ParquetInputFile file = + ParquetInputFile.fromPath(fileIO, new Path("/file.parquet"), delegate.data.length); + ParquetInputStream stream = file.newStream(); + assertThatThrownBy( + () -> + ParquetFileReader.readFooter( + file, ParquetReadOptions.builder().build(), stream, true)) + .isInstanceOf(RuntimeException.class); + assertThat(delegate.closeCount).isEqualTo(1); + } + + private static class CountingInputStream extends SeekableInputStream + implements VectoredReadable { + final byte[] data; + final List requests = new ArrayList<>(); + long position; + int vectoredCalls; + int closeCount; + + CountingInputStream(int length) { + data = new byte[length]; + for (int i = 0; i < length; i++) { + data[i] = (byte) (i * 31); + } + } + + @Override + public void seek(long desired) { + position = desired; + } + + @Override + public long getPos() { + return position; + } + + @Override + public int read() { + return position == data.length ? -1 : data[(int) position++] & 0xff; + } + + @Override + public int read(byte[] bytes, int offset, int length) { + int count = pread(position, bytes, offset, length); + if (count > 0) { + position += count; + } + return count; + } + + @Override + public int pread(long start, byte[] bytes, int offset, int length) { + requests.add(new long[] {start, length}); + int count = (int) Math.min(length, data.length - start); + if (count <= 0) { + return length == 0 ? 0 : -1; + } + System.arraycopy(data, (int) start, bytes, offset, count); + return count; + } + + @Override + public void readVectored(List ranges) { + vectoredCalls++; + for (FileRange range : ranges) { + byte[] bytes = new byte[range.getLength()]; + pread(range.getOffset(), bytes, 0, bytes.length); + range.getData().complete(bytes); + } + } + + @Override + public void close() { + closeCount++; + } + } +} diff --git a/paimon-format/src/test/java/org/apache/parquet/hadoop/SelectedRowGroupTest.java b/paimon-format/src/test/java/org/apache/parquet/hadoop/SelectedRowGroupTest.java new file mode 100644 index 000000000000..695d2508af7e --- /dev/null +++ b/paimon-format/src/test/java/org/apache/parquet/hadoop/SelectedRowGroupTest.java @@ -0,0 +1,143 @@ +/* + * 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.parquet.hadoop; + +import org.apache.paimon.format.parquet.ParquetInputFile; +import org.apache.paimon.format.parquet.ParquetInputStream; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.utils.RoaringBitmap32; + +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.EncodingStats; +import org.apache.parquet.column.page.DictionaryPage; +import org.apache.parquet.column.statistics.BinaryStatistics; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.filter2.predicate.FilterApi; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ColumnPath; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.metadata.FileMetaData; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Row IDs must reject unrelated groups before dictionary I/O, without weakening predicates. */ +class SelectedRowGroupTest { + + @Test + void testOnlySelectedDictionaryIsReadAndPredicateStillFilters() throws Exception { + assertSelection(5, false, "x", 1, Collections.singletonList(0)); + assertSelection(15, false, "x", 1, Collections.singletonList(1)); + assertSelection(5, false, "y", 0, Collections.singletonList(0)); + assertSelection(-1, false, "x", 0, Collections.emptyList()); + } + + @Test + void testMissingOffsetsKeepExistingFallback() throws Exception { + assertSelection(5, true, "x", 2, Collections.emptyList()); + } + + private void assertSelection( + int selected, + boolean missingOffset, + String value, + int groupCount, + List expectedDictionaries) + throws Exception { + MessageType schema = + MessageTypeParser.parseMessageType("message m { required binary tag (UTF8); }"); + List blocks = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + BlockMetaData block = new BlockMetaData(); + block.setRowCount(10); + block.setRowIndexOffset(missingOffset && i == 1 ? -1 : i * 10); + ColumnChunkMetaData column = + ColumnChunkMetaData.get( + ColumnPath.get("tag"), + schema.getType("tag").asPrimitiveType(), + CompressionCodecName.UNCOMPRESSED, + new EncodingStats.Builder() + .addDictEncoding(Encoding.PLAIN) + .addDataEncoding(Encoding.RLE_DICTIONARY) + .build(), + new HashSet<>(Arrays.asList(Encoding.PLAIN, Encoding.RLE_DICTIONARY)), + new BinaryStatistics(), + 20 + i * 100, + 4 + i * 100, + 10, + 50, + 50); + column.setRowGroupOrdinal(i); + block.addColumn(column); + blocks.add(block); + } + ParquetMetadata footer = + new ParquetMetadata( + new FileMetaData(schema, Collections.emptyMap(), "test"), blocks); + RoaringBitmap32 selection = new RoaringBitmap32(); + if (selected >= 0) { + selection.add(selected); + } + ParquetReadOptions options = + ParquetReadOptions.builder() + .withRecordFilter( + FilterCompat.get( + FilterApi.eq( + FilterApi.binaryColumn("tag"), + Binary.fromString(value)))) + .useStatsFilter(false) + .useDictionaryFilter(true) + .useBloomFilter(false) + .build(); + List dictionaries = new ArrayList<>(); + ParquetInputFile file = + ParquetInputFile.fromPath(LocalFileIO.create(), new Path("/unused.parquet"), 200); + ParquetInputStream input = + new ParquetInputStream( + SeekableInputStream.wrap(new ByteArrayInputStream(new byte[0]))); + try (ParquetFileReader reader = + new ParquetFileReader(file, footer, options, input, selection) { + @Override + DictionaryPage readDictionary(ColumnChunkMetaData column) { + dictionaries.add(column.getRowGroupOrdinal()); + return new DictionaryPage( + BytesInput.from(new byte[] {1, 0, 0, 0, 'x'}), 1, Encoding.PLAIN); + } + }) { + assertThat(reader.getRowGroups()).hasSize(groupCount); + assertThat(dictionaries).isEqualTo(expectedDictionaries); + } + } +}