Add Indexes and Stats format reader and writer

Add reader and writer for the Index and Statistics File Format.

Author: findepi

build.gradle

@@ -217,9 +217,11 @@ project(':iceberg-core') {
       exclude group: 'org.tukaani' // xz compression is not supported
     }
 
+    implementation 'io.airlift:aircompressor'
     implementation 'org.apache.httpcomponents.client5:httpclient5'
     implementation "com.fasterxml.jackson.core:jackson-databind"
     implementation "com.fasterxml.jackson.core:jackson-core"
+    implementation "com.fasterxml.jackson.datatype:jackson-datatype-jdk8"
     implementation "com.github.ben-manes.caffeine:caffeine"
     implementation "org.roaringbitmap:RoaringBitmap"
     compileOnly("org.apache.hadoop:hadoop-client") {

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

@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.stats;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+
+public class BlobMetadata {
+  private final String type;
+  private final List<Integer> columns;
+  private final long fileOffset;
+  private final long length;
+  private final Optional<String> compressionCodec;
+
+  @JsonCreator
+  public BlobMetadata(
+      @JsonProperty("type") String type,
+      @JsonProperty("columns") List<Integer> columns,
+      @JsonProperty("offset") long fileOffset,
+      @JsonProperty("length") long length,
+      @JsonProperty("compression_codec") Optional<String> compressionCodec) {
+    this.type = Objects.requireNonNull(type, "type is null");
+    this.columns = ImmutableList.copyOf(Objects.requireNonNull(columns, "columns is null"));
+    this.fileOffset = fileOffset;
+    this.length = length;
+    this.compressionCodec = Objects.requireNonNull(compressionCodec, "compressionCodec is null");
+  }
+
+  public String getType() {
+    return type;
+  }
+
+  public List<Integer> getColumns() {
+    return columns;
+  }
+
+  public long getFileOffset() {
+    return fileOffset;
+  }
+
+  public long getLength() {
+    return length;
+  }
+
+  public Optional<String> getCompressionCodec() {
+    return compressionCodec;
+  }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * 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.iceberg.stats;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+
+public class FileMetadata {
+  private final List<BlobMetadata> blobs;
+  private final Map<String, String> properties;
+
+  @JsonCreator
+  public FileMetadata(
+      @JsonProperty("blobs") List<BlobMetadata> blobs,
+      @JsonProperty("properties") Map<String, String> properties) {
+    this.blobs = ImmutableList.copyOf(Objects.requireNonNull(blobs, "blobs is null"));
+    this.properties = ImmutableMap.copyOf(Objects.requireNonNull(properties, "properties is null"));
+  }
+
+  public List<BlobMetadata> getBlobs() {
+    return blobs;
+  }
+
+  public Map<String, String> getProperties() {
+    return properties;
+  }
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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.iceberg.stats;
+
+public final class StandardBlobTypes {
+  private StandardBlobTypes() {
+  }
+
+  /**
+   * 8-bytes integer stored little-endian and representing number of distinct values
+   */
+  public static final String NDV_LONG_LITTLE_ENDIAN = "ndv-long-little-endian";
+
+  /**
+   * A serialized form of a "compact" Theta sketch produced by the <a href="https://datasketches.apache.org/">Apache DataSketches</a> library
+   */
+  public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+}

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

@@ -0,0 +1,48 @@
+/*
+ * 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.iceberg.stats;
+
+import java.util.Objects;
+
+public enum StatsCompressionCodec {
+  LZ4("lz4"),
+  ZSTD("zstd"),
+  /**/;
+
+  private final String codecName;
+
+  StatsCompressionCodec(String codecName) {
+    this.codecName = codecName;
+  }
+
+  public String getCodecName() {
+    return codecName;
+  }
+
+  public static StatsCompressionCodec forName(String codecName) {
+    Objects.requireNonNull(codecName, "codecName is null");
+    for (StatsCompressionCodec value : values()) {
+      if (value.getCodecName().equals(codecName)) {
+        return value;
+      }
+    }
+    throw new IllegalArgumentException("Unrecognized codec name " + codecName);
+  }
+}

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

@@ -0,0 +1,136 @@
+/*
+ * 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.iceberg.stats;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import io.airlift.compress.Compressor;
+import io.airlift.compress.lz4.Lz4Compressor;
+import io.airlift.compress.lz4.Lz4Decompressor;
+import io.airlift.compress.zstd.ZstdCompressor;
+import io.airlift.compress.zstd.ZstdDecompressor;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.math.BigInteger;
+import java.util.Arrays;
+import org.apache.commons.codec.binary.Base16;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+final class StatsFormat {
+  private StatsFormat() {
+  }
+
+  static final int CURRENT_FORMAT_VERSION = 1;
+
+  private static final byte[] MAGIC = new Base16().decode("50464953");
+  static final int MAGIC_AS_NUMBER_LE = new BigInteger(swap(MAGIC)).intValueExact();
+
+  static final int SUPPORTED_FLAGS = 0b1;
+  static final int FLAG_COMPRESSED = 0b1;
+
+  // TODO use LZ4 with frames, see https://trinodb.slack.com/archives/CP1MUNEUX/p1649676596198729
+  @Deprecated
+  static final int LZ4_UNCOMPRESSED_MAX_SIZE = 1024 * 1024;
+
+  // Not using JsonUtil.mapper() to avoid changing all JSON parsing
+  static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
+      .registerModule(new Jdk8Module());
+
+  static byte[] getMagic() {
+    return MAGIC.clone();
+  }
+
+  static void writeIntegerLittleEndian(OutputStream outputStream, int value) throws IOException {
+    outputStream.write(0xFF & value);
+    outputStream.write(0xFF & value >> 8);
+    outputStream.write(0xFF & value >> 16);
+    outputStream.write(0xFF & value >> 24);
+  }
+
+  static int readIntegerLittleEndian(byte[] data, int offset) {
+    return Byte.toUnsignedInt(data[offset]) |
+        (Byte.toUnsignedInt(data[offset + 1]) << 8) |
+        (Byte.toUnsignedInt(data[offset + 2]) << 16) |
+        (Byte.toUnsignedInt(data[offset + 3]) << 24);
+  }
+
+  static byte[] compressFooterPayload(byte[] payload) {
+    return compress(StatsCompressionCodec.LZ4, payload, 0, payload.length);
+  }
+
+  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 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) {
+    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);
+  }
+
+  private static Compressor getCompressor(StatsCompressionCodec codec) {
+    switch (codec) {
+      case LZ4:
+        return new Lz4Compressor();
+      case ZSTD:
+        return new ZstdCompressor();
+    }
+    throw new IllegalArgumentException("Unsupported codec: " + codec);
+  }
+
+  private static byte[] decompress(StatsCompressionCodec codec, byte[] input, int inputOffset, int inputLength) {
+    switch (codec) {
+      case LZ4: {
+        byte[] decompressed = new byte[LZ4_UNCOMPRESSED_MAX_SIZE];
+        int decompressedLength =
+            new Lz4Decompressor().decompress(input, inputOffset, inputLength, decompressed, 0, decompressed.length);
+        return Arrays.copyOf(decompressed, decompressedLength);
+      }
+
+      case ZSTD: {
+        byte[] decompressed =
+            new byte[Math.toIntExact(ZstdDecompressor.getDecompressedSize(input, inputOffset, inputLength))];
+        int decompressedLength =
+            new ZstdDecompressor().decompress(input, inputOffset, inputLength, decompressed, 0, decompressed.length);
+        Preconditions.checkState(decompressedLength == decompressed.length, "Invalid decompressed length");
+        return decompressed;
+      }
+    }
+
+    throw new UnsupportedOperationException("Unsupported codec: " + codec);
+  }
+
+  private static byte[] swap(byte[] bytes) {
+    byte[] swapped = new byte[bytes.length];
+    for (int i = 0; i < swapped.length; i++) {
+      swapped[i] = bytes[swapped.length - i - 1];
+    }
+    return swapped;
+  }
+}