The current value is just a magic number. In the long run, this value could become * configurable, but for now it is a conservative, relatively small value that should bring * safe improvements for small skips (e.g. in reading meta data), that would hurt the most * with frequent seeks. * *
The optimal value depends on the DFS implementation and configuration plus the * underlying filesystem. For now, this number is chosen "big enough" to provide * improvements for smaller seeks, and "small enough" to avoid disadvantages over real * seeks. While the minimum should be the page size, a true optimum per system would be the * amounts of bytes the can be consumed sequentially within the seektime. Unfortunately, * seektime is not constant and devices, OS, and DFS potentially also use read buffers and * read-ahead. */ private static final int MIN_SKIP_BYTES = 1024 * 1024; private final FSDataInputStream in; private HadoopSeekableInputStream(FSDataInputStream in) { this.in = in; } @Override public void seek(long seekPos) throws IOException { // We do some optimizations to avoid that some implementations of distributed FS perform // expensive seeks when they are actually not needed. long delta = seekPos - getPos(); if (delta > 0L && delta <= MIN_SKIP_BYTES) { // Instead of a small forward seek, we skip over the gap skipFully(delta); } else if (delta != 0L) { // For larger gaps and backward seeks, we do a real seek forceSeek(seekPos); } // Do nothing if delta is zero. } @Override public long getPos() throws IOException { return in.getPos(); } @Override public int read() throws IOException { return in.read(); } @Override public int read(byte[] b, int off, int len) throws IOException { return in.read(b, off, len); } @Override public void close() throws IOException { in.close(); } /** * Positions the stream to the given location. In contrast to {@link #seek(long)}, this * method will always issue a "seek" command to the dfs and may not replace it by {@link * #skip(long)} for small seeks. * *
Notice that the underlying DFS implementation can still decide to do skip instead of * seek. * * @param seekPos the position to seek to. */ public void forceSeek(long seekPos) throws IOException { in.seek(seekPos); } /** * Skips over a given amount of bytes in the stream. * * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { while (bytes > 0) { bytes -= in.skip(bytes); } } } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java [118:207]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - protected abstract FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOException; private static class HadoopSeekableInputStream extends SeekableInputStream { /** * Minimum amount of bytes to skip forward before we issue a seek instead of discarding * read. * *
The current value is just a magic number. In the long run, this value could become * configurable, but for now it is a conservative, relatively small value that should bring * safe improvements for small skips (e.g. in reading meta data), that would hurt the most * with frequent seeks. * *
The optimal value depends on the DFS implementation and configuration plus the * underlying filesystem. For now, this number is chosen "big enough" to provide * improvements for smaller seeks, and "small enough" to avoid disadvantages over real * seeks. While the minimum should be the page size, a true optimum per system would be the * amounts of bytes the can be consumed sequentially within the seektime. Unfortunately, * seektime is not constant and devices, OS, and DFS potentially also use read buffers and * read-ahead. */ private static final int MIN_SKIP_BYTES = 1024 * 1024; private final FSDataInputStream in; private HadoopSeekableInputStream(FSDataInputStream in) { this.in = in; } @Override public void seek(long seekPos) throws IOException { // We do some optimizations to avoid that some implementations of distributed FS perform // expensive seeks when they are actually not needed. long delta = seekPos - getPos(); if (delta > 0L && delta <= MIN_SKIP_BYTES) { // Instead of a small forward seek, we skip over the gap skipFully(delta); } else if (delta != 0L) { // For larger gaps and backward seeks, we do a real seek forceSeek(seekPos); } // Do nothing if delta is zero. } @Override public long getPos() throws IOException { return in.getPos(); } @Override public int read() throws IOException { return in.read(); } @Override public int read(byte[] b, int off, int len) throws IOException { return in.read(b, off, len); } @Override public void close() throws IOException { in.close(); } /** * Positions the stream to the given location. In contrast to {@link #seek(long)}, this * method will always issue a "seek" command to the dfs and may not replace it by {@link * #skip(long)} for small seeks. * *
Notice that the underlying DFS implementation can still decide to do skip instead of * seek. * * @param seekPos the position to seek to. */ public void forceSeek(long seekPos) throws IOException { in.seek(seekPos); } /** * Skips over a given amount of bytes in the stream. * * @param bytes the number of bytes to skip. */ public void skipFully(long bytes) throws IOException { while (bytes > 0) { bytes -= in.skip(bytes); } } } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -