This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 256f542d3e Core, Flink: Refactor InputFilesDecryptor and use
EncryptingFileIO.bulkDecrypt() (#18198)
256f542d3e is described below
commit 256f542d3e9f50f97fabf4ffea498f151fb44820
Author: gaborkaszab <[email protected]>
AuthorDate: Mon Oct 12 18:15:27 2026 +0200
Core, Flink: Refactor InputFilesDecryptor and use
EncryptingFileIO.bulkDecrypt() (#18198)
---
.../iceberg/encryption/InputFilesDecryptor.java | 81 +++++++---
.../encryption/TestInputFilesDecryptor.java | 180 +++++++++++++++++++++
.../apache/iceberg/flink/source/DataIterator.java | 4 +-
.../flink/source/RowDataFileScanTaskReader.java | 4 +-
.../operator/TestDataFileRewriteRunner.java | 2 +-
.../apache/iceberg/flink/source/DataIterator.java | 4 +-
.../flink/source/RowDataFileScanTaskReader.java | 4 +-
.../operator/TestDataFileRewriteRunner.java | 2 +-
.../apache/iceberg/flink/source/DataIterator.java | 4 +-
.../flink/source/RowDataFileScanTaskReader.java | 4 +-
.../operator/TestDataFileRewriteRunner.java | 2 +-
11 files changed, 258 insertions(+), 33 deletions(-)
diff --git
a/core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java
b/core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java
index 678d405fd7..6245fa1191 100644
--- a/core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java
+++ b/core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java
@@ -18,51 +18,84 @@
*/
package org.apache.iceberg.encryption;
-import java.nio.ByteBuffer;
-import java.util.Collections;
+import java.util.Collection;
import java.util.Map;
-import java.util.stream.Stream;
import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+/** Resolves the files referenced by scan tasks into readable, decrypted
{@link InputFile}s. */
public class InputFilesDecryptor {
- private final Map<String, InputFile> decryptedInputFiles;
+ private final Iterable<? extends ContentFile<?>> referencedFiles;
+ private final EncryptingFileIO encryptingIO;
+ private volatile Map<String, InputFile> lazyInputFiles = null;
+ /**
+ * @deprecated since 1.12.0, will be removed in 1.13.0; use {@link
#fromTasks(Iterable,
+ * EncryptingFileIO)} instead.
+ */
+ @Deprecated
public InputFilesDecryptor(
CombinedScanTask combinedTask, FileIO io, EncryptionManager encryption) {
- Map<String, ByteBuffer> keyMetadata = Maps.newHashMap();
- combinedTask.files().stream()
- .flatMap(
- fileScanTask ->
- Stream.concat(Stream.of(fileScanTask.file()),
fileScanTask.deletes().stream()))
- .forEach(file -> keyMetadata.put(file.location(), file.keyMetadata()));
- Stream<EncryptedInputFile> encrypted =
- keyMetadata.entrySet().stream()
- .map(
- entry ->
- EncryptedFiles.encryptedInput(
- io.newInputFile(entry.getKey()), entry.getValue()));
+ this(
+ () -> referencedFiles(combinedTask.files()).iterator(),
+ EncryptingFileIO.combine(io, encryption));
+ }
+
+ public static InputFilesDecryptor fromTasks(
+ Iterable<FileScanTask> tasks, EncryptingFileIO encryptingIO) {
+ return new InputFilesDecryptor(() -> referencedFiles(tasks).iterator(),
encryptingIO);
+ }
+
+ private InputFilesDecryptor(
+ Iterable<? extends ContentFile<?>> files, EncryptingFileIO encryptingIO)
{
+ this.referencedFiles = files;
+ this.encryptingIO = encryptingIO;
+ }
+
+ private Map<String, InputFile> inputFiles() {
+ if (lazyInputFiles == null) {
+ synchronized (this) {
+ if (lazyInputFiles == null) {
+ this.lazyInputFiles = encryptingIO.bulkDecrypt(referencedFiles);
+ }
+ }
+ }
+
+ return lazyInputFiles;
+ }
- // decrypt with the batch call to avoid multiple RPCs to a key server, if
possible
- @SuppressWarnings("StreamToIterable")
- Iterable<InputFile> decryptedFiles =
encryption.decrypt(encrypted::iterator);
+ private static Collection<ContentFile<?>>
referencedFiles(Iterable<FileScanTask> tasks) {
+ Map<String, ContentFile<?>> files = Maps.newHashMap();
+ for (FileScanTask task : tasks) {
+ files.put(task.file().location(), task.file());
+ for (DeleteFile delete : task.deletes()) {
+ files.put(delete.location(), delete);
+ }
+ }
- Map<String, InputFile> files =
Maps.newHashMapWithExpectedSize(keyMetadata.size());
- decryptedFiles.forEach(decrypted ->
files.putIfAbsent(decrypted.location(), decrypted));
- this.decryptedInputFiles = Collections.unmodifiableMap(files);
+ return files.values();
}
+ /**
+ * @deprecated since 1.12.0, will be removed in 1.13.0; use {@link
#getInputFile(String)} instead.
+ */
+ @Deprecated
public InputFile getInputFile(FileScanTask task) {
Preconditions.checkArgument(!task.isDataTask(), "Invalid task type");
- return decryptedInputFiles.get(task.file().location());
+ return getInputFile(task.file().location());
}
public InputFile getInputFile(String location) {
- return decryptedInputFiles.get(location);
+ InputFile inputFile = inputFiles().get(location);
+ Preconditions.checkArgument(
+ inputFile != null, "Cannot find input file for location: %s",
location);
+ return inputFile;
}
}
diff --git
a/core/src/test/java/org/apache/iceberg/encryption/TestInputFilesDecryptor.java
b/core/src/test/java/org/apache/iceberg/encryption/TestInputFilesDecryptor.java
new file mode 100644
index 0000000000..a0b82ba61e
--- /dev/null
+++
b/core/src/test/java/org/apache/iceberg/encryption/TestInputFilesDecryptor.java
@@ -0,0 +1,180 @@
+/*
+ * 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 java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+
+import java.util.List;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.MockFileScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+class TestInputFilesDecryptor {
+
+ private static final PartitionSpec SPEC = PartitionSpec.unpartitioned();
+
+ private static final DataFile DATA_FILE =
+ DataFiles.builder(SPEC)
+ .withPath("/path/to/data.parquet")
+ .withFileSizeInBytes(123L)
+ .withRecordCount(10L)
+ .withFormat(FileFormat.PARQUET)
+ .build();
+
+ private static final DataFile OTHER_DATA_FILE =
+ DataFiles.builder(SPEC)
+ .withPath("/path/to/other-data.parquet")
+ .withFileSizeInBytes(321L)
+ .withRecordCount(10L)
+ .withFormat(FileFormat.PARQUET)
+ .build();
+
+ private static final DeleteFile DELETE_FILE =
+ FileMetadata.deleteFileBuilder(SPEC)
+ .ofPositionDeletes()
+ .withPath("/path/to/deletes.parquet")
+ .withFileSizeInBytes(45L)
+ .withRecordCount(2L)
+ .withFormat(FileFormat.PARQUET)
+ .build();
+
+ private EncryptingFileIO encryptingIO;
+ private InputFile dataInputFile;
+ private InputFile deleteInputFile;
+
+ @BeforeEach
+ void before() {
+ this.dataInputFile = Mockito.mock(InputFile.class);
+ this.deleteInputFile = Mockito.mock(InputFile.class);
+ this.encryptingIO = Mockito.mock(EncryptingFileIO.class);
+ Mockito.when(encryptingIO.bulkDecrypt(any()))
+ .thenReturn(
+ ImmutableMap.of(
+ DATA_FILE.location(), dataInputFile,
+ DELETE_FILE.location(), deleteInputFile));
+ }
+
+ @Test
+ void doesNotDecryptOnCreation() {
+ InputFilesDecryptor.fromTasks(
+ ImmutableList.of(new MockFileScanTask(DATA_FILE, new DeleteFile[]
{DELETE_FILE})),
+ encryptingIO);
+
+ Mockito.verifyNoInteractions(encryptingIO);
+ }
+
+ @Test
+ void inputFilesResolvedByBulkDecrypt() {
+ InputFilesDecryptor decryptor =
+ InputFilesDecryptor.fromTasks(
+ ImmutableList.of(new MockFileScanTask(DATA_FILE, new DeleteFile[]
{DELETE_FILE})),
+ encryptingIO);
+
+
assertThat(decryptor.getInputFile(DATA_FILE.location())).isSameAs(dataInputFile);
+
assertThat(decryptor.getInputFile(DELETE_FILE.location())).isSameAs(deleteInputFile);
+
+ Mockito.verify(encryptingIO, Mockito.times(1)).bulkDecrypt(any());
+ }
+
+ @Test
+ void failsForFilesThatWereNotResolved() {
+ InputFilesDecryptor decryptor =
+ InputFilesDecryptor.fromTasks(
+ ImmutableList.of(new MockFileScanTask(DATA_FILE)), encryptingIO);
+
+ assertThatThrownBy(() -> decryptor.getInputFile("other/location"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot find input file for location: other/location");
+ }
+
+ @Test
+ void deduplicatesFiles() {
+ InputFilesDecryptor decryptor =
+ InputFilesDecryptor.fromTasks(
+ ImmutableList.of(
+ new MockFileScanTask(DATA_FILE, new DeleteFile[]
{DELETE_FILE}),
+ new MockFileScanTask(OTHER_DATA_FILE, new DeleteFile[]
{DELETE_FILE})),
+ encryptingIO);
+
+ decryptor.getInputFile(DATA_FILE.location());
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Iterable<ContentFile<?>>> captor =
ArgumentCaptor.forClass(Iterable.class);
+ Mockito.verify(encryptingIO).bulkDecrypt(captor.capture());
+ assertThat(captor.getValue())
+ .containsExactlyInAnyOrder(DATA_FILE, OTHER_DATA_FILE, DELETE_FILE);
+ }
+
+ @Test
+ void resolvesInputFilesOnceWhenAccessedConcurrently() throws Exception {
+ Mockito.when(encryptingIO.bulkDecrypt(any()))
+ .thenAnswer(
+ invocation -> {
+ Thread.sleep(50);
+ return ImmutableMap.of(DATA_FILE.location(), dataInputFile);
+ });
+
+ InputFilesDecryptor decryptor =
+ InputFilesDecryptor.fromTasks(
+ ImmutableList.of(new MockFileScanTask(DATA_FILE)), encryptingIO);
+
+ int numReaders = 8;
+ CyclicBarrier barrier = new CyclicBarrier(numReaders);
+ ExecutorService pool = Executors.newFixedThreadPool(numReaders);
+ try {
+ List<Future<InputFile>> futures = Lists.newArrayList();
+ for (int i = 0; i < numReaders; i++) {
+ futures.add(
+ pool.submit(
+ () -> {
+ barrier.await(30, SECONDS);
+ return decryptor.getInputFile(DATA_FILE.location());
+ }));
+ }
+
+ for (Future<InputFile> future : futures) {
+ assertThat(future.get()).isSameAs(dataInputFile);
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+
+ Mockito.verify(encryptingIO, Mockito.times(1)).bulkDecrypt(any());
+ }
+}
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
index 2cb7667a48..74cd535618 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
@@ -25,6 +25,7 @@ import java.util.Locale;
import org.apache.flink.annotation.Internal;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.encryption.EncryptingFileIO;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.encryption.InputFilesDecryptor;
import org.apache.iceberg.io.CloseableIterator;
@@ -56,7 +57,8 @@ public class DataIterator<T> implements CloseableIterator<T> {
EncryptionManager encryption) {
this.fileScanTaskReader = fileScanTaskReader;
- this.inputFilesDecryptor = new InputFilesDecryptor(task, io, encryption);
+ this.inputFilesDecryptor =
+ InputFilesDecryptor.fromTasks(task.files(),
EncryptingFileIO.combine(io, encryption));
this.combinedTask = task;
this.tasks = task.files().iterator();
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
index ee4aaf4a3d..88d451d33a 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
@@ -106,7 +106,9 @@ public class RowDataFileScanTaskReader implements
FileScanTaskReader<RowData> {
} else {
ReadBuilder<RowData, RowType> builder =
FormatModelRegistry.readBuilder(
- task.file().format(), RowData.class,
inputFilesDecryptor.getInputFile(task));
+ task.file().format(),
+ RowData.class,
+ inputFilesDecryptor.getInputFile(task.file().location()));
if (nameMapping != null) {
builder.withNameMapping(NameMappingParser.fromJson(nameMapping));
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
index 62b29e7c01..af78df6cce 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
@@ -225,7 +225,7 @@ class TestDataFileRewriteRunner extends OperatorTestBase {
.poll()
.getValue()
.getMessage())
- .contains("File does not exist: ");
+ .contains("Failed to open input stream for file: ");
}
}
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
index 2cb7667a48..74cd535618 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
@@ -25,6 +25,7 @@ import java.util.Locale;
import org.apache.flink.annotation.Internal;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.encryption.EncryptingFileIO;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.encryption.InputFilesDecryptor;
import org.apache.iceberg.io.CloseableIterator;
@@ -56,7 +57,8 @@ public class DataIterator<T> implements CloseableIterator<T> {
EncryptionManager encryption) {
this.fileScanTaskReader = fileScanTaskReader;
- this.inputFilesDecryptor = new InputFilesDecryptor(task, io, encryption);
+ this.inputFilesDecryptor =
+ InputFilesDecryptor.fromTasks(task.files(),
EncryptingFileIO.combine(io, encryption));
this.combinedTask = task;
this.tasks = task.files().iterator();
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
index ee4aaf4a3d..88d451d33a 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
@@ -106,7 +106,9 @@ public class RowDataFileScanTaskReader implements
FileScanTaskReader<RowData> {
} else {
ReadBuilder<RowData, RowType> builder =
FormatModelRegistry.readBuilder(
- task.file().format(), RowData.class,
inputFilesDecryptor.getInputFile(task));
+ task.file().format(),
+ RowData.class,
+ inputFilesDecryptor.getInputFile(task.file().location()));
if (nameMapping != null) {
builder.withNameMapping(NameMappingParser.fromJson(nameMapping));
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
index 62b29e7c01..af78df6cce 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
@@ -225,7 +225,7 @@ class TestDataFileRewriteRunner extends OperatorTestBase {
.poll()
.getValue()
.getMessage())
- .contains("File does not exist: ");
+ .contains("Failed to open input stream for file: ");
}
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
index 2cb7667a48..74cd535618 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java
@@ -25,6 +25,7 @@ import java.util.Locale;
import org.apache.flink.annotation.Internal;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.encryption.EncryptingFileIO;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.encryption.InputFilesDecryptor;
import org.apache.iceberg.io.CloseableIterator;
@@ -56,7 +57,8 @@ public class DataIterator<T> implements CloseableIterator<T> {
EncryptionManager encryption) {
this.fileScanTaskReader = fileScanTaskReader;
- this.inputFilesDecryptor = new InputFilesDecryptor(task, io, encryption);
+ this.inputFilesDecryptor =
+ InputFilesDecryptor.fromTasks(task.files(),
EncryptingFileIO.combine(io, encryption));
this.combinedTask = task;
this.tasks = task.files().iterator();
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
index ee4aaf4a3d..88d451d33a 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
@@ -106,7 +106,9 @@ public class RowDataFileScanTaskReader implements
FileScanTaskReader<RowData> {
} else {
ReadBuilder<RowData, RowType> builder =
FormatModelRegistry.readBuilder(
- task.file().format(), RowData.class,
inputFilesDecryptor.getInputFile(task));
+ task.file().format(),
+ RowData.class,
+ inputFilesDecryptor.getInputFile(task.file().location()));
if (nameMapping != null) {
builder.withNameMapping(NameMappingParser.fromJson(nameMapping));
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
index 62b29e7c01..af78df6cce 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
@@ -225,7 +225,7 @@ class TestDataFileRewriteRunner extends OperatorTestBase {
.poll()
.getValue()
.getMessage())
- .contains("File does not exist: ");
+ .contains("Failed to open input stream for file: ");
}
}
