This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9cb2d8a5e9a [feature](iceberg) Propagate per-file encryption keys to
scan splits (#70246)
9cb2d8a5e9a is described below
commit 9cb2d8a5e9a96b65647107a1e9bfd9237e860660
Author: Gabriel <[email protected]>
AuthorDate: Thu Nov 5 19:28:41 2026 +0800
[feature](iceberg) Propagate per-file encryption keys to scan splits
(#70246)
### Background
Iceberg decrypts encrypted manifests before native scan planning. Doris
already has a per-file encryption Thrift descriptor, but the Iceberg
planner does not populate its key and AAD fields.
### What problem does this PR solve?
Issue Number: #70245
Related PR: #70192 (metadata planning and fixture), #70213 (file-key
wire contract)
PR Type: Feature
Problem Summary:
Extract each data file's StandardKeyMetadata into every scan split so
native readers can receive the correct file key and AAD. Preserve
plaintext files and reject malformed metadata without exposing binary
payloads. Reject mixed-file metadata COUNT during bounded validation so
unsupported encryption cannot trigger eager full-table task and range
allocation before dispatch fails. Also fix the encrypted metadata
fixture's first reset after KMS restart: retained HMS tables previously
caused purge to fail while reading manifests encrypted with the lost
key.
### Details
- **Change map:** 238 production lines changed, 325 test lines, 38
documentation lines and 2 Checkstyle configuration lines; no generated
files. Production changes add a bounded SDK metadata bridge, an
immutable per-file carrier and split population; tests cover SDK
interoperability, split ownership, bounded COUNT rejection and a real
fixture restart.
- **Scope:** Parquet data-file key transport. Encrypted SQL admission
remains disabled; encrypted deletes and native reader integration are
subsequent stages. Reject independently encrypted deletes in
merge-on-read and native position_deletes metadata-table planning until
delete decryption is integrated.
- **Design:** Validate metadata size and encoded field lengths before
the SDK codec can allocate. Parse one immutable snapshot, reuse the
carrier only within a file, and copy mutable Thrift buffers for each
split. Null per-file metadata remains plaintext. Errors contain static
classifications without codec causes.
- **COUNT guard:** Both metadata COUNT algorithms validate live file
metadata in O(files) with bounded retained heap, without reading data
files. A per-file encryption marker rejects immediately with
`COUNT_READ_UNAVAILABLE`, including when the table has no encryption
declaration. Returning to ordinary scan planning would retain all tasks
and splits after COUNT scheduling had disabled streaming. Deterministic
rejection bypasses the manifest cache retry; actual cache failures
retain the SDK retry. Encrypted delete files also prevent
position-delete netting from bypassing the guard.
- **Usage:** No new configuration. Existing encrypted-read admission and
BE dispatch guards remain in place.
- **Default behavior:** Encrypted SQL remains rejected. Plaintext ranges
retain their previous wire shape.
- **Compatibility:** Reuses existing optional Thrift fields and the
Iceberg SDK's StandardKeyMetadata v1 codec; no protocol or plugin SPI
change.
- **Behavior change:** Native scan splits now carry per-file encryption
parameters; the test fixture can reset retained tables after its KMS
loses keys. Encrypted SQL admission is unchanged.
- **Fixture fix:** Drop only the two dedicated fixture HMS entries
without purge, then delete and verify their fixed object prefixes.
Cleanup therefore needs no old KMS key and leaves no orphan objects in
those prefixes. The restart regression retains HMS/storage, restarts
KMS/fixture, and requires the first reset to succeed without retrying
it.
### Testing
- Final local verification: **295 FE tests, zero
failures/errors/skips**, covering `IcebergScanPlanProviderTest`,
`IcebergScanRangeTest`, `IcebergFileEncryptionTest`,
`IcebergEncryptionReadTest` and `IcebergEncryptedMetadataIOTest`. Maven
reactor validation and FE Checkstyle passed.
- Red/green proof: the initial four-case COUNT matrix failed before the
fix. The final eight-case matrix passes across SDK/cache,
present/missing manifest row aggregates and manifest concurrency 1/8. A
batch threshold of 1 proves ordinary scanning would stream while COUNT
stays synchronous; assertions verify early rejection, no second SDK
planning pass, SDK iterable closure, no cache-failure accounting, and
stopping before the tail manifest in serial mode.
- Existing plaintext COUNT collapse, delete handling and both real
cache-failure SDK retries pass. Final verification uses
`-XX:ActiveProcessorCount=2 -XX:TieredStopAtLevel=1`, matching the FE
unit-test JVM setting. A diagnostic run without the processor cap hit
three existing cancellation-test fixture-count assertions caused by
automatic manifest merging; those test implementations were not changed.
- Previously verified, unchanged fixture lifecycle: the first reset
after a disposable KMS restart failed before the fixture fix and passes
afterward. SDK and Spark read historical snapshots `[1,2]` and
`[1,2,3,4]`; missing KMS, invalid credentials, denied access and wrong
keys remain rejected.
- Diff whitespace checks passed. Encrypted Doris SQL execution remains
unavailable at this stage; this update validates the planner guard
rather than enabling encrypted execution.
### Release note
None. This prepares file-key transport while encrypted query execution
remains disabled, and makes the external regression fixture restartable.
### Check List (For Author)
- Behavior changed: Yes (see Details: per-file transport and fixture
reset)
- Does this need documentation: No user-facing feature enablement;
internal contract and fixture instructions updated.
- Pre-submission status: Ready
### AI Disclosure
- Tool: Codex (GPT-6)
- Generated: Code, tests and this description.
- Human review: Pending maintainer review; agent self-review completed.
---
.../thirdparties/docker-compose/iceberg/README.md | 14 ++
.../encryption/IcebergEncryptionFixture.java | 14 +-
.../iceberg/encryption/test_restart.sh | 50 +++++++
fe/check/checkstyle/suppressions.xml | 2 +
fe/fe-connector/docs/iceberg-encryption-read.md | 24 +++
.../connector/iceberg/IcebergFileEncryption.java | 53 +++++++
.../connector/iceberg/IcebergScanPlanProvider.java | 72 +++++++--
.../doris/connector/iceberg/IcebergScanRange.java | 11 ++
.../iceberg/encryption/DorisKeyMetadata.java | 88 +++++++++++
.../iceberg/IcebergFileEncryptionTest.java | 110 ++++++++++++++
.../iceberg/IcebergScanPlanProviderTest.java | 165 ++++++++++++++++++++-
11 files changed, 582 insertions(+), 21 deletions(-)
diff --git a/docker/thirdparties/docker-compose/iceberg/README.md
b/docker/thirdparties/docker-compose/iceberg/README.md
index eca1822c4be..ab60fe4a704 100644
--- a/docker/thirdparties/docker-compose/iceberg/README.md
+++ b/docker/thirdparties/docker-compose/iceberg/README.md
@@ -195,3 +195,17 @@ services or the fixture credential provider in a
production deployment. Missing
services, decryption failures or incorrect reference results fail the enabled
suite.
See the connector's [metadata validation
contract](../../../../fe/fe-connector/docs/iceberg-encryption-read.md#hmskms-metadata-validation)
for the KMS failure coverage and encrypted-query boundary.
+
+The fixture reset removes only the two fixture HMS entries and their dedicated
+object prefixes, without decrypting old manifests. This also handles
`stop`/`up`:
+HMS retains its entries while the disposable KMS starts with new keys. Run the
+lifecycle regression against an isolated stack (it restarts the fixture and
KMS):
+
+```sh
+bash docker/thirdparties/docker-compose/iceberg/encryption/test_restart.sh \
+ docker/thirdparties/docker-compose/iceberg/iceberg.yaml \
+ isolated-project http://localhost:18385
+```
+
+The test requires the first reset after restart to succeed and then checks the
+SDK/Spark snapshot results and KMS failure cases again. It never retries reset.
diff --git
a/docker/thirdparties/docker-compose/iceberg/encryption/IcebergEncryptionFixture.java
b/docker/thirdparties/docker-compose/iceberg/encryption/IcebergEncryptionFixture.java
index 07785dce21f..b1ae1c491e2 100644
---
a/docker/thirdparties/docker-compose/iceberg/encryption/IcebergEncryptionFixture.java
+++
b/docker/thirdparties/docker-compose/iceberg/encryption/IcebergEncryptionFixture.java
@@ -50,6 +50,7 @@ import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.aws.s3.S3FileIO;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.GenericRecord;
@@ -165,20 +166,27 @@ public final class IcebergEncryptionFixture {
}
private synchronized void reset() throws Exception {
- try (HiveCatalog catalog = catalog(properties())) {
+ try (HiveCatalog catalog = catalog(properties()); S3FileIO cleanup =
new S3FileIO()) {
+ cleanup.initialize(properties());
Namespace namespace = Namespace.of(DB);
if (!catalog.namespaceExists(namespace)) {
catalog.createNamespace(namespace, Map.of("location",
"file:/tmp/" + DB));
}
for (String name : List.of("encrypted_v3", "plain_v3")) {
TableIdentifier id = TableIdentifier.of(DB, name);
- catalog.dropTable(id, true);
+ // HMS survives stop/up while Moto loses its keys. Purging via
old encrypted
+ // manifests would fail after dropping the HMS entry, leaking
the old objects.
+ // Only these dedicated fixture locations may be cleaned
without reading manifests.
+ catalog.dropTable(id, false);
+ String location = WAREHOUSE + "/" + name;
+ cleanup.deletePrefix(location + "/");
+ check(!cleanup.listPrefix(location +
"/").iterator().hasNext(), "Fixture cleanup is incomplete");
Map<String, String> tableProperties = new HashMap<>();
tableProperties.put(TableProperties.FORMAT_VERSION, "3");
if (name.equals("encrypted_v3")) {
tableProperties.put(TableProperties.ENCRYPTION_TABLE_KEY,
keyId);
}
- Table table = catalog.buildTable(id,
SCHEMA).withLocation(WAREHOUSE + "/" + name)
+ Table table = catalog.buildTable(id,
SCHEMA).withLocation(location)
.withProperties(tableProperties).create();
append(table, 1, 2);
append(table, 3, 4);
diff --git
a/docker/thirdparties/docker-compose/iceberg/encryption/test_restart.sh
b/docker/thirdparties/docker-compose/iceberg/encryption/test_restart.sh
new file mode 100644
index 00000000000..cbe58a6d090
--- /dev/null
+++ b/docker/thirdparties/docker-compose/iceberg/encryption/test_restart.sh
@@ -0,0 +1,50 @@
+#!/bin/bash
+# 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.
+
+set -euo pipefail
+
+if [[ $# != 3 ]]; then
+ echo "Usage: $0 COMPOSE_FILE PROJECT_NAME FIXTURE_URL" >&2
+ exit 2
+fi
+compose_file=$1
+project_name=$2
+fixture_url=${3%/}
+compose=(docker compose -f "${compose_file}" -p "${project_name}")
+
+wait_for_fixture() {
+ for attempt in {1..60}; do
+ if curl --noproxy '*' --fail --silent --max-time 2
"${fixture_url}/health" >/dev/null; then
+ return
+ fi
+ sleep 1
+ done
+ echo "Encryption fixture did not become ready" >&2
+ return 1
+}
+
+"${compose[@]}" up -d encryption-fixture
+wait_for_fixture
+curl --noproxy '*' --fail --show-error --silent --max-time 120 -X POST
"${fixture_url}/reset"
+# Retain HMS and object storage while Moto forgets its keys, as in a normal
stop/up cycle.
+"${compose[@]}" stop encryption-fixture encryption-kms
+"${compose[@]}" up -d encryption-fixture
+wait_for_fixture
+# Never retry reset: purge used to drop the HMS entry before failing, hiding
the bug on retry.
+curl --noproxy '*' --fail --show-error --silent --max-time 120 -X POST
"${fixture_url}/reset"
+curl --noproxy '*' --fail --show-error --silent --max-time 120
"${fixture_url}/verify"
diff --git a/fe/check/checkstyle/suppressions.xml
b/fe/check/checkstyle/suppressions.xml
index 96d40705bfe..595616afabc 100644
--- a/fe/check/checkstyle/suppressions.xml
+++ b/fe/check/checkstyle/suppressions.xml
@@ -23,6 +23,8 @@ under the License.
"https://checkstyle.org/dtds/suppressions_1_2.dtd";>
<suppressions>
+ <!-- The SDK metadata codec is package-private; only this small access
bridge shares its package. -->
+ <suppress
files="org[\\/]apache[\\/]iceberg[\\/]encryption[\\/]DorisKeyMetadata\.java"
checks="ImportControl"/>
<!-- Excludes test files from having Javadocs for classes and methods -->
<suppress files="[\\/]jmockit[\\/]" checks=".*" />
<suppress files="[\\/]test[\\/]" checks="MissingJavadocMethod" />
diff --git a/fe/fe-connector/docs/iceberg-encryption-read.md
b/fe/fe-connector/docs/iceberg-encryption-read.md
index 4c63c7b0ef1..cdf8ad1a11a 100644
--- a/fe/fe-connector/docs/iceberg-encryption-read.md
+++ b/fe/fe-connector/docs/iceberg-encryption-read.md
@@ -159,3 +159,27 @@ with table caching on/off and repeated reads. No
additional test-mode setting is
needed. The fixture fails rather than skipping when a required service or check
is unavailable. It prepares input for the subsequent file-key and native-reader
work; it does not admit encrypted SQL execution.
+
+## Per-file key descriptors
+
+Native data-file scan ranges carry the file data key and AAD from the SDK's
+`StandardKeyMetadata`. Manifest decryption has already exposed this metadata;
+extracting it does not make an additional KMS request for every split. A
bounded
+adapter validates encoded lengths before invoking the SDK codec, and rejects
+unknown versions, corrupt metadata, invalid AES key lengths and absent AAD
+without including binary payloads in errors.
+
+The immutable file carrier is reused only across that file's byte-slices; every
+Thrift descriptor receives independent mutable buffers. A null per-file
metadata
+field remains plaintext, including in a mixed snapshot. This is key transport
+only: the deployment admission and backend dispatch guards remain closed until
+native decryption, encrypted deletes and pruning are integrated.
+
+Both metadata COUNT algorithms validate live file metadata while retaining only
+one representative task. A per-file encryption marker causes an immediate
+`Iceberg encryption [COUNT_READ_UNAVAILABLE]` rejection, including in a table
+without a table-level encryption declaration. COUNT scheduling has already
+disabled streaming; normal-scan fallback would retain all file tasks and ranges
+before the dispatch guard rejects them. This deterministic rejection must not
+trigger the manifest cache's SDK retry. Plaintext COUNT retains its existing
+collapse and cache-failure fallback behavior.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergFileEncryption.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergFileEncryption.java
new file mode 100644
index 00000000000..6b2829695c6
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergFileEncryption.java
@@ -0,0 +1,53 @@
+// 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.doris.connector.iceberg;
+
+import org.apache.doris.thrift.TFileEncryptionInfo;
+
+import org.apache.iceberg.encryption.DorisKeyMetadata;
+import org.apache.iceberg.encryption.NativeEncryptionKeyMetadata;
+import org.apache.iceberg.util.ByteBuffers;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+
+/** Immutable, scan-local file key shared only by the byte-slices of that
file. */
+final class IcebergFileEncryption implements Serializable {
+ private static final long serialVersionUID = 1L;
+ private final byte[] key;
+ private final byte[] aad;
+
+ private IcebergFileEncryption(NativeEncryptionKeyMetadata metadata) {
+ key = ByteBuffers.toByteArray(metadata.encryptionKey());
+ aad = ByteBuffers.toByteArray(metadata.aadPrefix());
+ }
+
+ static IcebergFileEncryption fromMetadata(ByteBuffer metadata) {
+ // A null per-file descriptor denotes plaintext, including files in
mixed snapshots.
+ return metadata == null ? null : new
IcebergFileEncryption(DorisKeyMetadata.parse(metadata));
+ }
+
+ TFileEncryptionInfo toThrift() {
+ TFileEncryptionInfo info = new TFileEncryptionInfo();
+ info.setVersion(1);
+ // Thrift binary fields are mutable; one split's serialization must
not mutate another's key.
+ info.setKey(key.clone());
+ info.setAadPrefix(aad.clone());
+ return info;
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index 7076b1a4720..ac4649fa36e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -1297,6 +1297,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
List<NestedField> outputPartitionFields, boolean
enableMappingVarbinary, ZoneId zone,
UnaryOperator<String> uriNormalizer) {
DeleteFile deleteFile = task.file();
+ rejectEncryptedDelete(deleteFile);
String originalPath = deleteFile.path().toString();
IcebergScanRange.Builder builder = new IcebergScanRange.Builder()
.path(uriNormalizer.apply(originalPath))
@@ -1520,8 +1521,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
/**
* Build a collapsed COUNT(*) range from current manifest-list aggregates.
Summing each data manifest's
- * added and existing row counts is O(manifests), while only the first
live {@link FileScanTask} is needed as
- * the representative range. Old manifest lists that omit these aggregates
use the bounded O(files) fallback.
+ * added and existing row counts gives the count, but all live file
metadata must still be checked before
+ * selecting a representative range. Old manifest lists without aggregates
use the bounded per-file sum.
* Equality deletes and non-netted position deletes make the optimization
unsafe and tell the caller to
* perform a normal scan.
*/
@@ -1617,8 +1618,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
cacheBackedFileScanTasks(scan, session, table, filter,
statsQueryId, readHandle),
exactCount, formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes);
- } catch (CancellationException e) {
- // A stopped query must not restart file planning through the SDK
fallback.
+ } catch (CancellationException | EncryptedCountException e) {
+ // Cancellation and deterministic encryption rejection must not
restart planning through the SDK.
throw e;
} catch (Exception e) {
requireActiveManifestScheduler(session);
@@ -1636,22 +1637,30 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
Table table, Schema scanSchema, CloseableIterable<FileScanTask>
tasks, long exactCount,
int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, boolean netPositionDeletes) {
+ FileScanTask representative = null;
try (CloseableIterable<FileScanTask> closeableTasks = tasks) {
for (FileScanTask task : closeableTasks) {
+ // Manifest totals cannot prove that later files are
plaintext; inspect every live file.
+ checkCountFileEncryption(task);
// Data-manifest rows and the separately netted live delete
files are authoritative, but retain
// this defensive check for malformed metadata before exposing
the aggregate as a query result.
if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) ||
task.file().recordCount() < 0) {
return Optional.empty();
}
- return Optional.of(Collections.singletonList(buildRange(
- table, scanSchema, task.file(), task, formatVersion,
partitioned,
- orderedPartitionKeys, zone, uriNormalizer, exactCount,
-1, null)));
+ if (representative == null) {
+ representative = task;
+ }
}
} catch (IOException e) {
throw new RuntimeException("Failed to plan iceberg count-pushdown
file, error message is:"
+ e.getMessage(), e);
}
- return exactCount == 0 ? Optional.of(Collections.emptyList()) :
Optional.empty();
+ if (representative == null) {
+ return exactCount == 0 ? Optional.of(Collections.emptyList()) :
Optional.empty();
+ }
+ return Optional.of(Collections.singletonList(buildRange(
+ table, scanSchema, representative.file(), representative,
formatVersion, partitioned,
+ orderedPartitionKeys, zone, uriNormalizer, exactCount, -1,
null)));
}
private Optional<List<ConnectorScanRange>> planCountPushdownFromFileTasks(
@@ -1671,8 +1680,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
cacheBackedFileScanTasks(scan, session, table, filter,
statsQueryId, readHandle), formatVersion,
partitioned, orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes,
positionDeleteRows);
- } catch (CancellationException e) {
- // A stopped query must not restart file planning through the SDK
fallback.
+ } catch (CancellationException | EncryptedCountException e) {
+ // Cancellation and deterministic encryption rejection must not
restart planning through the SDK.
throw e;
} catch (Exception e) {
requireActiveManifestScheduler(session);
@@ -1695,7 +1704,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
long exactCount = 0;
try (CloseableIterable<FileScanTask> closeableTasks = tasks) {
for (FileScanTask task : closeableTasks) {
- // A metadata count is safe only when manifests alone prove
the exact visible row count.
+ checkCountFileEncryption(task);
if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) ||
task.file().recordCount() < 0) {
return Optional.empty();
}
@@ -1727,12 +1736,29 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
orderedPartitionKeys, zone, uriNormalizer,
visibleRows.getAsLong(), -1, null)));
}
+ private static void checkCountFileEncryption(FileScanTask task) {
+ if (task.file().keyMetadata() != null) {
+ // COUNT scheduling has already disabled streaming on the
assumption of one collapsed range.
+ // Falling back would retain all tasks/ranges before dispatch
rejects unsupported encryption.
+ throw new EncryptedCountException();
+ }
+ }
+
+ private static final class EncryptedCountException extends
DorisConnectorException {
+ private static final long serialVersionUID = 1L;
+
+ private EncryptedCountException() {
+ super("Iceberg encryption [COUNT_READ_UNAVAILABLE]");
+ }
+ }
+
private static boolean hasNonIgnorableTaskDeletes(FileScanTask task,
boolean netPositionDeletes) {
if (task.deletes() == null) {
return false;
}
for (DeleteFile delete : task.deletes()) {
- if (!netPositionDeletes || delete.content() !=
FileContent.POSITION_DELETES) {
+ if (delete.keyMetadata() != null || !netPositionDeletes
+ || delete.content() != FileContent.POSITION_DELETES) {
return true;
}
}
@@ -1766,7 +1792,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
try (ManifestReader<DeleteFile> reader =
ManifestFiles.readDeleteManifest(
manifest, table.io(), table.specs())) {
for (DeleteFile delete : reader) {
- if (delete.content() != FileContent.POSITION_DELETES ||
delete.recordCount() < 0) {
+ if (delete.keyMetadata() != null || delete.content() !=
FileContent.POSITION_DELETES
+ || delete.recordCount() < 0) {
return OptionalLong.empty();
}
try {
@@ -1830,6 +1857,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// gate (see IcebergScanRange#getScannedPartitionKey) costs one
table.spec() per file, not per slice.
private boolean countsAsScannedPartition;
private List<IcebergScanRange.DeleteFile> deleteCarriers =
Collections.emptyList();
+ private IcebergFileEncryption encryption;
private String fileFormat;
private Long firstRowId;
private Long lastUpdatedSequenceNumber;
@@ -1878,6 +1906,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
.length(task.length())
.fileSize(dataFile.fileSizeInBytes())
.fileFormat(file.fileFormat)
+ .encryption(file.encryption)
.formatVersion(formatVersion)
.partitionSpecId(file.partitionSpecId)
.partitionDataJson(file.partitionDataJson)
@@ -1959,6 +1988,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// 2-arg LocationPath.of(path, storagePropertiesMap));
original_file_path stays raw so BE can match
// position-delete entries against the raw iceberg path (legacy
setOriginalFilePath:304).
String rawDataPath = dataFile.path().toString();
+ IcebergFileEncryption encryption =
IcebergFileEncryption.fromMetadata(dataFile.keyMetadata());
+ // A PME file descriptor cannot be interpreted by the native ORC
reader.
+ if (encryption != null && !"parquet".equals(fileFormat)) {
+ throw new IllegalArgumentException("Iceberg encryption
[UNSUPPORTED_FILE_FORMAT]");
+ }
PerFileScratch file = scratch != null ? scratch : new PerFileScratch();
file.file = dataFile;
file.partitionSpecId = partitionSpecId;
@@ -1975,6 +2009,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
file.rawDataPath = rawDataPath;
file.normalizedPath = uriNormalizer.apply(rawDataPath);
file.deleteCarriers = buildDeleteFiles(task, uriNormalizer);
+ file.encryption = encryption;
return file;
}
@@ -1996,6 +2031,14 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return result;
}
+ private static void rejectEncryptedDelete(DeleteFile delete) {
+ // Plaintext data can reference an independently encrypted delete.
Until delete decryption is
+ // integrated, dropping its key would send BE a plaintext descriptor
for encrypted content.
+ if (delete.keyMetadata() != null) {
+ throw new UnsupportedOperationException("Iceberg encryption
[DELETE_READ_UNAVAILABLE]");
+ }
+ }
+
/**
* Convert one iceberg {@link DeleteFile} into a BE-facing carrier, a
faithful port of legacy
* {@code getDeleteFileFilters} + {@code IcebergDeleteFileFilter.create*}
+ {@code setIcebergParams}:
@@ -2012,6 +2055,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* Package-private for direct unit testing.
*/
IcebergScanRange.DeleteFile convertDelete(DeleteFile delete,
UnaryOperator<String> uriNormalizer) {
+ rejectEncryptedDelete(delete);
String path = uriNormalizer.apply(delete.path().toString());
FileContent content = delete.content();
if (content == FileContent.POSITION_DELETES) {
@@ -3320,7 +3364,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* each through the manifest cache on demand, and yields the surviving
(metrics + residual pruned)
* {@link BaseFileScanTask}s one file at a time so the consumer never
holds the whole table's task list. Not
* thread-safe: single-pass, driven by ONE consumer (the sync materialize
loop, the streaming pump, or the
- * COUNT take-first). {@code schemaJson}/{@code currentSpecJson} are
hoisted loop invariants (constant per
+ * COUNT file validation). {@code schemaJson}/{@code currentSpecJson} are
hoisted loop invariants (constant per
* table / per manifest spec), byte-identical to the pre-PERF-04 per-file
computation.
*/
private final class ManifestCacheFileScanTaskIterator implements
CloseableIterator<FileScanTask> {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
index 286a31fcfb5..b0f1035ca3b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
@@ -59,6 +59,7 @@ public class IcebergScanRange implements ConnectorScanRange {
private final long length;
private final long fileSize;
private final String fileFormat;
+ private final IcebergFileEncryption encryption;
private final int formatVersion;
private final Integer partitionSpecId;
private final String partitionDataJson;
@@ -136,6 +137,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
this.length = builder.length;
this.fileSize = builder.fileSize;
this.fileFormat = builder.fileFormat;
+ this.encryption = builder.encryption;
this.formatVersion = builder.formatVersion;
this.partitionSpecId = builder.partitionSpecId;
this.partitionDataJson = builder.partitionDataJson;
@@ -415,6 +417,9 @@ public class IcebergScanRange implements ConnectorScanRange
{
return;
}
fileDesc.setFormatVersion(formatVersion);
+ if (encryption != null) {
+ fileDesc.setEncryptionInfo(encryption.toThrift());
+ }
// original_file_path = the RAW (un-normalized) data-file path; BE
matches position-delete entries
// against it (legacy setOriginalFilePath:304 uses the raw
originalPath, not the normalized location).
// This stays raw even though the range path (getPath) is
scheme-normalized for BE to open.
@@ -493,6 +498,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
// Default empty (NOT "jni", which is not a real iceberg file format).
Production callers set the real
// orc/parquet from the data file's format; mirrors
PaimonScanRange.Builder.
private String fileFormat = "";
+ private IcebergFileEncryption encryption;
// Default 2 (the iceberg metadata default); production callers set
the real table format version.
private int formatVersion = 2;
private Integer partitionSpecId;
@@ -574,6 +580,11 @@ public class IcebergScanRange implements
ConnectorScanRange {
return this;
}
+ Builder encryption(IcebergFileEncryption encryption) {
+ this.encryption = encryption;
+ return this;
+ }
+
public Builder formatVersion(int formatVersion) {
this.formatVersion = formatVersion;
return this;
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/encryption/DorisKeyMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/encryption/DorisKeyMetadata.java
new file mode 100644
index 00000000000..dba06e43214
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/encryption/DorisKeyMetadata.java
@@ -0,0 +1,88 @@
+// 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 org.apache.avro.io.Decoder;
+import org.apache.avro.io.DecoderFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/** Minimal access to the SDK's package-private StandardKeyMetadata codec. */
+public final class DorisKeyMetadata {
+ private static final int MAX_METADATA_BYTES = 64 * 1024;
+
+ private DorisKeyMetadata() {
+ }
+
+ /** Parse bounded v1 metadata with the SDK codec without advancing the
caller's buffer. */
+ public static NativeEncryptionKeyMetadata parse(ByteBuffer metadata) {
+ ByteBuffer source = metadata.duplicate();
+ if (!source.hasRemaining() || source.remaining() > MAX_METADATA_BYTES)
{
+ throw invalid();
+ }
+ byte[] bytes = new byte[source.remaining()];
+ source.get(bytes);
+ if (bytes[0] != 1) {
+ throw new IllegalArgumentException("Iceberg encryption
[UNSUPPORTED_KEY_METADATA_VERSION]");
+ }
+ try {
+ ByteArrayInputStream input = new ByteArrayInputStream(bytes, 1,
bytes.length - 1);
+ Decoder decoder = DecoderFactory.get().directBinaryDecoder(input,
null);
+ long keyLength = decoder.readLong();
+ if (keyLength != 16 && keyLength != 24 && keyLength != 32) {
+ throw invalid();
+ }
+ skipBytes(decoder, input, keyLength);
+ // Avro allocates from encoded lengths. A tiny corrupt buffer
claiming a 2GB AAD
+ // must fail before the SDK codec can allocate it; skipping never
allocates that field.
+ if (decoder.readIndex() != 1) {
+ throw invalid();
+ }
+ skipBytes(decoder, input, decoder.readLong());
+ int lengthBranch = decoder.readIndex();
+ if (lengthBranch == 1) {
+ if (decoder.readLong() < 0) {
+ throw invalid();
+ }
+ } else if (lengthBranch != 0) {
+ throw invalid();
+ }
+ if (input.available() != 0) {
+ throw invalid();
+ }
+ // Parse the validated snapshot, rather than a buffer shared with
manifest-cache readers.
+ return StandardKeyMetadata.parse(ByteBuffer.wrap(bytes));
+ } catch (IOException | RuntimeException e) {
+ // Codec exceptions can contain binary values. Do not retain their
message or cause.
+ throw invalid();
+ }
+ }
+
+ private static void skipBytes(Decoder decoder, ByteArrayInputStream input,
long length) throws IOException {
+ if (length < 0 || length > input.available()) {
+ throw invalid();
+ }
+ decoder.skipFixed((int) length);
+ }
+
+ private static IllegalArgumentException invalid() {
+ return new IllegalArgumentException("Iceberg encryption
[INVALID_KEY_METADATA]");
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergFileEncryptionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergFileEncryptionTest.java
new file mode 100644
index 00000000000..44c30b0375c
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergFileEncryptionTest.java
@@ -0,0 +1,110 @@
+// 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.doris.connector.iceberg;
+
+import org.apache.doris.thrift.TFileEncryptionInfo;
+
+import org.apache.avro.io.Encoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.iceberg.encryption.KeyManagementClient;
+import org.apache.iceberg.encryption.NativeEncryptionKeyMetadata;
+import org.apache.iceberg.encryption.StandardEncryptionManager;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.util.ByteBuffers;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+class IcebergFileEncryptionTest {
+ @ParameterizedTest
+ @ValueSource(ints = {16, 24, 32})
+ void matchesSdkCodecAndOwnsSplitBuffers(int keyLength) {
+ StandardEncryptionManager manager = new StandardEncryptionManager(
+ List.of(), "synthetic-key", keyLength, new UnusedKms());
+ NativeEncryptionKeyMetadata metadata = manager.encrypt(
+ new
InMemoryFileIO().newOutputFile("memory://fixture/data.parquet")).keyMetadata();
+ for (NativeEncryptionKeyMetadata source : List.of(metadata,
metadata.copyWithLength(4096))) {
+ ByteBuffer encoded = source.buffer();
+ ByteBuffer direct = ByteBuffer.allocateDirect(encoded.remaining()
+ 8);
+ direct.position(3);
+ direct.put(encoded.duplicate()).flip().position(3);
+ IcebergFileEncryption encryption =
IcebergFileEncryption.fromMetadata(direct.asReadOnlyBuffer());
+ TFileEncryptionInfo first = encryption.toThrift();
+ Assertions.assertEquals(1, first.getVersion());
+
Assertions.assertArrayEquals(ByteBuffers.toByteArray(source.encryptionKey()),
first.getKey());
+
Assertions.assertArrayEquals(ByteBuffers.toByteArray(source.aadPrefix()),
first.getAadPrefix());
+ Arrays.fill(first.getKey(), (byte) 0);
+ Arrays.fill(first.getAadPrefix(), (byte) 0);
+
Assertions.assertArrayEquals(ByteBuffers.toByteArray(source.encryptionKey()),
encryption.toThrift().getKey());
+
Assertions.assertArrayEquals(ByteBuffers.toByteArray(source.aadPrefix()),
encryption.toThrift().getAadPrefix());
+ Assertions.assertEquals(3, direct.position());
+ }
+ }
+
+ @Test
+ void absentMetadataIsPlaintextButEmptyUnknownAndCorruptMetadataFail()
throws IOException {
+ Assertions.assertNull(IcebergFileEncryption.fromMetadata(null));
+ for (byte[] bytes : List.of(new byte[0], new byte[] {1}, new byte[]
{1, 32},
+ encoded(15, 1, 1), encoded(16, 0, 0), encoded(16, 1,
Integer.MAX_VALUE), new byte[65537])) {
+ IllegalArgumentException error =
Assertions.assertThrows(IllegalArgumentException.class,
+ () ->
IcebergFileEncryption.fromMetadata(ByteBuffer.wrap(bytes)));
+ Assertions.assertEquals("Iceberg encryption
[INVALID_KEY_METADATA]", error.getMessage());
+ Assertions.assertNull(error.getCause());
+ }
+ IllegalArgumentException error =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> IcebergFileEncryption.fromMetadata(ByteBuffer.wrap(new
byte[] {99})));
+ Assertions.assertEquals("Iceberg encryption
[UNSUPPORTED_KEY_METADATA_VERSION]", error.getMessage());
+ }
+
+ private static byte[] encoded(int keyLength, int aadBranch, long
aadLength) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ bytes.write(1);
+ Encoder encoder = EncoderFactory.get().directBinaryEncoder(bytes,
null);
+ encoder.writeBytes(new byte[keyLength]);
+ encoder.writeIndex(aadBranch);
+ if (aadBranch == 1) {
+ encoder.writeLong(aadLength);
+ }
+ encoder.flush();
+ return bytes.toByteArray();
+ }
+
+ private static class UnusedKms implements KeyManagementClient {
+ @Override
+ public ByteBuffer wrapKey(ByteBuffer key, String wrappingKeyId) {
+ throw new AssertionError("Per-file metadata does not require KMS
wrapping");
+ }
+
+ @Override
+ public ByteBuffer unwrapKey(ByteBuffer key, String wrappingKeyId) {
+ throw new AssertionError("Manifest processing already unwrapped
the file metadata");
+ }
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 7f3106bc65f..1354f60a6b0 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -101,6 +101,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.FileNotFoundException;
@@ -164,6 +165,162 @@ public class IcebergScanPlanProviderTest {
Types.NestedField.required(1, "id", Types.IntegerType.get()),
Types.NestedField.required(2, "p", Types.IntegerType.get()));
+ @Test
+ public void
encryptedFileMetadataReachesEverySplitWithoutAffectingPlainFiles() {
+ Table table = createTable("file_keys", SCHEMA,
PartitionSpec.unpartitioned());
+ // StandardKeyMetadata v1: a 16-byte key, a one-byte AAD, and no file
length.
+ byte[] metadata = new byte[22];
+ metadata[0] = 1;
+ metadata[1] = 32;
+ Arrays.fill(metadata, 2, 18, (byte) 7);
+ metadata[18] = 2;
+ metadata[19] = 2;
+ metadata[20] = 9;
+ DataFile encrypted =
DataFiles.builder(table.spec()).withPath("s3://bucket/encrypted.parquet")
+ .withFileSizeInBytes(96L <<
20).withRecordCount(4).withFormat("parquet")
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(metadata)).build();
+ byte[] otherMetadata = metadata.clone();
+ Arrays.fill(otherMetadata, 2, 18, (byte) 13);
+ DataFile other =
DataFiles.builder(table.spec()).withPath("s3://bucket/other.parquet")
+ .withFileSizeInBytes(96L <<
20).withRecordCount(4).withFormat("parquet")
+
.withEncryptionKeyMetadata(ByteBuffer.wrap(otherMetadata)).build();
+ table.newAppend().appendFile(encrypted)
+ .appendFile(dataFile(table.spec(),
"s3://bucket/plain.parquet", 512, null, null))
+ .appendFile(other).commit();
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ List<ConnectorScanRange> ranges = provider.planScan(null,
+ ConnectorScanRequest.builder(new IcebergTableHandle("db1",
"file_keys"), List.of()).build());
+ int encryptedSplits = 0;
+ for (ConnectorScanRange range : ranges) {
+ TTableFormatFileDesc desc = new TTableFormatFileDesc();
+ range.populateRangeParams(desc, new TFileRangeDesc());
+ String path = range.getPath().orElseThrow();
+ if (!path.endsWith("plain.parquet")) {
+
Assertions.assertTrue(desc.getIcebergParams().isSetEncryptionInfo());
+ Assertions.assertArrayEquals(new byte[] {9},
desc.getIcebergParams().getEncryptionInfo().getAadPrefix());
+ byte[] key = new byte[16];
+ Arrays.fill(key, path.endsWith("other.parquet") ? (byte) 13 :
(byte) 7);
+ Assertions.assertArrayEquals(key,
desc.getIcebergParams().getEncryptionInfo().getKey());
+ encryptedSplits++;
+ } else {
+
Assertions.assertFalse(desc.getIcebergParams().isSetEncryptionInfo());
+ }
+ }
+ Assertions.assertTrue(encryptedSplits > 2);
+ }
+
+ @Test
+ public void encryptedDeleteFilesAreRejectedBeforeLosingTheirKeys() {
+ for (boolean equality : new boolean[] {false, true}) {
+ Table table = createTable("encrypted_deletes_" + equality, SCHEMA,
PartitionSpec.unpartitioned());
+ FileMetadata.Builder builder =
FileMetadata.deleteFileBuilder(table.spec());
+ if (equality) {
+ builder.ofEqualityDeletes(1);
+ } else {
+ builder.ofPositionDeletes();
+ }
+ DeleteFile delete =
builder.withPath("s3://bucket/delete.parquet").withFormat("parquet")
+ .withFileSizeInBytes(100).withRecordCount(1)
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(new byte[]
{1})).build();
+ if (!equality) {
+ Table withDeletes = tableWithPositionDelete(delete);
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> planPositionDeletes(withDeletes,
Collections.emptyList()));
+ IcebergScanPlanProvider countProvider = new
IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(withDeletes));
+ ConnectorSession countSession = new FakeScanSession("UTC",
+
Collections.singletonMap("ignore_iceberg_dangling_delete", "true"));
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> planCount(countProvider, countSession, true));
+ }
+ UnsupportedOperationException error =
Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> provider().convertDelete(delete,
UnaryOperator.identity()));
+ Assertions.assertEquals("Iceberg encryption
[DELETE_READ_UNAVAILABLE]", error.getMessage());
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false,false,1", "false,true,1", "true,false,1", "true,true,1",
+ "false,false,8", "false,true,8", "true,false,8", "true,true,8"})
+ public void mixedEncryptedFilesCannotCollapseMetadataCount(boolean cached,
boolean missingManifestRows,
+ int manifestConcurrency) {
+ Table table = createTable("mixed_count", SCHEMA,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap("format-version", "2"));
+ byte[] metadata = new byte[22];
+ metadata[0] = 1;
+ metadata[1] = 32;
+ Arrays.fill(metadata, 2, 18, (byte) 7);
+ metadata[18] = 2;
+ metadata[19] = 2;
+ metadata[20] = 9;
+ DataFile encrypted =
DataFiles.builder(table.spec()).withPath("s3://bucket/encrypted.parquet")
+
.withFileSizeInBytes(512).withRecordCount(4).withFormat("parquet")
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(metadata)).build();
+ // New manifests precede old ones, leaving an unread tail after the
encrypted manifest.
+ table.newFastAppend()
+ .appendFile(dataFile(table.spec(), "s3://bucket/tail.parquet",
512, null, null)).commit();
+ table.newFastAppend().appendFile(encrypted).commit();
+ table.newFastAppend()
+ .appendFile(dataFile(table.spec(),
"s3://bucket/plain.parquet", 512, null, null)).commit();
+ Table plannedTable = missingManifestRows
+ ? tableWithMissingManifestRowsAndIo(table, table.io()) : table;
+ AtomicInteger sdkPlans = new AtomicInteger();
+ AtomicBoolean tasksClosed = new AtomicBoolean();
+ AtomicBoolean encryptedConsumed = new AtomicBoolean();
+ Table observed = (Table) observeCountPlanning(plannedTable, sdkPlans,
tasksClosed, encryptedConsumed);
+ IcebergManifestCache cache = new IcebergManifestCache();
+ IcebergScanPlanProvider provider = manifestProvider(
+ cached ? manifestCacheProps() : Collections.emptyMap(),
observed, cache);
+ ConnectorSession session = new FakeScanSession("UTC",
+ ImmutableMap.of("num_files_in_batch_mode", "1",
+ "iceberg_manifest_load_concurrency",
Integer.toString(manifestConcurrency)));
+ Assertions.assertEquals(3, provider.streamingSplitEstimate(session,
+ new IcebergTableHandle("db1", "mixed_count"),
Optional.empty(), false));
+ Assertions.assertEquals(-1, provider.streamingSplitEstimate(session,
+ new IcebergTableHandle("db1", "mixed_count"),
Optional.empty(), true));
+
+ // COUNT has already disabled streaming. Reject during bounded
validation, before a normal scan
+ // can retain every task/range or treat this deterministic refusal as
a cache failure.
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planCount(provider, session, true));
+ Assertions.assertEquals("Iceberg encryption [COUNT_READ_UNAVAILABLE]",
error.getMessage());
+ Assertions.assertEquals(cached ? 0 : 1, sdkPlans.get(), "must not
restart file planning");
+ Assertions.assertEquals(0L, cache.takeStats("q")[2], "encryption
rejection is not a cache failure");
+ if (cached) {
+ Assertions.assertTrue(cache.size() > 0, "must exercise the
manifest-cache planner");
+ if (manifestConcurrency == 1) {
+ // Parallel loads may finish speculatively; serial loading
makes the stop position exact.
+ Assertions.assertEquals(2, cache.size(), "must stop before
reading the tail manifest");
+ }
+ } else {
+ Assertions.assertTrue(encryptedConsumed.get());
+ Assertions.assertTrue(tasksClosed.get(), "early rejection must
close SDK tasks");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Object observeCountPlanning(Object value, AtomicInteger
sdkPlans, AtomicBoolean tasksClosed,
+ AtomicBoolean encryptedConsumed) {
+ if (!(value instanceof Table) && !(value instanceof TableScan)) {
+ return value;
+ }
+ Class<?> type = value instanceof Table ? Table.class : TableScan.class;
+ return Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]
{type}, (proxy, method, args) -> {
+ Object result = invoke(method, value, args);
+ if (method.getName().equals("planFiles")) {
+ sdkPlans.incrementAndGet();
+ CloseableIterable<FileScanTask> tasks =
(CloseableIterable<FileScanTask>) result;
+ return
CloseableIterable.whenComplete(CloseableIterable.transform(tasks, task -> {
+ Assertions.assertFalse(encryptedConsumed.get(), "must stop
consuming after an encrypted file");
+ encryptedConsumed.set(task.file().keyMetadata() != null);
+ return task;
+ }), () -> tasksClosed.set(true));
+ }
+ return observeCountPlanning(result, sdkPlans, tasksClosed,
encryptedConsumed);
+ });
+ }
+
// --- in-memory iceberg table helpers (offline; DataFile metadata only,
no real data files) ---
private static Table createTable(String name, Schema schema, PartitionSpec
spec) {
@@ -5663,7 +5820,7 @@ public class IcebergScanPlanProviderTest {
}
@Test
- public void countPushdownManifestCacheReadsOnlyRepresentativeFile() {
+ public void countPushdownManifestCacheValidatesEveryFileBeforeCollapse() {
// Three appends -> three data manifests; manifest-list live-row
aggregates sum to 10+20+30 = 60.
Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
table.newAppend().appendFile(dataFile(table.spec(),
"s3://b/db/t1/f1.parquet", 1024, null, null)).commit();
@@ -5684,9 +5841,9 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals(1, cached.size());
Assertions.assertEquals(60L, cached.get(0).getPushDownRowCount());
Assertions.assertEquals(sdk.get(0).getPushDownRowCount(),
cached.get(0).getPushDownRowCount());
- // The manifest list already carries exact live-row aggregates. Only
the first manifest's entries are
- // needed to obtain a representative file for the collapsed range.
- Assertions.assertEquals(1, cache.size());
+ // Manifest aggregates contain no per-file encryption marker. Validate
all entries before choosing
+ // a representative, including manifests after the first plaintext
file.
+ Assertions.assertEquals(totalManifests, cache.size());
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
