fixup! Add Indexes and Stats format reader and writer

ByteBuffer for writes

Author: findepi

core/src/main/java/org/apache/iceberg/stats/StatsFormat.java

@@ -25,7 +25,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.math.BigInteger;
-import java.util.Arrays;
+import java.nio.ByteBuffer;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 final class StatsFormat {
@@ -57,27 +57,28 @@ static int readIntegerLittleEndian(byte[] data, int offset) {
         (Byte.toUnsignedInt(data[offset + 3]) << 24);
   }
 
-  static byte[] compressFooterPayload(byte[] payload) {
-    return compress(StatsCompressionCodec.LZ4, payload, 0, payload.length);
+  static ByteBuffer compressFooterPayload(ByteBuffer payload) {
+    return compress(StatsCompressionCodec.LZ4, payload);
   }
 
   static byte[] decompressFooterPayload(byte[] footer, int offset, int length) {
     return decompress(StatsCompressionCodec.LZ4, footer, offset, length);
   }
 
-  static byte[] compressBlob(StatsCompressionCodec codec, byte[] data, int dataOffset, int dataLength) {
-    return compress(codec, data, dataOffset, dataLength);
+  static ByteBuffer compressBlob(StatsCompressionCodec codec, ByteBuffer data) {
+    return compress(codec, data);
   }
 
   static byte[] decompressBlob(StatsCompressionCodec codec, byte[] data, int dataOffset, int dataLength) {
     return decompress(codec, data, dataOffset, dataLength);
   }
 
-  private static byte[] compress(StatsCompressionCodec codec, byte[] input, int inputOffset, int inputLength) {
+  private static ByteBuffer compress(StatsCompressionCodec codec, ByteBuffer input) {
     Compressor compressor = getCompressor(codec);
-    byte[] output = new byte[compressor.maxCompressedLength(inputLength)];
-    int length = compressor.compress(input, inputOffset, inputLength, output, 0, output.length);
-    return length == output.length ? output : Arrays.copyOf(output, length);
+    ByteBuffer output = ByteBuffer.allocate(compressor.maxCompressedLength(input.remaining()));
+    compressor.compress(input,  output);
+    output.flip();
+    return output;
   }
 
   private static Compressor getCompressor(StatsCompressionCodec codec) {

core/src/main/java/org/apache/iceberg/stats/StatsWriter.java

@@ -21,6 +21,8 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
@@ -58,32 +60,21 @@ public void addFileProperty(String name, String value) {
       throw new IllegalStateException(String.format("Property '%s' already set", name));
     }
   }
-
   public void append(
       String type,
       Set<Integer> columnsCovered,
-      byte[] blobData,
-      int dataOffset,
-      int dataLength,
+      ByteBuffer blobData,
       Optional<StatsCompressionCodec> compression) throws IOException {
     checkNotFinished();
     writeHeaderIfNeeded();
 
     Objects.requireNonNull(type, "type is null");
     long fileOffset = outputStream.getPos();
-    byte[] data;
-    int offset;
-    int length;
-    if (compression.isPresent()) {
-      data = StatsFormat.compressBlob(compression.get(), blobData, dataOffset, dataLength);
-      offset = 0;
-      length = data.length;
-    } else {
-      data = blobData;
-      offset = dataOffset;
-      length = dataLength;
-    }
-    outputStream.write(data, offset, length);
+    ByteBuffer data;
+    data = compression.map(codec -> StatsFormat.compressBlob(codec, blobData))
+        .orElse(blobData);
+    int length = data.remaining();
+    Channels.newChannel(outputStream).write(data);
     @Nullable String codecName = compression.map(StatsCompressionCodec::getCodecName).orElse(null);
     blobs.add(new BlobMetadata(type, columnsCovered, fileOffset, length, codecName));
   }
@@ -120,13 +111,13 @@ public void finish() throws IOException {
     }
 
     FileMetadata fileMetadata = new FileMetadata(blobs, properties);
-    byte[] footerJson = FileMetadataParser.toJson(fileMetadata).getBytes(StandardCharsets.UTF_8);
-    byte[] footerPayload = compressFooter ? StatsFormat.compressFooterPayload(footerJson) : footerJson;
+    ByteBuffer footerJson = ByteBuffer.wrap(FileMetadataParser.toJson(fileMetadata).getBytes(StandardCharsets.UTF_8));
+    ByteBuffer footerPayload = compressFooter ? StatsFormat.compressFooterPayload(footerJson) : footerJson;
     long footerOffset = outputStream.getPos();
     byte[] magic = StatsFormat.getMagic();
     outputStream.write(magic);
-    outputStream.write(footerPayload);
-    StatsFormat.writeIntegerLittleEndian(outputStream, footerPayload.length);
+    int footerPayloadLength = Channels.newChannel(outputStream).write(footerPayload);
+    StatsFormat.writeIntegerLittleEndian(outputStream, footerPayloadLength);
     StatsFormat.writeIntegerLittleEndian(outputStream, 0); // Reserved
     StatsFormat.writeIntegerLittleEndian(outputStream, getFileFlags());
     StatsFormat.writeIntegerLittleEndian(outputStream, StatsFormat.CURRENT_FORMAT_VERSION);

core/src/test/java/org/apache/iceberg/stats/TestStatsWriter.java

@@ -19,6 +19,7 @@
 
 package org.apache.iceberg.stats;
 
+import java.nio.ByteBuffer;
 import java.util.Optional;
 import org.apache.iceberg.io.InMemoryOutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -33,7 +34,7 @@
 
 public class TestStatsWriter {
   @Test
-  public void testEmptyFooterCompressed() throws Exception {
+  public void testEmptyFooterCompressed() {
     InMemoryOutputFile outputFile = new InMemoryOutputFile();
     StatsWriter writer = new StatsWriter(outputFile);
     writer.setCompressFooter(true);
@@ -91,13 +92,13 @@ private void testWriteMetric(Optional<StatsCompressionCodec> compression, String
     try (StatsWriter writer = new StatsWriter(outputFile)) {
       writer.addFileProperty("writer.version", "1234");
 
-      writer.append("some-blob", ImmutableSet.of(1), "abcdefghi".getBytes(UTF_8), 0, 9, compression);
+      writer.append("some-blob", ImmutableSet.of(1), ByteBuffer.wrap("abcdefghi".getBytes(UTF_8)), compression);
 
       /// "xxx"s are stripped away by data offsets
       byte[] bytes =
           "xxx some blob \u0000 binary data 🤯 that is not very very very very very very long, is it? xxx".getBytes(
               UTF_8);
-      writer.append("some-other-blob", ImmutableSet.of(2), bytes, 4, bytes.length - 8, compression);
+      writer.append("some-other-blob", ImmutableSet.of(2), ByteBuffer.wrap(bytes, 4, bytes.length - 8), compression);
     }
 
     byte[] expected = readTestResource(expectedResource);