Skip to content

Commit be6bb83

Browse files
findepifindepi
committed
Add reader and writer for Puffin, indexes and stats file format
1 parent 547152c commit be6bb83

26 files changed

Lines changed: 1787 additions & 0 deletions

‎build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -220,6 +220,7 @@ project(':iceberg-core') {
220220
exclude group: 'org.tukaani' // xz compression is not supported
221221
}
222222

223+
implementation 'io.airlift:aircompressor'
223224
implementation 'org.apache.httpcomponents.client5:httpclient5'
224225
implementation "com.fasterxml.jackson.core:jackson-databind"
225226
implementation "com.fasterxml.jackson.core:jackson-core"

‎bundled-guava/src/main/java/org/apache/iceberg/GuavaClasses.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@
4949
import com.google.common.hash.Hashing;
5050
import com.google.common.io.CountingOutputStream;
5151
import com.google.common.io.Files;
52+
import com.google.common.io.Resources;
5253
import com.google.common.primitives.Bytes;
5354
import com.google.common.util.concurrent.MoreExecutors;
5455
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -89,6 +90,7 @@ public class GuavaClasses {
8990
Hashing.class.getName();
9091
Files.class.getName();
9192
Bytes.class.getName();
93+
Resources.class.getName();
9294
MoreExecutors.class.getName();
9395
ThreadFactoryBuilder.class.getName();
9496
Iterables.class.getName();

‎core/src/main/java/org/apache/iceberg/io/IOUtil.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,16 @@
2222
import java.io.EOFException;
2323
import java.io.IOException;
2424
import java.io.InputStream;
25+
import java.io.OutputStream;
26+
import java.nio.ByteBuffer;
2527

2628
public class IOUtil {
2729
// not meant to be instantiated
2830
private IOUtil() {
2931
}
3032

33+
private static final int WRITE_CHUNK_SIZE = 8192;
34+
3135
/**
3236
* Reads into a buffer from a stream, making multiple read calls if necessary.
3337
*
@@ -46,6 +50,21 @@ public static void readFully(InputStream stream, byte[] bytes, int offset, int l
4650
}
4751
}
4852

53+
/**
54+
* Writes a buffer into a stream, making multiple write calls if necessary.
55+
*/
56+
public static void writeFully(OutputStream outputStream, ByteBuffer buffer) throws IOException {
57+
if (!buffer.hasRemaining()) {
58+
return;
59+
}
60+
byte[] chunk = new byte[WRITE_CHUNK_SIZE];
61+
while (buffer.hasRemaining()) {
62+
int chunkSize = Math.min(chunk.length, buffer.remaining());
63+
buffer.get(chunk, 0, chunkSize);
64+
outputStream.write(chunk, 0, chunkSize);
65+
}
66+
}
67+
4968
/**
5069
* Reads into a buffer from a stream, making multiple read calls if necessary
5170
* returning the number of bytes read until end of stream.

‎core/src/main/java/org/apache/iceberg/puffin/Blob.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iceberg.puffin;
21+
22+
import java.nio.ByteBuffer;
23+
import java.util.List;
24+
import java.util.Map;
25+
import javax.annotation.Nullable;
26+
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
27+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
28+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
29+
30+
public final class Blob {
31+
private final String type;
32+
private final List<Integer> inputFields;
33+
private final ByteBuffer blobData;
34+
private final PuffinCompressionCodec requestedCompression;
35+
private final Map<String, String> properties;
36+
37+
public Blob(String type, List<Integer> inputFields, ByteBuffer blobData) {
38+
this(type, inputFields, blobData, null, ImmutableMap.of());
39+
}
40+
41+
public Blob(
42+
String type, List<Integer> inputFields, ByteBuffer blobData,
43+
@Nullable PuffinCompressionCodec requestedCompression, Map<String, String> properties) {
44+
Preconditions.checkNotNull(type, "type is null");
45+
Preconditions.checkNotNull(inputFields, "inputFields is null");
46+
Preconditions.checkNotNull(blobData, "blobData is null");
47+
Preconditions.checkNotNull(properties, "properties is null");
48+
this.type = type;
49+
this.inputFields = ImmutableList.copyOf(inputFields);
50+
this.blobData = blobData;
51+
this.requestedCompression = requestedCompression;
52+
this.properties = ImmutableMap.copyOf(properties);
53+
}
54+
55+
public String type() {
56+
return type;
57+
}
58+
59+
public List<Integer> inputFields() {
60+
return inputFields;
61+
}
62+
63+
public ByteBuffer blobData() {
64+
return blobData;
65+
}
66+
67+
@Nullable
68+
public PuffinCompressionCodec requestedCompression() {
69+
return requestedCompression;
70+
}
71+
72+
public Map<String, String> properties() {
73+
return properties;
74+
}
75+
}

‎core/src/main/java/org/apache/iceberg/puffin/BlobMetadata.java‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iceberg.puffin;
21+
22+
import java.util.List;
23+
import java.util.Map;
24+
import javax.annotation.Nullable;
25+
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
26+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
27+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
28+
29+
public class BlobMetadata {
30+
private final String type;
31+
private final List<Integer> inputFields;
32+
private final long offset;
33+
private final long length;
34+
private final String compressionCodec;
35+
private final Map<String, String> properties;
36+
37+
public BlobMetadata(
38+
String type, List<Integer> inputFields, long offset, long length,
39+
@Nullable String compressionCodec, Map<String, String> properties) {
40+
Preconditions.checkNotNull(type, "type is null");
41+
Preconditions.checkNotNull(inputFields, "inputFields is null");
42+
Preconditions.checkNotNull(properties, "properties is null");
43+
this.type = type;
44+
this.inputFields = ImmutableList.copyOf(inputFields);
45+
this.offset = offset;
46+
this.length = length;
47+
this.compressionCodec = compressionCodec;
48+
this.properties = ImmutableMap.copyOf(properties);
49+
}
50+
51+
public String type() {
52+
return type;
53+
}
54+
55+
public List<Integer> inputFields() {
56+
return inputFields;
57+
}
58+
59+
/**
60+
* Offset in the file
61+
*/
62+
public long offset() {
63+
return offset;
64+
}
65+
66+
/**
67+
* Length in the file
68+
*/
69+
public long length() {
70+
return length;
71+
}
72+
73+
@Nullable
74+
public String compressionCodec() {
75+
return compressionCodec;
76+
}
77+
78+
public Map<String, String> properties() {
79+
return properties;
80+
}
81+
}

‎core/src/main/java/org/apache/iceberg/puffin/FileMetadata.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iceberg.puffin;
21+
22+
import java.util.List;
23+
import java.util.Map;
24+
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
25+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
26+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
27+
28+
public class FileMetadata {
29+
private final List<BlobMetadata> blobs;
30+
private final Map<String, String> properties;
31+
32+
public FileMetadata(List<BlobMetadata> blobs, Map<String, String> properties) {
33+
Preconditions.checkNotNull(blobs, "blobs is null");
34+
Preconditions.checkNotNull(properties, "properties is null");
35+
this.blobs = ImmutableList.copyOf(blobs);
36+
this.properties = ImmutableMap.copyOf(properties);
37+
}
38+
39+
public List<BlobMetadata> blobs() {
40+
return blobs;
41+
}
42+
43+
public Map<String, String> properties() {
44+
return properties;
45+
}
46+
}

0 commit comments

Comments
 (0)