You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/02/18 22:16:59 UTC
[incubator-iceberg] branch master updated: Add stream-level
encryption API. (#85)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new b25c1a7 Add stream-level encryption API. (#85)
b25c1a7 is described below
commit b25c1a79ca880a652c4d2d3f3e1a2e23447c60f6
Author: mccheah <mc...@palantir.com>
AuthorDate: Mon Feb 18 14:16:55 2019 -0800
Add stream-level encryption API. (#85)
---
.../main/java/com/netflix/iceberg/DataFile.java | 12 +++-
api/src/main/java/com/netflix/iceberg/Table.java | 11 +++-
.../iceberg/encryption/EncryptedInputFile.java | 45 +++++++++++++
.../iceberg/encryption/EncryptedOutputFile.java | 46 +++++++++++++
.../iceberg/encryption/EncryptionKeyMetadata.java | 38 +++++++++++
.../iceberg/encryption/EncryptionManager.java | 75 ++++++++++++++++++++++
.../test/java/com/netflix/iceberg/TestHelpers.java | 6 ++
.../main/java/com/netflix/iceberg/DataFiles.java | 20 +++++-
.../java/com/netflix/iceberg/GenericDataFile.java | 24 ++++++-
.../netflix/iceberg/SerializableByteBufferMap.java | 22 +------
.../java/com/netflix/iceberg/TableOperations.java | 10 +++
.../iceberg/encryption/BaseEncryptedInputFile.java | 43 +++++++++++++
.../encryption/BaseEncryptedOutputFile.java | 43 +++++++++++++
.../encryption/BaseEncryptionKeyMetadata.java | 43 +++++++++++++
.../netflix/iceberg/encryption/EncryptedFiles.java | 58 +++++++++++++++++
.../iceberg/encryption/EncryptionKeyMetadatas.java | 35 ++++++++++
.../java/com/netflix/iceberg/util/ByteBuffers.java | 56 ++++++++++++++++
.../com/netflix/iceberg/spark/source/Writer.java | 9 ---
18 files changed, 562 insertions(+), 34 deletions(-)
diff --git a/api/src/main/java/com/netflix/iceberg/DataFile.java b/api/src/main/java/com/netflix/iceberg/DataFile.java
index d0e728e..7e432c9 100644
--- a/api/src/main/java/com/netflix/iceberg/DataFile.java
+++ b/api/src/main/java/com/netflix/iceberg/DataFile.java
@@ -19,6 +19,7 @@
package com.netflix.iceberg;
+import com.netflix.iceberg.encryption.EncryptionKeyMetadata;
import com.netflix.iceberg.types.Types.BinaryType;
import com.netflix.iceberg.types.Types.IntegerType;
import com.netflix.iceberg.types.Types.ListType;
@@ -57,8 +58,9 @@ public interface DataFile {
optional(125, "lower_bounds", MapType.ofRequired(126, 127,
IntegerType.get(), BinaryType.get())),
optional(128, "upper_bounds", MapType.ofRequired(129, 130,
- IntegerType.get(), BinaryType.get()))
- // NEXT ID TO ASSIGN: 131
+ IntegerType.get(), BinaryType.get())),
+ optional(131, "key_metadata", BinaryType.get())
+ // NEXT ID TO ASSIGN: 132
);
}
@@ -128,6 +130,12 @@ public interface DataFile {
Map<Integer, ByteBuffer> upperBounds();
/**
+ * @return metadata about how this file is encrypted, or null if the file is stored in plain
+ * text.
+ */
+ ByteBuffer keyMetadata();
+
+ /**
* Copies this {@link DataFile data file}. Manifest readers can reuse data file instances; use
* this method to copy data when collecting files from tasks.
*
diff --git a/api/src/main/java/com/netflix/iceberg/Table.java b/api/src/main/java/com/netflix/iceberg/Table.java
index 73e59cb..ccf3ab4 100644
--- a/api/src/main/java/com/netflix/iceberg/Table.java
+++ b/api/src/main/java/com/netflix/iceberg/Table.java
@@ -19,6 +19,7 @@
package com.netflix.iceberg;
+import com.netflix.iceberg.encryption.EncryptionManager;
import com.netflix.iceberg.io.FileIO;
import com.netflix.iceberg.io.LocationProvider;
import java.util.Map;
@@ -187,8 +188,16 @@ public interface Table {
FileIO io();
/**
+ * @return an {@link com.netflix.iceberg.encryption.EncryptionManager} to encrypt and decrypt
+ * data files.
+ */
+ default EncryptionManager encryption() {
+ // TODO coming soon
+ throw new UnsupportedOperationException("Encryption is a work in progress.");
+ }
+
+ /**
* @return a {@link LocationProvider} to provide locations for new data files
*/
LocationProvider locationProvider();
-
}
diff --git a/api/src/main/java/com/netflix/iceberg/encryption/EncryptedInputFile.java b/api/src/main/java/com/netflix/iceberg/encryption/EncryptedInputFile.java
new file mode 100644
index 0000000..eff540c
--- /dev/null
+++ b/api/src/main/java/com/netflix/iceberg/encryption/EncryptedInputFile.java
@@ -0,0 +1,45 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.io.InputFile;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Thin wrapper around an {@link InputFile} instance that is encrypted.
+ * <p>
+ * The {@link EncryptionManager} takes instances of these and uses the attached
+ * {@link #keyMetadata()} to find an encryption key and decrypt the enclosed
+ * {@link #encryptedInputFile()}.
+ */
+public interface EncryptedInputFile {
+
+ /**
+ * The {@link InputFile} that is reading raw encrypted bytes from the underlying file system.
+ */
+ InputFile encryptedInputFile();
+
+ /**
+ * Metadata pointing to some encryption key that would be used to decrypt the input file provided
+ * by {@link #encryptedInputFile()}.
+ */
+ EncryptionKeyMetadata keyMetadata();
+}
diff --git a/api/src/main/java/com/netflix/iceberg/encryption/EncryptedOutputFile.java b/api/src/main/java/com/netflix/iceberg/encryption/EncryptedOutputFile.java
new file mode 100644
index 0000000..53dd863
--- /dev/null
+++ b/api/src/main/java/com/netflix/iceberg/encryption/EncryptedOutputFile.java
@@ -0,0 +1,46 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.io.OutputFile;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Thin wrapper around a {@link OutputFile} that is encrypting bytes written to the underlying
+ * file system, via an encryption key that is symbolized by the enclosed
+ * {@link EncryptionKeyMetadata}.
+ * <p>
+ * The {@link EncryptionManager} returns instances of these when passed output files that should
+ * be encrypted as they are being written to the backing file system.
+ */
+public interface EncryptedOutputFile {
+
+ /**
+ * An OutputFile instance that encrypts the bytes that are written to its output streams.
+ */
+ OutputFile encryptingOutputFile();
+
+ /**
+ * Metadata about the encryption key that is being used to encrypt the associated
+ * {@link #encryptingOutputFile()}.
+ */
+ EncryptionKeyMetadata keyMetadata();
+}
diff --git a/api/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadata.java b/api/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadata.java
new file mode 100644
index 0000000..f88980c
--- /dev/null
+++ b/api/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadata.java
@@ -0,0 +1,38 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Light typedef over a ByteBuffer that indicates that the given bytes represent metadata about
+ * an encrypted data file's encryption key.
+ * <p>
+ * This is preferred over passing a ByteBuffer directly in order to be more explicit.
+ */
+public interface EncryptionKeyMetadata {
+
+ /**
+ * Opaque blob representing metadata about a file's encryption key.
+ */
+ ByteBuffer keyMetadata();
+
+ EncryptionKeyMetadata copy();
+}
diff --git a/api/src/main/java/com/netflix/iceberg/encryption/EncryptionManager.java b/api/src/main/java/com/netflix/iceberg/encryption/EncryptionManager.java
new file mode 100644
index 0000000..6555986
--- /dev/null
+++ b/api/src/main/java/com/netflix/iceberg/encryption/EncryptionManager.java
@@ -0,0 +1,75 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.google.common.collect.Iterables;
+import com.netflix.iceberg.io.InputFile;
+import com.netflix.iceberg.io.OutputFile;
+
+import java.io.Serializable;
+
+/**
+ * Module for encrypting and decrypting table data files.
+ * <p>
+ * This must be serializable because an instance may be instantiated in one place and sent across
+ * the wire in some Iceberg integrations, notably Spark.
+ */
+public interface EncryptionManager extends Serializable {
+
+ /**
+ * Given an {@link EncryptedInputFile#encryptedInputFile()} representing the raw encrypted bytes
+ * from the underlying file system, and given metadata about how the file was encrypted via
+ * {@link EncryptedInputFile#keyMetadata()}, return an {@link InputFile} that returns decrypted
+ * input streams.
+ */
+ InputFile decrypt(EncryptedInputFile encrypted);
+
+ /**
+ * Variant of {@link #decrypt(EncryptedInputFile)} that provides a sequence of files that all
+ * need to be decrypted in a single context.
+ * <p>
+ * By default this calls the single-file decryption method for each element in the iterator.
+ * Implementations can override this for a variety of optimizations. For example, an
+ * implementation can perform lookahead on the input iterator and fetch encryption keys in batch.
+ */
+ default Iterable<InputFile> decrypt(Iterable<EncryptedInputFile> encrypted) {
+ return Iterables.transform(encrypted, this::decrypt);
+ }
+
+ /**
+ * Given a handle on an {@link OutputFile} that writes raw bytes to the underlying file system,
+ * return a bundle of an {@link EncryptedOutputFile#encryptingOutputFile()} that writes encrypted
+ * bytes to the underlying file system, and the {@link EncryptedOutputFile#keyMetadata()} that
+ * points to the encryption key that is being used to encrypt this file.
+ */
+ EncryptedOutputFile encrypt(OutputFile rawOutput);
+
+ /**
+ * Variant of {@link #encrypt(OutputFile)} that provides a sequence of files that all need to be
+ * encrypted in a single context.
+ * <p>
+ * By default this calls the single-file encryption method for each element in the iterator.
+ * Implementations can override this for a variety of optimizations. For example, an
+ * implementation can perform lookahead on the input iterator and fetch encryption keys in batch.
+ */
+ default Iterable<EncryptedOutputFile> encrypt(Iterable<OutputFile> rawOutput) {
+ return Iterables.transform(rawOutput, this::encrypt);
+ }
+}
diff --git a/api/src/test/java/com/netflix/iceberg/TestHelpers.java b/api/src/test/java/com/netflix/iceberg/TestHelpers.java
index ceb1eed..70dab64 100644
--- a/api/src/test/java/com/netflix/iceberg/TestHelpers.java
+++ b/api/src/test/java/com/netflix/iceberg/TestHelpers.java
@@ -19,6 +19,7 @@
package com.netflix.iceberg;
+import com.netflix.iceberg.encryption.EncryptionKeyMetadata;
import com.netflix.iceberg.expressions.BoundPredicate;
import com.netflix.iceberg.expressions.Expression;
import com.netflix.iceberg.expressions.ExpressionVisitors;
@@ -370,6 +371,11 @@ public class TestHelpers {
}
@Override
+ public ByteBuffer keyMetadata() {
+ return null;
+ }
+
+ @Override
public DataFile copy() {
return this;
}
diff --git a/core/src/main/java/com/netflix/iceberg/DataFiles.java b/core/src/main/java/com/netflix/iceberg/DataFiles.java
index b3bac11..8692522 100644
--- a/core/src/main/java/com/netflix/iceberg/DataFiles.java
+++ b/core/src/main/java/com/netflix/iceberg/DataFiles.java
@@ -20,9 +20,11 @@
package com.netflix.iceberg;
import com.google.common.base.Preconditions;
+import com.netflix.iceberg.encryption.EncryptionKeyMetadata;
import com.netflix.iceberg.hadoop.HadoopInputFile;
import com.netflix.iceberg.io.InputFile;
import com.netflix.iceberg.types.Conversions;
+import com.netflix.iceberg.util.ByteBuffers;
import org.apache.hadoop.fs.FileStatus;
import java.nio.ByteBuffer;
import java.util.List;
@@ -185,6 +187,7 @@ public class DataFiles {
private Map<Integer, Long> nullValueCounts = null;
private Map<Integer, ByteBuffer> lowerBounds = null;
private Map<Integer, ByteBuffer> upperBounds = null;
+ private ByteBuffer keyMetadata = null;
public Builder() {
this.spec = null;
@@ -228,6 +231,8 @@ public class DataFiles {
this.nullValueCounts = toCopy.nullValueCounts();
this.lowerBounds = toCopy.lowerBounds();
this.upperBounds = toCopy.upperBounds();
+ this.keyMetadata = toCopy.keyMetadata() == null ? null
+ : ByteBuffers.copy(toCopy.keyMetadata());
return this;
}
@@ -301,6 +306,19 @@ public class DataFiles {
return this;
}
+ public Builder withEncryptionKeyMetadata(ByteBuffer keyMetadata) {
+ this.keyMetadata = keyMetadata;
+ return this;
+ }
+
+ public Builder withEncryptionKeyMetadata(EncryptionKeyMetadata keyMetadata) {
+ return withEncryptionKeyMetadata(keyMetadata.keyMetadata());
+ }
+
+ public Builder withEncryptionKeyMetadata(byte[] keyMetadata) {
+ return withEncryptionKeyMetadata(ByteBuffer.wrap(keyMetadata));
+ }
+
public DataFile build() {
Preconditions.checkArgument(filePath != null, "File path is required");
if (format == null) {
@@ -317,7 +335,7 @@ public class DataFiles {
return new GenericDataFile(
filePath, format, isPartitioned ? partitionData.copy() : null,
fileSizeInBytes, blockSizeInBytes, new Metrics(
- recordCount, columnSizes, valueCounts, nullValueCounts, lowerBounds, upperBounds));
+ recordCount, columnSizes, valueCounts, nullValueCounts, lowerBounds, upperBounds), keyMetadata);
}
}
}
diff --git a/core/src/main/java/com/netflix/iceberg/GenericDataFile.java b/core/src/main/java/com/netflix/iceberg/GenericDataFile.java
index 0828377..b5bfa06 100644
--- a/core/src/main/java/com/netflix/iceberg/GenericDataFile.java
+++ b/core/src/main/java/com/netflix/iceberg/GenericDataFile.java
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableMap;
import com.netflix.iceberg.avro.AvroSchemaUtil;
import com.netflix.iceberg.types.Type;
import com.netflix.iceberg.types.Types;
+import com.netflix.iceberg.util.ByteBuffers;
import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.specific.SpecificData;
import java.io.Serializable;
@@ -59,6 +60,7 @@ class GenericDataFile
private Map<Integer, Long> nullValueCounts = null;
private Map<Integer, ByteBuffer> lowerBounds = null;
private Map<Integer, ByteBuffer> upperBounds = null;
+ private ByteBuffer keyMetadata = null;
// cached schema
private transient org.apache.avro.Schema avroSchema = null;
@@ -165,6 +167,13 @@ class GenericDataFile
this.fromProjectionPos = null;
}
+ GenericDataFile(String filePath, FileFormat format, PartitionData partition,
+ long fileSizeInBytes, long blockSizeInBytes, Metrics metrics,
+ ByteBuffer keyMetadata) {
+ this(filePath, format, partition, fileSizeInBytes, blockSizeInBytes, metrics);
+ this.keyMetadata = keyMetadata;
+ }
+
/**
* Copy constructor.
*
@@ -187,6 +196,7 @@ class GenericDataFile
this.lowerBounds = toCopy.lowerBounds;
this.upperBounds = toCopy.upperBounds;
this.fromProjectionPos = toCopy.fromProjectionPos;
+ this.keyMetadata = toCopy.keyMetadata == null ? null : ByteBuffers.copy(toCopy.keyMetadata);
}
/**
@@ -261,6 +271,11 @@ class GenericDataFile
}
@Override
+ public ByteBuffer keyMetadata() {
+ return keyMetadata;
+ }
+
+ @Override
public org.apache.avro.Schema getSchema() {
if (avroSchema == null) {
this.avroSchema = getAvroSchema(partitionType);
@@ -315,8 +330,10 @@ class GenericDataFile
this.lowerBounds = SerializableByteBufferMap.wrap((Map<Integer, ByteBuffer>) v);
return;
case 12:
- this.upperBounds= SerializableByteBufferMap.wrap((Map<Integer, ByteBuffer>) v);
+ this.upperBounds = SerializableByteBufferMap.wrap((Map<Integer, ByteBuffer>) v);
return;
+ case 13:
+ this.keyMetadata = (ByteBuffer) v;
default:
// ignore the object, it must be from a newer version of the format
}
@@ -356,6 +373,8 @@ class GenericDataFile
return lowerBounds;
case 12:
return upperBounds;
+ case 13:
+ return keyMetadata;
default:
throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
}
@@ -370,7 +389,7 @@ class GenericDataFile
@Override
public int size() {
- return 13;
+ return 14;
}
@Override
@@ -402,6 +421,7 @@ class GenericDataFile
.add("null_value_counts", nullValueCounts)
.add("lower_bounds", lowerBounds)
.add("upper_bounds", upperBounds)
+ .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
.toString();
}
diff --git a/core/src/main/java/com/netflix/iceberg/SerializableByteBufferMap.java b/core/src/main/java/com/netflix/iceberg/SerializableByteBufferMap.java
index 18b1f96..f67834e 100644
--- a/core/src/main/java/com/netflix/iceberg/SerializableByteBufferMap.java
+++ b/core/src/main/java/com/netflix/iceberg/SerializableByteBufferMap.java
@@ -20,10 +20,11 @@
package com.netflix.iceberg;
import com.google.common.collect.Maps;
+import com.netflix.iceberg.util.ByteBuffers;
+
import java.io.ObjectStreamException;
import java.io.Serializable;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
@@ -85,30 +86,13 @@ class SerializableByteBufferMap implements Map<Integer, ByteBuffer>, Serializabl
int i = 0;
for (Map.Entry<Integer, ByteBuffer> entry : entries) {
keys[i] = entry.getKey();
- values[i] = copy(entry.getValue());
+ values[i] = ByteBuffers.toByteArray(entry.getValue());
i += 1;
}
return new MapSerializationProxy(keys, values);
}
- private byte[] copy(ByteBuffer buffer) {
- if (buffer.hasArray()) {
- byte[] array = buffer.array();
- if (buffer.arrayOffset() == 0 && buffer.position() == 0 && array.length == buffer.remaining()) {
- return array;
- } else {
- int start = buffer.arrayOffset() + buffer.position();
- int end = start + buffer.remaining();
- return Arrays.copyOfRange(array, start, end);
- }
- } else {
- byte[] bytes = new byte[buffer.remaining()];
- buffer.get(bytes);
- return bytes;
- }
- }
-
@Override
public int size() {
return wrapped.size();
diff --git a/core/src/main/java/com/netflix/iceberg/TableOperations.java b/core/src/main/java/com/netflix/iceberg/TableOperations.java
index 67f509a..e32c9a1 100644
--- a/core/src/main/java/com/netflix/iceberg/TableOperations.java
+++ b/core/src/main/java/com/netflix/iceberg/TableOperations.java
@@ -19,6 +19,7 @@
package com.netflix.iceberg;
+import com.netflix.iceberg.encryption.EncryptionManager;
import com.netflix.iceberg.io.FileIO;
import com.netflix.iceberg.io.LocationProvider;
import java.util.UUID;
@@ -62,6 +63,15 @@ public interface TableOperations {
FileIO io();
/**
+ * @return a {@link com.netflix.iceberg.encryption.EncryptionManager} to encrypt and decrypt
+ * data files.
+ */
+ default EncryptionManager encryption() {
+ // TODO coming soon
+ throw new UnsupportedOperationException("Encryption is a work in progress.");
+ }
+
+ /**
* Given the name of a metadata file, obtain the full path of that file using an appropriate base
* location of the implementation's choosing.
* <p>
diff --git a/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedInputFile.java b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedInputFile.java
new file mode 100644
index 0000000..540abe3
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedInputFile.java
@@ -0,0 +1,43 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.io.InputFile;
+
+class BaseEncryptedInputFile implements EncryptedInputFile {
+
+ private final InputFile encryptedInputFile;
+ private final EncryptionKeyMetadata keyMetadata;
+
+ public BaseEncryptedInputFile(InputFile encryptedInputFile, EncryptionKeyMetadata keyMetadata) {
+ this.encryptedInputFile = encryptedInputFile;
+ this.keyMetadata = keyMetadata;
+ }
+
+ @Override
+ public InputFile encryptedInputFile() {
+ return encryptedInputFile;
+ }
+
+ @Override
+ public EncryptionKeyMetadata keyMetadata() {
+ return keyMetadata;
+ }
+}
diff --git a/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedOutputFile.java b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedOutputFile.java
new file mode 100644
index 0000000..9cb8d06
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptedOutputFile.java
@@ -0,0 +1,43 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.io.OutputFile;
+
+class BaseEncryptedOutputFile implements EncryptedOutputFile {
+
+ private final OutputFile encryptingOutputFile;
+ private final EncryptionKeyMetadata keyMetadata;
+
+ BaseEncryptedOutputFile(OutputFile encryptingOutputFile, EncryptionKeyMetadata keyMetadata) {
+ this.encryptingOutputFile = encryptingOutputFile;
+ this.keyMetadata = keyMetadata;
+ }
+
+ @Override
+ public OutputFile encryptingOutputFile() {
+ return encryptingOutputFile;
+ }
+
+ @Override
+ public EncryptionKeyMetadata keyMetadata() {
+ return keyMetadata;
+ }
+}
diff --git a/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptionKeyMetadata.java b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptionKeyMetadata.java
new file mode 100644
index 0000000..e24749a
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/encryption/BaseEncryptionKeyMetadata.java
@@ -0,0 +1,43 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.util.ByteBuffers;
+
+import java.nio.ByteBuffer;
+
+class BaseEncryptionKeyMetadata implements EncryptionKeyMetadata {
+ private final ByteBuffer keyMetadata;
+
+ BaseEncryptionKeyMetadata(ByteBuffer keyMetadata) {
+ this.keyMetadata = keyMetadata;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return keyMetadata;
+ }
+
+ @Override
+ public EncryptionKeyMetadata copy() {
+ return new BaseEncryptionKeyMetadata(
+ ByteBuffers.copy(keyMetadata));
+ }
+}
diff --git a/core/src/main/java/com/netflix/iceberg/encryption/EncryptedFiles.java b/core/src/main/java/com/netflix/iceberg/encryption/EncryptedFiles.java
new file mode 100644
index 0000000..027f314
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/encryption/EncryptedFiles.java
@@ -0,0 +1,58 @@
+/*
+ * 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 com.netflix.iceberg.encryption;
+
+import com.netflix.iceberg.io.InputFile;
+import com.netflix.iceberg.io.OutputFile;
+
+import java.nio.ByteBuffer;
+
+public class EncryptedFiles {
+
+ public static EncryptedInputFile encryptedInput(
+ InputFile encryptedInputFile, EncryptionKeyMetadata keyMetadata) {
+ return new BaseEncryptedInputFile(encryptedInputFile, keyMetadata);
+ }
+
+ public static EncryptedInputFile encryptedInput(
+ InputFile encryptedInputFile, ByteBuffer keyMetadata) {
+ return encryptedInput(encryptedInputFile, new BaseEncryptionKeyMetadata(keyMetadata));
+ }
+
+ public static EncryptedInputFile encryptedInput(InputFile encryptedInputFile, byte[] keyMetadata) {
+ return encryptedInput(encryptedInputFile, ByteBuffer.wrap(keyMetadata));
+ }
+
+ public static EncryptedOutputFile encryptedOutput(
+ OutputFile encryptingOutputFile, EncryptionKeyMetadata keyMetadata) {
+ return new BaseEncryptedOutputFile(encryptingOutputFile, keyMetadata);
+ }
+
+ public static EncryptedOutputFile encryptedOutput(
+ OutputFile encryptingOutputFile, ByteBuffer keyMetadata) {
+ return encryptedOutput(encryptingOutputFile, new BaseEncryptionKeyMetadata(keyMetadata));
+ }
+
+ public static EncryptedOutputFile encryptedOutput(OutputFile encryptedOutputFile, byte[] keyMetadata) {
+ return encryptedOutput(encryptedOutputFile, ByteBuffer.wrap(keyMetadata));
+ }
+
+ private EncryptedFiles() {}
+}
diff --git a/core/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadatas.java b/core/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadatas.java
new file mode 100644
index 0000000..12395ab
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/encryption/EncryptionKeyMetadatas.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 com.netflix.iceberg.encryption;
+
+import java.nio.ByteBuffer;
+
+public class EncryptionKeyMetadatas {
+
+ public static EncryptionKeyMetadata of(ByteBuffer keyMetadata) {
+ return new BaseEncryptionKeyMetadata(keyMetadata);
+ }
+
+ public static EncryptionKeyMetadata of(byte[] keyMetadata) {
+ return of(ByteBuffer.wrap(keyMetadata));
+ }
+
+ private EncryptionKeyMetadatas() {}
+}
diff --git a/core/src/main/java/com/netflix/iceberg/util/ByteBuffers.java b/core/src/main/java/com/netflix/iceberg/util/ByteBuffers.java
new file mode 100644
index 0000000..9fe1aaa
--- /dev/null
+++ b/core/src/main/java/com/netflix/iceberg/util/ByteBuffers.java
@@ -0,0 +1,56 @@
+/*
+ * 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 com.netflix.iceberg.util;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+public class ByteBuffers {
+
+ public static byte[] toByteArray(ByteBuffer buffer) {
+ if (buffer.hasArray()) {
+ byte[] array = buffer.array();
+ if (buffer.arrayOffset() == 0 && buffer.position() == 0
+ && array.length == buffer.remaining()) {
+ return array;
+ } else {
+ int start = buffer.arrayOffset() + buffer.position();
+ int end = start + buffer.remaining();
+ return Arrays.copyOfRange(array, start, end);
+ }
+ } else {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ return bytes;
+ }
+ }
+
+ public static ByteBuffer copy(ByteBuffer buffer) {
+ int size = buffer.remaining();
+ byte[] copyArray = new byte[size];
+ ByteBuffer readerBuffer = buffer.asReadOnlyBuffer();
+ for (int i = 0; i < size && readerBuffer.hasRemaining(); i++) {
+ copyArray[i] = readerBuffer.get();
+ }
+ return ByteBuffer.wrap(copyArray);
+ }
+
+ private ByteBuffers() {}
+}
diff --git a/spark/src/main/java/com/netflix/iceberg/spark/source/Writer.java b/spark/src/main/java/com/netflix/iceberg/spark/source/Writer.java
index 5c86b24..a1a2968 100644
--- a/spark/src/main/java/com/netflix/iceberg/spark/source/Writer.java
+++ b/spark/src/main/java/com/netflix/iceberg/spark/source/Writer.java
@@ -33,7 +33,6 @@ import com.netflix.iceberg.Metrics;
import com.netflix.iceberg.PartitionSpec;
import com.netflix.iceberg.Schema;
import com.netflix.iceberg.Table;
-import com.netflix.iceberg.TableProperties;
import com.netflix.iceberg.avro.Avro;
import com.netflix.iceberg.exceptions.RuntimeIOException;
import com.netflix.iceberg.io.FileAppender;
@@ -382,12 +381,4 @@ class Writer implements DataSourceWriter {
}
}
}
-
- private static String stripTrailingSlash(String path) {
- String result = path;
- while (result.endsWith("/")) {
- result = result.substring(0, path.length() - 1);
- }
- return result;
- }
}