Skip to content

Commit 9430d36

Browse files
findepifindepi
committed
Add reader and writer for Puffin, indexes and stats file format
1 parent a42fbb5 commit 9430d36

24 files changed

Lines changed: 1710 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/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+
}

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

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
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 com.fasterxml.jackson.core.JsonGenerator;
23+
import com.fasterxml.jackson.databind.JsonNode;
24+
import java.io.IOException;
25+
import java.io.StringWriter;
26+
import java.io.UncheckedIOException;
27+
import java.util.List;
28+
import java.util.Map;
29+
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
30+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
31+
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
32+
import org.apache.iceberg.util.JsonUtil;
33+
34+
public final class FileMetadataParser {
35+
36+
private FileMetadataParser() {
37+
}
38+
39+
private static final String BLOBS = "blobs";
40+
private static final String PROPERTIES = "properties";
41+
42+
private static final String TYPE = "type";
43+
private static final String FIELDS = "fields";
44+
private static final String OFFSET = "offset";
45+
private static final String LENGTH = "length";
46+
private static final String COMPRESSION_CODEC = "compression-codec";
47+
48+
public static String toJson(FileMetadata fileMetadata, boolean pretty) {
49+
try {
50+
StringWriter writer = new StringWriter();
51+
JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
52+
if (pretty) {
53+
generator.useDefaultPrettyPrinter();
54+
}
55+
toJson(fileMetadata, generator);
56+
generator.flush();
57+
return writer.toString();
58+
} catch (IOException e) {
59+
throw new UncheckedIOException("Failed to write json for: " + fileMetadata, e);
60+
}
61+
}
62+
63+
public static FileMetadata fromJson(String json) {
64+
try {
65+
return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
66+
} catch (IOException e) {
67+
throw new UncheckedIOException(e);
68+
}
69+
}
70+
71+
static FileMetadata fromJson(JsonNode json) {
72+
return fileMetadataFromJson(json);
73+
}
74+
75+
static void toJson(FileMetadata fileMetadata, JsonGenerator generator) throws IOException {
76+
generator.writeStartObject();
77+
78+
generator.writeArrayFieldStart(BLOBS);
79+
for (BlobMetadata blobMetadata : fileMetadata.blobs()) {
80+
toJson(blobMetadata, generator);
81+
}
82+
generator.writeEndArray();
83+
84+
generator.writeObjectFieldStart(PROPERTIES);
85+
for (Map.Entry<String, String> entry : fileMetadata.properties().entrySet()) {
86+
generator.writeStringField(entry.getKey(), entry.getValue());
87+
}
88+
generator.writeEndObject();
89+
90+
generator.writeEndObject();
91+
}
92+
93+
static FileMetadata fileMetadataFromJson(JsonNode json) {
94+
95+
ImmutableList.Builder<BlobMetadata> blobs = ImmutableList.builder();
96+
JsonNode blobsJson = json.get(BLOBS);
97+
Preconditions.checkArgument(blobsJson != null && blobsJson.isArray(),
98+
"Cannot parse blobs from non-array: %s", blobsJson);
99+
for (JsonNode blobJson : blobsJson) {
100+
blobs.add(blobMetadataFromJson(blobJson));
101+
}
102+
103+
Map<String, String> properties = ImmutableMap.of();
104+
JsonNode propertiesJson = json.get(PROPERTIES);
105+
if (propertiesJson != null) {
106+
properties = JsonUtil.getStringMap(PROPERTIES, json);
107+
}
108+
109+
return new FileMetadata(
110+
blobs.build(),
111+
properties);
112+
}
113+
114+
static void toJson(BlobMetadata blobMetadata, JsonGenerator generator) throws IOException {
115+
generator.writeStartObject();
116+
117+
generator.writeStringField(TYPE, blobMetadata.type());
118+
119+
generator.writeArrayFieldStart(FIELDS);
120+
for (int field : blobMetadata.inputFields()) {
121+
generator.writeNumber(field);
122+
}
123+
generator.writeEndArray();
124+
125+
generator.writeNumberField(OFFSET, blobMetadata.offset());
126+
generator.writeNumberField(LENGTH, blobMetadata.length());
127+
128+
if (blobMetadata.compressionCodec() != null) {
129+
generator.writeStringField(COMPRESSION_CODEC, blobMetadata.compressionCodec());
130+
}
131+
132+
if (!blobMetadata.properties().isEmpty()) {
133+
generator.writeObjectFieldStart(PROPERTIES);
134+
for (Map.Entry<String, String> entry : blobMetadata.properties().entrySet()) {
135+
generator.writeStringField(entry.getKey(), entry.getValue());
136+
}
137+
generator.writeEndObject();
138+
}
139+
140+
generator.writeEndObject();
141+
}
142+
143+
static BlobMetadata blobMetadataFromJson(JsonNode json) {
144+
String type = JsonUtil.getString(TYPE, json);
145+
List<Integer> fields = JsonUtil.getIntegerList(FIELDS, json);
146+
long offset = JsonUtil.getLong(OFFSET, json);
147+
long length = JsonUtil.getLong(LENGTH, json);
148+
String compressionCodec = JsonUtil.getStringOrNull(COMPRESSION_CODEC, json);
149+
Map<String, String> properties = ImmutableMap.of();
150+
JsonNode propertiesJson = json.get(PROPERTIES);
151+
if (propertiesJson != null) {
152+
properties = JsonUtil.getStringMap(PROPERTIES, json);
153+
}
154+
155+
156+
return new BlobMetadata(
157+
type,
158+
fields,
159+
offset,
160+
length,
161+
compressionCodec,
162+
properties);
163+
}
164+
}

0 commit comments

Comments
 (0)