-
Notifications
You must be signed in to change notification settings - Fork 2.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Core: Key metadata in Avro format #6450
Changes from 13 commits
c890c7b
dc729c8
8980aa9
786ea38
364dc54
af0c163
8fcd041
14748cf
b0985ca
fe9be9d
bab1e7e
a385a0a
0fe6913
aa55b7f
1f2ae8a
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
/* | ||
* 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.data.avro; | ||
|
||
import java.io.IOException; | ||
import java.io.InputStream; | ||
import java.util.function.Function; | ||
import org.apache.avro.AvroRuntimeException; | ||
import org.apache.avro.Schema; | ||
import org.apache.avro.io.BinaryDecoder; | ||
import org.apache.avro.io.DatumReader; | ||
import org.apache.avro.io.DecoderFactory; | ||
import org.apache.avro.message.MessageDecoder; | ||
import org.apache.iceberg.avro.ProjectionDatumReader; | ||
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; | ||
|
||
public class RawDecoder<D> extends MessageDecoder.BaseDecoder<D> { | ||
private static final ThreadLocal<BinaryDecoder> DECODER = new ThreadLocal<>(); | ||
|
||
private final DatumReader<D> reader; | ||
|
||
/** | ||
* Creates a new {@link MessageDecoder} that constructs datum instances described by the {@link | ||
* Schema readSchema}. | ||
* | ||
* <p>The {@code readSchema} is used for the expected schema and the {@code writeSchema} is the | ||
* schema used to decode buffers. The {@code writeSchema} must be the schema that was used to | ||
* encode all buffers decoded by this class. | ||
*/ | ||
public RawDecoder( | ||
org.apache.iceberg.Schema readSchema, | ||
Function<Schema, DatumReader<?>> readerFunction, | ||
Schema writeSchema) { | ||
this.reader = new ProjectionDatumReader<>(readerFunction, readSchema, ImmutableMap.of(), null); | ||
this.reader.setSchema(writeSchema); | ||
} | ||
|
||
@Override | ||
public D decode(InputStream stream, D reuse) { | ||
BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(stream, DECODER.get()); | ||
DECODER.set(decoder); | ||
try { | ||
return reader.read(reuse, decoder); | ||
} catch (IOException e) { | ||
throw new AvroRuntimeException("Decoding datum failed", e); | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,134 @@ | ||
/* | ||
* 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.encryption; | ||
|
||
import static org.apache.iceberg.types.Types.NestedField.optional; | ||
import static org.apache.iceberg.types.Types.NestedField.required; | ||
|
||
import java.io.IOException; | ||
import java.io.UncheckedIOException; | ||
import java.nio.ByteBuffer; | ||
import java.util.Map; | ||
import org.apache.avro.generic.IndexedRecord; | ||
import org.apache.iceberg.Schema; | ||
import org.apache.iceberg.avro.AvroSchemaUtil; | ||
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; | ||
import org.apache.iceberg.types.Types; | ||
|
||
class KeyMetadata implements EncryptionKeyMetadata, IndexedRecord { | ||
private static final byte V1 = 1; | ||
private static final Schema SCHEMA_V1 = | ||
new Schema( | ||
required(0, "encryption_key", Types.BinaryType.get()), | ||
optional(1, "aad_prefix", Types.BinaryType.get())); | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
private static final org.apache.avro.Schema AVRO_SCHEMA_V1 = | ||
AvroSchemaUtil.convert(SCHEMA_V1, KeyMetadata.class.getCanonicalName()); | ||
|
||
private static final Map<Byte, Schema> schemaVersions = ImmutableMap.of(V1, SCHEMA_V1); | ||
private static final Map<Byte, org.apache.avro.Schema> avroSchemaVersions = | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. These should be ALL_CAPS as well. |
||
ImmutableMap.of(V1, AVRO_SCHEMA_V1); | ||
|
||
private static final KeyMetadataEncoder keyMetadataEncoder = new KeyMetadataEncoder(V1); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Shouldn't all the |
||
private static final KeyMetadataDecoder keyMetadataDecoder = new KeyMetadataDecoder(V1); | ||
|
||
private ByteBuffer encryptionKey; | ||
private ByteBuffer aadPrefix; | ||
private org.apache.avro.Schema avroSchema; | ||
|
||
/** Used by Avro reflection to instantiate this class * */ | ||
KeyMetadata(org.apache.avro.Schema avroSchema) { | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
this.avroSchema = avroSchema; | ||
} | ||
|
||
KeyMetadata(ByteBuffer encryptionKey, ByteBuffer aadPrefix) { | ||
this.encryptionKey = encryptionKey; | ||
this.aadPrefix = aadPrefix; | ||
this.avroSchema = AVRO_SCHEMA_V1; | ||
} | ||
|
||
static Map<Byte, Schema> supportedSchemaVersions() { | ||
return schemaVersions; | ||
} | ||
|
||
static Map<Byte, org.apache.avro.Schema> supportedAvroSchemaVersions() { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why is this exposed? I don't think that we want to expose Avro schemas in the API. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is for package-internal usage by KeyMetadataEncoder and KeyMetadataDecoder - they need an access to the map of supported schema versions. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. At least we can remove it if we refactor. |
||
return avroSchemaVersions; | ||
} | ||
|
||
ByteBuffer encryptionKey() { | ||
return encryptionKey; | ||
} | ||
|
||
ByteBuffer aadPrefix() { | ||
return aadPrefix; | ||
} | ||
|
||
static KeyMetadata parse(ByteBuffer buffer) { | ||
try { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Please adhere to style guidelines throughout. There should not be code blocks that aren't separated by whitespace. |
||
return keyMetadataDecoder.decode(buffer); | ||
} catch (IOException e) { | ||
throw new UncheckedIOException("Failed to parse envelope encryption metadata", e); | ||
} | ||
} | ||
|
||
@Override | ||
public ByteBuffer buffer() { | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
try { | ||
return keyMetadataEncoder.encode(this); | ||
} catch (IOException e) { | ||
throw new UncheckedIOException("Failed to serialize envelope key metadata", e); | ||
} | ||
} | ||
|
||
@Override | ||
public EncryptionKeyMetadata copy() { | ||
KeyMetadata metadata = new KeyMetadata(encryptionKey(), aadPrefix()); | ||
return metadata; | ||
} | ||
|
||
@Override | ||
public void put(int i, Object v) { | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
switch (i) { | ||
case 0: | ||
this.encryptionKey = (ByteBuffer) v; | ||
return; | ||
case 1: | ||
this.aadPrefix = (ByteBuffer) v; | ||
return; | ||
default: | ||
// ignore the object, it must be from a newer version of the format | ||
} | ||
} | ||
|
||
@Override | ||
public Object get(int i) { | ||
switch (i) { | ||
case 0: | ||
return encryptionKey; | ||
case 1: | ||
return aadPrefix; | ||
default: | ||
throw new UnsupportedOperationException("Unknown field ordinal: " + i); | ||
} | ||
} | ||
|
||
@Override | ||
public org.apache.avro.Schema getSchema() { | ||
return avroSchema; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,76 @@ | ||
/* | ||
* 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.encryption; | ||
|
||
import java.io.IOException; | ||
import java.io.InputStream; | ||
import java.io.UncheckedIOException; | ||
import java.util.Map; | ||
import org.apache.avro.Schema; | ||
import org.apache.avro.message.MessageDecoder; | ||
import org.apache.iceberg.avro.GenericAvroReader; | ||
import org.apache.iceberg.data.avro.RawDecoder; | ||
import org.apache.iceberg.relocated.com.google.common.collect.MapMaker; | ||
|
||
class KeyMetadataDecoder extends MessageDecoder.BaseDecoder<KeyMetadata> { | ||
private final org.apache.iceberg.Schema readSchema; | ||
private final Map<Byte, RawDecoder<KeyMetadata>> decoders = new MapMaker().makeMap(); | ||
|
||
/** | ||
* Creates a new decoder that constructs key metadata instances described by schema version. | ||
* | ||
* <p>The {@code readSchemaVersion} is as used the version of the expected (read) schema. Datum | ||
* instances created by this class will are described by the expected schema. | ||
*/ | ||
KeyMetadataDecoder(byte readSchemaVersion) { | ||
this.readSchema = KeyMetadata.supportedSchemaVersions().get(readSchemaVersion); | ||
} | ||
|
||
@Override | ||
public KeyMetadata decode(InputStream stream, KeyMetadata reuse) throws IOException { | ||
rdblue marked this conversation as resolved.
Show resolved
Hide resolved
|
||
byte writeSchemaVersion; | ||
|
||
try { | ||
writeSchemaVersion = (byte) stream.read(); | ||
} catch (IOException e) { | ||
throw new UncheckedIOException("Failed to read the version byte", e); | ||
} | ||
|
||
if (writeSchemaVersion < 0) { | ||
throw new IOException("Version byte - end of stream reached"); | ||
} | ||
|
||
Schema writeSchema = KeyMetadata.supportedAvroSchemaVersions().get(writeSchemaVersion); | ||
|
||
if (writeSchema == null) { | ||
throw new UnsupportedOperationException( | ||
"Cannot resolve schema for version: " + writeSchemaVersion); | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This decoder method looks good, other than using a buffer to read the version byte. |
||
|
||
RawDecoder<KeyMetadata> decoder = decoders.get(writeSchemaVersion); | ||
|
||
if (decoder == null) { | ||
decoder = new RawDecoder<>(readSchema, GenericAvroReader::create, writeSchema); | ||
|
||
decoders.put(writeSchemaVersion, decoder); | ||
} | ||
|
||
return decoder.decode(stream, reuse); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Catch IOException and wrap with UncheckedIOException here? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the decoder.decode method doesn't throw an IOException |
||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Note about this to myself for later...
This needs to use
GenericAvroReader
for a couple reasons. First,GenericAvroReader
is the only one that supports creating records with a specific class. We could extend support so thatDataReader
can createStructLike
instances, but that hits the second reason: using Iceberg generics will use different representations for some types, like usingLocalDate
instead ofint
fordate
type. The representation differences don't affect theKeyMetadata
class right now, but it's still more correct to use the internal representation.