[core] Extract PageFileInput and PageFileOutput from Hash lookup

paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java

@@ -23,6 +23,8 @@
 import io.airlift.compress.zstd.ZstdCompressor;
 import io.airlift.compress.zstd.ZstdDecompressor;
 
+import javax.annotation.Nullable;
+
 /**
  * Each compression codec has an implementation of {@link BlockCompressionFactory} to create
  * compressors and decompressors.
@@ -34,7 +36,12 @@ public interface BlockCompressionFactory {
     BlockDecompressor getDecompressor();
 
     /** Creates {@link BlockCompressionFactory} according to the configuration. */
-    static BlockCompressionFactory create(String compression) {
+    @Nullable
+    static BlockCompressionFactory create(@Nullable String compression) {
+        if (compression == null) {
+            return null;
+        }
+
         switch (compression.toUpperCase()) {
             case "LZ4":
                 return new Lz4BlockCompressionFactory();

paimon-common/src/main/java/org/apache/paimon/io/CompressedPageFileInput.java

@@ -0,0 +1,115 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockDecompressor;
+import org.apache.paimon.utils.MathUtils;
+
+import java.io.IOException;
+import java.io.RandomAccessFile;
+
+/** A class to wrap compressed {@link RandomAccessFile}. */
+public class CompressedPageFileInput implements PageFileInput {
+
+    private final RandomAccessFile file;
+    private final int pageSize;
+    private final long uncompressBytes;
+    private final long[] pagePositions;
+
+    private final BlockDecompressor decompressor;
+    private final byte[] uncompressedBuffer;
+    private final byte[] compressedBuffer;
+
+    private final int pageSizeBits;
+    private final int pageSizeMask;
+
+    public CompressedPageFileInput(
+            RandomAccessFile file,
+            int pageSize,
+            BlockCompressionFactory compressionFactory,
+            long uncompressBytes,
+            long[] pagePositions) {
+        this.file = file;
+        this.pageSize = pageSize;
+        this.uncompressBytes = uncompressBytes;
+        this.pagePositions = pagePositions;
+
+        this.uncompressedBuffer = new byte[pageSize];
+        this.decompressor = compressionFactory.getDecompressor();
+        this.compressedBuffer =
+                new byte[compressionFactory.getCompressor().getMaxCompressedSize(pageSize)];
+
+        this.pageSizeBits = MathUtils.log2strict(pageSize);
+        this.pageSizeMask = pageSize - 1;
+    }
+
+    @Override
+    public RandomAccessFile file() {
+        return file;
+    }
+
+    @Override
+    public long uncompressBytes() {
+        return uncompressBytes;
+    }
+
+    @Override
+    public int pageSize() {
+        return pageSize;
+    }
+
+    @Override
+    public byte[] readPage(int pageIndex) throws IOException {
+        long position = pagePositions[pageIndex];
+        file.seek(position);
+        int compressLength = file.readInt();
+        file.readFully(compressedBuffer, 0, compressLength);
+
+        int uncompressedLength =
+                decompressor.decompress(compressedBuffer, 0, compressLength, uncompressedBuffer, 0);
+
+        byte[] result = new byte[uncompressedLength];
+        System.arraycopy(uncompressedBuffer, 0, result, 0, uncompressedLength);
+        return result;
+    }
+
+    @Override
+    public byte[] readPosition(long position, int length) throws IOException {
+        int offset = (int) (position & this.pageSizeMask);
+        int pageIndex = (int) (position >>> this.pageSizeBits);
+
+        byte[] result = new byte[length];
+        int n = 0;
+        do {
+            byte[] page = readPage(pageIndex);
+            int currentLength = Math.min(page.length - offset, length - n);
+            System.arraycopy(page, offset, result, n, currentLength);
+            offset = 0;
+            n += currentLength;
+            pageIndex++;
+        } while (n < length);
+        return result;
+    }
+
+    @Override
+    public void close() throws IOException {
+        this.file.close();
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/CompressedPageFileOutput.java

@@ -0,0 +1,109 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressor;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/** A class to output bytes with compression. */
+public class CompressedPageFileOutput implements PageFileOutput {
+
+    private final FileOutputStream out;
+    private final byte[] page;
+    private final BlockCompressor compressor;
+    private final byte[] compressedPage;
+    private final List<Long> pages;
+
+    private long uncompressBytes;
+    private long position;
+    private int count;
+
+    public CompressedPageFileOutput(
+            File file, int pageSize, BlockCompressionFactory compressionFactory)
+            throws FileNotFoundException {
+        this.out = new FileOutputStream(file);
+        this.page = new byte[pageSize];
+        this.compressor = compressionFactory.getCompressor();
+        this.compressedPage = new byte[compressor.getMaxCompressedSize(pageSize)];
+        this.pages = new ArrayList<>();
+
+        this.uncompressBytes = 0;
+        this.position = 0;
+        this.count = 0;
+    }
+
+    private void flushBuffer() throws IOException {
+        if (count > 0) {
+            pages.add(position);
+            int len = compressor.compress(page, 0, count, compressedPage, 0);
+            // write length
+            out.write((len >>> 24) & 0xFF);
+            out.write((len >>> 16) & 0xFF);
+            out.write((len >>> 8) & 0xFF);
+            out.write(len & 0xFF);
+            // write page
+            out.write(compressedPage, 0, len);
+            count = 0;
+            position += (len + 4);
+        }
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        uncompressBytes += len;
+
+        while (len > 0) {
+            if (count >= page.length) {
+                flushBuffer();
+            }
+            int toWrite = Math.min(len, page.length - count);
+            System.arraycopy(b, off, page, count, toWrite);
+            off += toWrite;
+            len -= toWrite;
+            count += toWrite;
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        try (OutputStream ignored = out) {
+            flushBuffer();
+        }
+    }
+
+    public long uncompressBytes() {
+        return uncompressBytes;
+    }
+
+    public long[] pages() {
+        long[] pages = new long[this.pages.size()];
+        for (int i = 0; i < pages.length; i++) {
+            pages[i] = this.pages.get(i);
+        }
+        return pages;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/PageFileInput.java

@@ -0,0 +1,43 @@
+/*
+ * 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.io;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+
+/** An interface to read pages from file. */
+public interface PageFileInput extends Closeable {
+
+    RandomAccessFile file();
+
+    long uncompressBytes();
+
+    int pageSize();
+
+    byte[] readPage(int pageIndex) throws IOException;
+
+    byte[] readPosition(long position, int length) throws IOException;
+
+    static PageFileInput create(File file, int pageSize) throws IOException {
+        RandomAccessFile accessFile = new RandomAccessFile(file, "r");
+        return new UncompressedPageFileInput(accessFile, pageSize);
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/PageFileOutput.java

@@ -0,0 +1,33 @@
+/*
+ * 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.io;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+
+/** An interface to write bytes with pages into file. */
+public interface PageFileOutput extends Closeable {
+
+    void write(byte[] bytes, int off, int len) throws IOException;
+
+    static PageFileOutput create(File file) throws IOException {
+        return new UncompressedPageFileOutput(file);
+    }
+}