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 56dfbb1c273 [feature](fe) Add encrypted Iceberg read admission checks
(#70159)
56dfbb1c273 is described below
commit 56dfbb1c273050aa3cc0f32d20a242be24af4900
Author: Gabriel <[email protected]>
AuthorDate: Wed Nov 4 15:15:15 2026 +0800
[feature](fe) Add encrypted Iceberg read admission checks (#70159)
### What problem does this PR solve?
Related PR: #70087
Problem Summary:
Enabling Parquet modular encryption in Arrow does not implement native
encrypted Iceberg reads in Doris. Scan planning currently has no shared
admission check for an encrypted table. Ordinary scans can emit ranges
without establishing encryption support, while metadata-only COUNT and
system-table routes can bypass the native data reader altogether.
For example, a v3 table whose SDK metadata contains `encryption.key-id`
can reach scan planning and COUNT pushdown without any proof that the
selected backends understand encrypted Iceberg tasks. This PR
establishes the first-stage read contract before native execution is
enabled.
- Add `enable_encryption_read` to the Iceberg plugin's `iceberg.conf`,
defaulting to `false`. This is deployment configuration requiring an FE
restart, with no catalog/session override or fe.conf fallback.
- Share admission checks across table metadata and snapshot binding,
ordinary scans, COUNT eligibility/pushdown, streaming scans, system
tables and explicitly empty snapshot pins.
- Detect encryption from the table key property, retained v3 encryption
keys, or a non-plaintext SDK manager. Removing the current write key
must not admit old encrypted snapshots as plaintext.
- Return stable classifications for disabled reads, unsupported
catalogs/table formats/custom managers, and unavailable backend
capabilities. Error text excludes key identifiers and encryption
objects.
- Even with the switch on, reject encrypted execution until backend
capability negotiation and native readers exist. A declared encrypted
table is rejected before KMS manager initialization or manifest I/O.
Plaintext scans and exact metadata counts retain their behavior.
The included developer contract defines the initial HMS / v3 / native
Parquet `AES_GCM_V1` target and the requirements for explicit BE
advertisement, placement, retries, key ownership and redaction. This PR
deliberately adds no SPI/wire fields and grants no encrypted-write
capability. Actual per-file format/algorithm validation and undeclared
externally encrypted files remain part of the subsequent planner/reader
work.
### Validation
```sh
JAVA_TOOL_OPTIONS='-XX:ActiveProcessorCount=8' mvn -f fe/pom.xml -pl
:fe-connector-iceberg -am package -Dmaven.build.cache.enabled=false
mvn -f fe/pom.xml -pl :fe-connector-iceberg checkstyle:check
```
- The complete reactor package succeeded, including connector packaging
and upstream module tests. Iceberg connector results: **1923 tests
reported, 1909 passed, 14 existing skips, 0 failures, 0 errors**.
- Added 8 contract tests and 1 configuration test. Real SDK metadata
fixtures cover empty/populated tables, direct/cached acquisition,
statement scopes, metadata binding, COUNT, streaming, system tables,
retained keys, unsupported catalogs/v2/custom managers, error redaction
and rejection before KMS initialization or manifest/data I/O.
- Before the implementation, five initial contract tests failed because
encrypted reads were admitted; the plaintext compatibility test passed.
The final added contract/configuration tests all pass.
- FE Checkstyle: **0 violations**. Verified that the plugin zip includes
the admission class and the updated configuration template.
- Local HTTP mock servers were run without proxy interception. The test
JVM uses 8 processors so the existing cancellation fixtures do not
trigger the SDK's automatic manifest-merge threshold. No existing test
was changed or excluded for this.
SQL regression coverage is included in
`test_iceberg_encryption_read_guard`:
```sh
bash run-regression-test.sh --run -d external_table_p0/iceberg \
-s test_iceberg_encryption_read_guard
```
- Built the full FE from this PR using `build.sh --fe`, and ran the
suite with an isolated FE and an existing ASAN BE against the standard
REST/MinIO fixture.
- The SQL suite directly asserts the production default,
`READ_DISABLED`, without a separate regression option for the expected
FE switch state. It also sets a catalog-level
`enable_encryption_read=true` to verify that a catalog cannot bypass the
deployment-level setting.
- Default plugin configuration: passed normal expected-output
comparison; **96 admission-error assertions**, with both table-cache
settings, both COUNT pushdown settings, populated/empty declarations,
repeated queries, SELECT, COUNT, DESC and `$files` / `$snapshots` /
`$position_deletes`. Plaintext rows and counts remain correct.
- Connector unit tests explicitly construct both enabled and disabled
plugin contexts; enabled REST contexts reject with
`UNSUPPORTED_CATALOG`, and eligible HMS contexts reject with
`BACKEND_CAPABILITY_UNAVAILABLE`. The ordinary SQL regression does not
require an enabled FE deployment.
- Mutation control: in a temporary runtime-only artifact, replaced the
admission helper with a no-op. The suite failed at the first
declared-table SELECT because the required `READ_DISABLED` error was
absent. Restored and hash-verified the unmodified production artifact
before the passing runs; no test hook is included in the source change.
- The expected `.out` file was generated by the standard regression
runner and then used for normal comparison. The suite was rerun against
its existing remote tables, exercising repeatable fixture cleanup. FE
Checkstyle was rechecked: 0 violations.
These SQL tests use real v3 tables and plaintext data files with
synthetic encryption declarations to isolate admission from decryption
failures. They validate the FE/SQL guard, not native ciphertext
decoding. Live KMS decryption, mixed-version cluster integration and
full encrypted-query execution remain subsequent work.
### Release note
Declared encrypted Iceberg tables now fail with clear read-admission
errors. The plugin read switch defaults off, and enabling it cannot
bypass unavailable backend support or enable encrypted writes.
### Check List (For Author)
- Test
- [x] Unit Test
- [x] Regression test
- [ ] Manual test
- Behavior changed:
- [x] Yes. Declared encrypted tables are rejected consistently before
scan execution or metadata-count shortcuts.
- Does this need documentation?
- [x] Yes. Developer contract and configuration comments are included in
this PR; public query documentation follows when encrypted execution is
implemented.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
fe/fe-connector/README.md | 4 +
fe/fe-connector/docs/iceberg-encryption-read.md | 118 ++++++++
.../doris/connector/iceberg/IcebergConf.java | 14 +-
.../iceberg/IcebergConnectorMetadata.java | 19 +-
.../connector/iceberg/IcebergEncryptionRead.java | 73 +++++
.../connector/iceberg/IcebergScanPlanProvider.java | 15 +-
.../src/main/resources/iceberg.conf.template | 11 +-
.../doris/connector/iceberg/FakeIcebergTable.java | 3 +-
.../iceberg/IcebergConnectorConfTest.java | 14 +-
.../iceberg/IcebergEncryptionReadTest.java | 297 +++++++++++++++++++++
.../iceberg/test_iceberg_encryption_read_guard.out | 44 +++
.../test_iceberg_encryption_read_guard.groovy | 141 ++++++++++
12 files changed, 736 insertions(+), 17 deletions(-)
diff --git a/fe/fe-connector/README.md b/fe/fe-connector/README.md
index abac08783f1..1b61b82c0ab 100644
--- a/fe/fe-connector/README.md
+++ b/fe/fe-connector/README.md
@@ -129,6 +129,10 @@ and reports observed partition, file, split, and
unfinished-producer counts.
producer count. A positive value caps producers per scan; a whole-scan
streaming
source uses one producer.
+**Iceberg encryption.** The [encrypted-read admission
contract](docs/iceberg-encryption-read.md)
+defines the default-off configuration, initial scope and backend negotiation
+requirements. Encrypted scan execution remains unavailable at this stage.
+
**Classloading.** Plugins load child-first, each carrying its own runtime
closure. Wherever engine code crosses into a plugin — or a bundled library
resolves classes by name, or a plugin spawns worker threads — the
diff --git a/fe/fe-connector/docs/iceberg-encryption-read.md
b/fe/fe-connector/docs/iceberg-encryption-read.md
new file mode 100644
index 00000000000..ae840b30e95
--- /dev/null
+++ b/fe/fe-connector/docs/iceberg-encryption-read.md
@@ -0,0 +1,118 @@
+# Iceberg encrypted-read contract
+
+This is the admission contract for the first stage of native Iceberg encrypted
+reads. Encrypted queries remain unavailable until key transport, native
readers,
+delete handling and backend capability negotiation are implemented. Building
+Arrow with Parquet modular encryption alone does not establish those
capabilities.
+
+## Configuration and admission
+
+`enable_encryption_read` in the Iceberg plugin's `iceberg.conf` defaults to
+`false`. It is deployment configuration: set it consistently on every FE and
+restart the FE after changes. It is neither a catalog property nor a session
+variable. Enabling it expresses operator intent; it does not override missing
+execution capabilities or enable encrypted writes.
+
+Plaintext tables retain their existing paths, including metadata-only counts.
+For encrypted tables the same check applies to metadata binding, snapshot
+resolution, ordinary scans, metadata tables, streaming scans and COUNT
pushdown.
+The check runs against the base table before constructing a metadata table or
+reading its manifests. An empty snapshot does not exempt an encrypted table.
+
+The initial target matrix is:
+
+| Dimension | Initial target | Current admission |
+| --- | --- | --- |
+| Catalog | HMS with the SDK standard encryption manager and an
operator-configured KMS | Other catalogs are rejected |
+| Table format | Iceberg v3 | Other versions are rejected for encrypted reads |
+| Data and delete files | Native Parquet modular encryption, `AES_GCM_V1` | No
encrypted ranges are emitted yet |
+| Manifest/list files | SDK-managed encryption | No plaintext fallback is
allowed |
+| Backend | Explicit encrypted-read protocol and reader capability on every
eligible BE | Capability is currently unverified, so admission fails |
+| Writes | Outside this read contract | The read switch grants no write
capability |
+
+Table encryption is recognized from `encryption.key-id`, retained v3
+`encryption-keys`, or a non-plaintext SDK encryption manager. Retained keys
matter
+when an administrator removes the current write key while older snapshots still
+reference encrypted files. The check must not log key identifiers, key
metadata,
+KMS credentials or key material.
+
+The writer's `write.format.default` is not evidence of the format of existing
+files. File format, algorithm, key metadata and AAD must be validated on the
+actual data/delete tasks by the subsequent planning and reader implementations.
+This stage rejects encrypted tables before that work, even with the switch on.
+It does not inspect every plaintext table's manifests just to discover
encryption.
+Externally encrypted files with no Iceberg encryption declaration are outside
+this table-level contract; data-task validation must reject them when
encountered.
+
+## Failure classification
+
+Errors contain a stable `Iceberg encryption [CLASSIFICATION]` prefix:
+
+| Classification | Meaning |
+| --- | --- |
+| `READ_DISABLED` | Encrypted reads were not enabled in the plugin
configuration |
+| `UNSUPPORTED_CATALOG` | The catalog is outside the initial HMS scope |
+| `UNSUPPORTED_TABLE_FORMAT` | The encrypted table is not v3 |
+| `UNSUPPORTED_ENCRYPTION_MANAGER` | Encryption detection observed a custom
SDK encryption manager |
+| `BACKEND_CAPABILITY_UNAVAILABLE` | The engine cannot prove encrypted-read
support |
+
+The first applicable failure is returned in this order. For a table with an
+encryption declaration, backend admission fails without initializing its KMS
+manager; manager and per-file validation belong to subsequent read
implementation.
+A rejected read must
+not fall back to plaintext, metadata counts, a JNI reader or another catalog.
+Plaintext reads in those catalogs continue to work.
+
+## Backend negotiation required before enabling execution
+
+The protocol change must introduce an explicit, versioned
encrypted-Iceberg-read
+capability, advertised by each BE. An absent field, an older BE, an unknown
+capability or a build without PME support means **unsupported**. A generic
+execution version, the FE's Iceberg version and the FE configuration switch are
+not substitutes for this advertisement.
+
+The scheduler must restrict placement to compatible BEs and fail if it cannot
+establish that all selected execution targets support the complete
encrypted-read
+contract. Reassignment and retries must repeat that validation. The native
reader
+must also reject an encrypted range it cannot handle, so a stale heartbeat
cannot
+silently downgrade a read. FE-to-BE key transport must have explicit ownership,
+bounded lifetime and redaction rules; keys must not enter explain output,
profiles
+or persistent scan caches.
+
+Until those protocol and reader changes land together, the connector
deliberately
+reports `BACKEND_CAPABILITY_UNAVAILABLE` rather than manufacturing a successful
+capability result. No SPI or wire-format capability is added by this first
stage.
+
+See the upstream [Iceberg encryption
documentation](https://iceberg.apache.org/docs/nightly/encryption/)
+and [table specification](https://iceberg.apache.org/spec/#encryption-keys) for
+the SDK encryption scheme and retained table keys.
+
+## SQL regression
+
+`test_iceberg_encryption_read_guard` uses the standard external Iceberg REST
and
+MinIO fixture. With `enableIcebergTest=true`, run:
+
+```sh
+bash run-regression-test.sh --run -d external_table_p0/iceberg \
+ -s test_iceberg_encryption_read_guard
+```
+
+The suite writes ordinary v3 tables, then uses the REST fixture to add an
+encryption declaration to a populated table and an empty table. These are
+metadata admission fixtures, not encrypted payloads: the expected failure must
+precede file decryption and requires no KMS. It checks ordinary SELECT, COUNT
+with pushdown enabled/disabled, DESC and system tables, with the table cache
+enabled/disabled and repeated reads. Plaintext row/count results are checked
+before and after rejection. Missing fixture services fail the enabled suite.
+
+The suite runs against the default FE deployment, with `enable_encryption_read`
+absent or set to `false` in every FE's Iceberg plugin configuration. It always
+expects `READ_DISABLED`, including when the catalog requests encrypted reads;
+there is no separate regression setting for the expected deployment state.
+An incorrect error classification fails the suite.
+
+Connector unit tests explicitly construct enabled and disabled plugin contexts
+to cover both switch states, unsupported catalogs and table formats, retained
+v3 keys, custom managers and unavailable backend capabilities. The SQL suite
+and these unit tests validate admission; they do not exercise native decryption
+or grant encrypted-write support.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConf.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConf.java
index f8d1f377d56..79c73f229b6 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConf.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConf.java
@@ -23,10 +23,11 @@ import org.apache.doris.connector.spi.ConnectorContext;
/**
* This connector's deployment-level settings — the "B class" of the connector
property convention, as
* opposed to the per-catalog properties in {@link IcebergCatalogProperties}.
These are read from the
- * plugin's own {@code iceberg.conf} (named after {@code
ConnectorProvider.name()}), each falling back to
- * the {@code ENV_} name below it, which is the fe.conf key it used to live
under and still works.
+ * plugin's own {@code iceberg.conf} (named after {@code
ConnectorProvider.name()}). Existing shared settings
+ * fall back to the {@code ENV_} name below them; new settings have no fe.conf
fallback.
*
- * <p>Both conf keys are shared with other connectors at the fe.conf end — one
{@code jdbc_drivers_dir}
+ * <p>The driver and metastore conf keys are shared with other connectors at
the fe.conf end — one
+ * {@code jdbc_drivers_dir}
* and one {@code hive_metastore_client_timeout_second} serve jdbc, iceberg
and paimon. A plugin conf
* cannot express that, so a deployment that moves to these files sets the
value in each plugin's own
* conf. That is the accepted cost of a per-plugin file; the fe.conf keys stay
as the shared fallback.
@@ -35,6 +36,7 @@ public final class IcebergConf {
public static final String CONF_DRIVERS_DIR = "drivers_dir";
public static final String CONF_METASTORE_CLIENT_TIMEOUT_SECOND =
"metastore_client_timeout_second";
+ public static final String CONF_ENABLE_ENCRYPTION_READ =
"enable_encryption_read";
/** The fe.conf name of {@link #CONF_DRIVERS_DIR}, forwarded through the
engine environment. */
public static final String ENV_JDBC_DRIVERS_DIR = "jdbc_drivers_dir";
@@ -49,6 +51,12 @@ public final class IcebergConf {
private IcebergConf() {
}
+ /** Operator intent only: the read contract independently requires
verified backend capabilities. */
+ public static boolean encryptionReadEnabled(ConnectorContext context) {
+ return context != null && Boolean.parseBoolean(
+ ConnectorConf.get(context, CONF_ENABLE_ENCRYPTION_READ, null,
"false"));
+ }
+
/**
* The directory a bare driver jar name resolves under, from this plugin's
own {@code iceberg.conf}
* or fe.conf's {@code jdbc_drivers_dir}. Null when neither names one —
{@code JdbcDriverSupport}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index e55a9e372d0..ac669d74d13 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -782,18 +782,22 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
* the sys-table path ({@link #loadSysTable}), which builds its metadata
view from this same scoped base.
*/
private Table resolveTableForRead(ConnectorSession session,
IcebergTableHandle handle) {
+ Table table;
if (tableCache != null) {
- return IcebergStatementScope.sharedBorrowedTable(session,
handle.getDbName(), handle.getTableName(),
+ table = IcebergStatementScope.sharedBorrowedTable(session,
handle.getDbName(), handle.getTableName(),
() ->
tableCache.borrow(TableIdentifier.of(handle.getDbName(), handle.getTableName()),
() -> catalogOps.loadTable(handle.getDbName(),
handle.getTableName())),
() -> catalogOps.loadTable(handle.getDbName(),
handle.getTableName()));
+ } else {
+ table = resourceTracker == null
+ ? IcebergStatementScope.sharedTable(session,
handle.getDbName(), handle.getTableName(),
+ () -> catalogOps.loadTable(handle.getDbName(),
handle.getTableName()))
+ : IcebergStatementScope.sharedTrackedTable(session,
handle.getDbName(), handle.getTableName(),
+ resourceTracker, () ->
catalogOps.loadTable(handle.getDbName(), handle.getTableName()),
+ value ->
IcebergConnector.cachedTableCleanup(value, catalogProps.getFlavor()));
}
- return resourceTracker == null
- ? IcebergStatementScope.sharedTable(session,
handle.getDbName(), handle.getTableName(),
- () -> catalogOps.loadTable(handle.getDbName(),
handle.getTableName()))
- : IcebergStatementScope.sharedTrackedTable(session,
handle.getDbName(), handle.getTableName(),
- resourceTracker, () ->
catalogOps.loadTable(handle.getDbName(), handle.getTableName()),
- table -> IcebergConnector.cachedTableCleanup(table,
catalogProps.getFlavor()));
+ IcebergEncryptionRead.check(table, catalogProps, context);
+ return table;
}
/**
@@ -2795,6 +2799,7 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
}
private IcebergLatestSnapshotCache.CachedSnapshot latestSnapshotPin(Table
table) {
+ IcebergEncryptionRead.check(table, catalogProps, context);
Snapshot current = table.currentSnapshot();
ConnectorMvccPartitionView.Style emptyPartitionStyle =
IcebergPartitionUtils.isValidRelatedTable(table)
? ConnectorMvccPartitionView.Style.RANGE
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergEncryptionRead.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergEncryptionRead.java
new file mode 100644
index 00000000000..45faf6b70f0
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergEncryptionRead.java
@@ -0,0 +1,73 @@
+// 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.connector.spi.ConnectorContext;
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
+import org.apache.iceberg.encryption.StandardEncryptionManager;
+
+/** Shared admission contract; encrypted ranges remain forbidden until the
engine can prove BE support. */
+final class IcebergEncryptionRead {
+ private IcebergEncryptionRead() {
+ }
+
+ static void check(Table table, IcebergCatalogProperties catalog,
ConnectorContext context) {
+ TableMetadata metadata = table instanceof HasTableOperations
+ ? ((HasTableOperations) table).operations().current() : null;
+ // Removing the current write key does not make old snapshots
plaintext. Retained table keys must
+ // fence metadata-only COUNT and system tables before either path can
bypass native reader admission.
+ boolean declared =
table.properties().containsKey(TableProperties.ENCRYPTION_TABLE_KEY)
+ || (metadata != null && !metadata.encryptionKeys().isEmpty());
+ EncryptionManager manager = declared ? null : table.encryption();
+ if (!declared && (manager == null || manager instanceof
PlaintextEncryptionManager)) {
+ return;
+ }
+ if (!IcebergConf.encryptionReadEnabled(context)) {
+ throw failure("READ_DISABLED", "Encrypted Iceberg reads are
disabled; enable_encryption_read in "
+ + "iceberg.conf defaults to false");
+ }
+ if (!IcebergCatalogProperties.TYPE_HMS.equals(catalog.getFlavor())) {
+ throw failure("UNSUPPORTED_CATALOG", "Encrypted Iceberg reads
require an HMS catalog");
+ }
+ if (metadata == null || metadata.formatVersion() != 3) {
+ throw failure("UNSUPPORTED_TABLE_FORMAT", "Encrypted Iceberg reads
require table format v3");
+ }
+ // A declared table needs no KMS initialization just to reject an
unverified execution path.
+ // Custom managers observed during detection can already be classified
without resolving any keys.
+ if (manager != null && !(manager instanceof
StandardEncryptionManager)) {
+ throw failure("UNSUPPORTED_ENCRYPTION_MANAGER", "Encrypted Iceberg
reads require the SDK standard "
+ + "encryption scheme with native Parquet AES_GCM_V1
files");
+ }
+ // A PME-enabled Arrow build does not prove key-transport,
delete-reader or mixed-version scheduling
+ // support. Until explicit BE negotiation is wired, even an enabled FE
must never emit encrypted tasks.
+ throw failure("BACKEND_CAPABILITY_UNAVAILABLE", "Encrypted Iceberg
read capability is not verified "
+ + "by the engine; native encrypted scan execution is not
available");
+ }
+
+ private static DorisConnectorException failure(String classification,
String message) {
+ // Never attach table properties, key identifiers or SDK encryption
objects to this error.
+ return new DorisConnectorException("Iceberg encryption [" +
classification + "]: " + message);
+ }
+}
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 0944c2bbfea..7076b1a4720 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
@@ -489,6 +489,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
public long streamingSplitEstimate(ConnectorSession session,
ConnectorTableHandle handle,
Optional<ConnectorExpression> filter, boolean countPushdown) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ if (iceHandle.isResolvedEmptySnapshot() && !iceHandle.isSystemTable())
{
+ resolveTable(session, iceHandle);
+ }
if (iceHandle.isResolvedEmptySnapshot() || iceHandle.isSystemTable()
|| !sessionBool(session, ENABLE_EXTERNAL_TABLE_BATCH_MODE,
true)) {
return -1;
@@ -570,6 +573,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
validateMetadataColumnReader(session, columns);
if (iceHandle.isResolvedEmptySnapshot()) {
+ resolveTable(session, iceHandle);
// The batch decision is made before the engine pins MVCC; once
pinned empty, streaming must
// preserve that boundary instead of interpreting Iceberg's
sentinel as the latest snapshot.
return emptySplitSource();
@@ -866,7 +870,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
Optional<ConnectorExpression> filter,
boolean countPushdown) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
- if (iceHandle.isResolvedEmptySnapshot() &&
!isSnapshotIndependentSystemTable(iceHandle)) {
+ if (iceHandle.isResolvedEmptySnapshot() && !iceHandle.isSystemTable())
{
+ // Empty pins still need admission: otherwise COUNT/empty scans
bypass the encryption fence.
+ resolveTable(session, iceHandle);
// Iceberg has no snapshot id that can represent "before the first
commit". Returning no ranges is
// the read-side MVCC fence; otherwise a refreshed Table would
turn -1 into "latest" and expose a
// concurrent first append to MERGE after its anti-join had
already decided the row was absent. Static
@@ -1084,6 +1090,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private List<ConnectorScanRange> doPlanSystemTableScan(IcebergTableHandle
handle,
List<ConnectorColumnHandle> columns, Optional<ConnectorExpression>
filter,
ConnectorSession session) {
+ if (handle.isResolvedEmptySnapshot() &&
!isSnapshotIndependentSystemTable(handle)) {
+ // Keep the empty-table fence inside the same auth scope as every
other system-table load.
+ resolveSysBaseTable(session, handle);
+ return Collections.emptyList();
+ }
if (IcebergLabelsSysTable.isLabels(handle)) {
return doPlanLabelsSystemTableScan(handle, columns, session);
}
@@ -3625,6 +3636,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
TableIdentifier.of(handle.getDbName(),
handle.getTableName()), directLoader),
directLoader);
rejectServerSideScanPlanning(raw, handle);
+ IcebergEncryptionRead.check(raw, catalogProps, context);
return wrapTableForScan(raw);
}
@@ -3715,6 +3727,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
directLoader),
directLoader);
rejectServerSideScanPlanning(base, handle);
+ IcebergEncryptionRead.check(base, catalogProps, context);
return base;
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/resources/iceberg.conf.template
b/fe/fe-connector/fe-connector-iceberg/src/main/resources/iceberg.conf.template
index 668c9328ac3..1fe0309ff9b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/resources/iceberg.conf.template
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/resources/iceberg.conf.template
@@ -5,10 +5,10 @@
# This file must exist on EVERY FE node -- it is not replicated through Doris
metadata.
# Changes take effect after an FE restart.
#
-# Every setting below is optional. Left commented out, each falls back to the
fe.conf key named
-# after it, and then to the built-in default -- so an existing deployment
needs no change here.
+# Every setting below is optional. Existing shared settings fall back to the
fe.conf key named
+# after them, and then to the built-in default. New settings use only their
documented default.
#
-# NOTE: both settings below are shared with other connectors at the fe.conf
end -- one
+# NOTE: the driver and metastore settings are shared with other connectors at
the fe.conf end -- one
# jdbc_drivers_dir and one hive_metastore_client_timeout_second serve jdbc,
iceberg and paimon.
# A per-plugin conf cannot express that, so if you move them here you must
also set them in
# jdbc.conf and paimon.conf. Leaving them commented out keeps the single
shared fe.conf value.
@@ -21,3 +21,8 @@
# unless the catalog overrides it. Falls back to fe.conf's
hive_metastore_client_timeout_second,
# whose default is 10.
# metastore_client_timeout_second=10
+
+# Encrypted-read admission intent only; default false, with no fe.conf
fallback.
+# Enabling this does not enable encrypted writes or bypass missing backend
read capabilities.
+# Encrypted scan execution remains unavailable until explicit backend
negotiation and readers exist.
+# enable_encryption_read=false
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeIcebergTable.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeIcebergTable.java
index bb14a86c934..39347bf4b31 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeIcebergTable.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeIcebergTable.java
@@ -42,6 +42,7 @@ import org.apache.iceberg.UpdatePartitionSpec;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
@@ -298,7 +299,7 @@ final class FakeIcebergTable implements Table {
@Override
public EncryptionManager encryption() {
- throw new UnsupportedOperationException();
+ return PlaintextEncryptionManager.instance();
}
@Override
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorConfTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorConfTest.java
index f166b94eaa0..2c8ddac01fe 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorConfTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorConfTest.java
@@ -27,8 +27,8 @@ import java.util.HashMap;
import java.util.Map;
/**
- * How this connector's two deployment-level settings are resolved: its own
{@code iceberg.conf}
- * first, then the fe.conf key each used to live under.
+ * Deployment-level settings use {@code iceberg.conf}; existing shared
settings also retain their
+ * legacy fe.conf fallback.
*
* <p>Asserted rather than trusted, because both failure directions are
silent. Reading fe.conf first
* would make an administrator's edit to iceberg.conf do nothing. Dropping the
fe.conf fallback would
@@ -37,6 +37,16 @@ import java.util.Map;
*/
public class IcebergConnectorConfTest {
+ @Test
+ public void encryptionReadIsDefaultOffAndUsesOnlyPluginConfiguration() {
+ Assertions.assertFalse(IcebergConf.encryptionReadEnabled(null));
+
Assertions.assertFalse(IcebergConf.encryptionReadEnabled(context(Map.of(),
Map.of())));
+ Assertions.assertFalse(IcebergConf.encryptionReadEnabled(
+ context(Map.of(),
Map.of(IcebergConf.CONF_ENABLE_ENCRYPTION_READ, "true"))));
+ Assertions.assertTrue(IcebergConf.encryptionReadEnabled(
+ context(Map.of(IcebergConf.CONF_ENABLE_ENCRYPTION_READ,
"true"), Map.of())));
+ }
+
@Test
public void driversDirPrefersThePluginConfThenFeConf() {
Assertions.assertEquals("/from/plugin/conf", IcebergConf.driversDir(
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergEncryptionReadTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergEncryptionReadTest.java
new file mode 100644
index 00000000000..32a0afe94a8
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergEncryptionReadTest.java
@@ -0,0 +1,297 @@
+// 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.connector.spi.ConnectorContext;
+import org.apache.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.ConnectorStatementScope;
+import org.apache.doris.connector.spi.scan.ConnectorScanRequest;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptedInputFile;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/** Real SDK tables exercise scan, metadata and COUNT entry points without
reading data files. */
+public class IcebergEncryptionReadTest {
+ private static final Schema SCHEMA = new
Schema(Types.NestedField.required(1, "id", Types.LongType.get()));
+ private static final IcebergTableHandle HANDLE = new
IcebergTableHandle("db", "tbl");
+
+ @Test
+ public void encryptedReadsAreDisabledAcrossPlanningRoutes() throws
Exception {
+ for (boolean populated : new boolean[] {false, true}) {
+ for (boolean cached : new boolean[] {false, true}) {
+ IcebergTableCache cache = new IcebergTableCache(100, 10);
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = encryptedView(table(catalog, populated), 3);
+ RecordingIcebergCatalogOps ops = ops(table);
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties("hms"),
+ ignored -> ops, context(Map.of()), null, cached ?
cache : null);
+ for (ConnectorStatementScope scope :
List.of(ConnectorStatementScope.NONE,
+ new TestStatementScope())) {
+ ConnectorSession session = new
IcebergStatementScopeTest.ScopeSession(1L, "query", scope);
+ try {
+ reject("READ_DISABLED", () ->
provider.planScan(session, request(HANDLE, false)));
+ reject("READ_DISABLED", () ->
provider.planScan(session, request(HANDLE, true)));
+ reject("READ_DISABLED", () ->
provider.canServeMetadataOnlyCount(
+ session, HANDLE, Optional.empty()));
+ reject("READ_DISABLED", () ->
provider.streamingSplitEstimate(
+ session, HANDLE, Optional.empty(), false));
+ reject("READ_DISABLED", () ->
provider.streamSplits(session, request(HANDLE, false)));
+ for (String name : List.of("files", "snapshots",
"position_deletes", "labels")) {
+ IcebergTableHandle sys =
IcebergTableHandle.forSystemTable(
+ "db", "tbl", name, -1L, null, -1L);
+ reject("READ_DISABLED", () ->
provider.planScan(session, request(sys, false)));
+ }
+ } finally {
+ scope.closeAll();
+ }
+ }
+ } finally {
+ cache.close();
+ }
+ }
+ }
+ }
+
+ @Test
+ public void
enablingReadStillRequiresSupportedCatalogAndBackendCapability() throws
Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = encryptedView(table(catalog, true), 3);
+ for (String flavor : List.of("hms", "rest", "hadoop", "jdbc",
"glue", "dlf", "s3tables")) {
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties(flavor),
+ ops(table), context(Map.of("enable_encryption_read",
"true")));
+ reject(flavor.equals("hms") ? "BACKEND_CAPABILITY_UNAVAILABLE"
: "UNSUPPORTED_CATALOG",
+ () -> provider.planScan(null, request(HANDLE, true)));
+ }
+ }
+ }
+
+ @Test
+ public void unsupportedTableVersionHasAnExplicitError() throws Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = encryptedView(table(catalog, false), 2);
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties("hms"),
+ ops(table), context(Map.of("enable_encryption_read",
"true")));
+ reject("UNSUPPORTED_TABLE_FORMAT", () -> provider.planScan(null,
request(HANDLE, false)));
+ }
+ }
+
+ @Test
+ public void metadataBindingCannotBypassReadChecks() throws Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = encryptedView(table(catalog, true), 3);
+ IcebergConnectorMetadata metadata = new IcebergConnectorMetadata(
+ ops(table), properties("hms"), context(Map.of()));
+ reject("READ_DISABLED", () -> metadata.getTableSchema(null,
HANDLE));
+ reject("READ_DISABLED", () -> metadata.beginQuerySnapshot(null,
HANDLE));
+ IcebergTableHandle sys = IcebergTableHandle.forSystemTable("db",
"tbl", "files", -1L, null, -1L);
+ reject("READ_DISABLED", () -> metadata.getTableSchema(null, sys));
+ }
+ }
+
+ @Test
+ public void
retainedEncryptionKeysAndEmptySnapshotPinsCannotBypassReadChecks() throws
Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = table(catalog, false);
+ ObjectNode json = (ObjectNode) JsonUtil.mapper().readTree(
+ TableMetadataParser.toJson(((BaseTable)
table).operations().current()));
+ ObjectNode key = json.putArray("encryption-keys").addObject();
+ key.put("key-id", "test-key");
+ key.put("encrypted-key-metadata", "AA==");
+ Table retained = new BaseTable(new StaticTableOperations(
+ TableMetadataParser.fromJson(json.toString()),
table.io()), "db.tbl");
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties("hms"), ops(retained));
+ IcebergTableHandle empty = HANDLE.withSnapshot(-1L, null, 0L);
+ reject("READ_DISABLED", () -> provider.planScan(null,
request(empty, false)));
+ reject("READ_DISABLED", () -> provider.streamSplits(null,
request(empty, false)));
+ reject("READ_DISABLED", () -> provider.planScan(null,
request(HANDLE, true)));
+ }
+ }
+
+ @Test
+ public void customEncryptionManagerIsDetectedWithoutTableProperties()
throws Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = table(catalog, false);
+ StaticTableOperations operations = new StaticTableOperations(
+ ((BaseTable) table).operations().current(), new
NoReadFileIO()) {
+ @Override
+ public EncryptionManager encryption() {
+ return new EncryptionManager() {
+ @Override
+ public InputFile decrypt(EncryptedInputFile input) {
+ throw new AssertionError("Admission must not
decrypt files");
+ }
+
+ @Override
+ public EncryptedOutputFile encrypt(OutputFile output) {
+ throw new AssertionError("Read admission must not
encrypt files");
+ }
+ };
+ }
+ };
+ Table custom = new BaseTable(operations, "db.tbl");
+ reject("READ_DISABLED", () -> new
IcebergScanPlanProvider(properties("hms"), ops(custom))
+ .planScan(null, request(HANDLE, false)));
+ reject("UNSUPPORTED_ENCRYPTION_MANAGER", () -> new
IcebergScanPlanProvider(properties("hms"),
+ ops(custom), context(Map.of("enable_encryption_read",
"true")))
+ .planScan(null, request(HANDLE, false)));
+ }
+ }
+
+ @Test
+ public void unverifiedBackendRejectsBeforeKmsManagerInitialization()
throws Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ BaseTable encrypted = (BaseTable) encryptedView(table(catalog,
false), 3);
+ Table table = new BaseTable(new StaticTableOperations(
+ encrypted.operations().current(), new NoReadFileIO()) {
+ @Override
+ public EncryptionManager encryption() {
+ throw new AssertionError("Unverified execution must not
initialize a KMS client");
+ }
+ }, "db.tbl");
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties("hms"), ops(table),
+ context(Map.of("enable_encryption_read", "true")));
+ reject("BACKEND_CAPABILITY_UNAVAILABLE", () ->
provider.planScan(null, request(HANDLE, false)));
+ }
+ }
+
+ @Test
+ public void plaintextScansAndCountsKeepWorkingWithEitherSwitchValue()
throws Exception {
+ try (InMemoryCatalog catalog = catalog()) {
+ Table table = table(catalog, true);
+ for (String enabled : List.of("false", "true")) {
+ IcebergScanPlanProvider provider = new
IcebergScanPlanProvider(properties("hadoop"),
+ ops(table), context(Map.of("enable_encryption_read",
enabled)));
+ Assertions.assertEquals(1, provider.planScan(null,
request(HANDLE, false)).size());
+ Assertions.assertEquals(10L, provider.planScan(null,
request(HANDLE, true))
+ .get(0).getPushDownRowCount());
+ Assertions.assertTrue(provider.canServeMetadataOnlyCount(null,
HANDLE, Optional.empty()));
+ }
+ }
+ }
+
+ private static InMemoryCatalog catalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Map.of());
+ catalog.createNamespace(Namespace.of("db"));
+ return catalog;
+ }
+
+ private static Table table(InMemoryCatalog catalog, boolean populated) {
+ Table table = catalog.createTable(TableIdentifier.of("db", "tbl"),
SCHEMA, PartitionSpec.unpartitioned(),
+ null, Map.of("format-version", "3"));
+ if (populated) {
+
table.newAppend().appendFile(DataFiles.builder(table.spec()).withPath("memory:///data.parquet")
+
.withFormat("parquet").withFileSizeInBytes(128).withRecordCount(10).build()).commit();
+ }
+ return table;
+ }
+
+ private static Table encryptedView(Table table, int version) throws
Exception {
+ // Metadata-only fixtures must not need a KMS or a usable key to
exercise admission control.
+ TableMetadata metadata = ((BaseTable) table).operations().current();
+ ObjectNode json = (ObjectNode)
JsonUtil.mapper().readTree(TableMetadataParser.toJson(metadata));
+ json.put("format-version", version);
+ json.with("properties").put(TableProperties.ENCRYPTION_TABLE_KEY,
"test-key");
+ metadata = TableMetadataParser.fromJson(json.toString());
+ return new BaseTable(new StaticTableOperations(metadata, new
NoReadFileIO()), "db.tbl");
+ }
+
+ private static RecordingIcebergCatalogOps ops(Table table) {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ ops.tableExists = true;
+ return ops;
+ }
+
+ private static IcebergCatalogProperties properties(String flavor) {
+ return IcebergCatalogProperties.of(Map.of("iceberg.catalog.type",
flavor));
+ }
+
+ private static ConnectorContext context(Map<String, String> config) {
+ return new ConnectorContext() {
+ @Override
+ public String getCatalogName() {
+ return "test";
+ }
+
+ @Override
+ public long getCatalogId() {
+ return 1L;
+ }
+
+ @Override
+ public Map<String, String> getConnectorConfig() {
+ return config;
+ }
+ };
+ }
+
+ private static ConnectorScanRequest request(IcebergTableHandle handle,
boolean count) {
+ return ConnectorScanRequest.builder(handle,
List.of()).countPushdown(count).build();
+ }
+
+ private static void reject(String classification, Executable action) {
+ RuntimeException failure =
Assertions.assertThrows(RuntimeException.class, action);
+ Assertions.assertTrue(failure.getMessage().contains("Iceberg
encryption [" + classification + "]"),
+ failure::getMessage);
+ Assertions.assertFalse(failure.getMessage().contains("test-key"), "key
identifiers must not be disclosed");
+ }
+
+ private static final class NoReadFileIO implements FileIO {
+ @Override
+ public InputFile newInputFile(String path) {
+ throw new AssertionError("Admission must precede manifest and data
I/O");
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ throw new AssertionError("Read admission must not write files");
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ throw new AssertionError("Read admission must not delete files");
+ }
+ }
+}
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_read_guard.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_read_guard.out
new file mode 100644
index 00000000000..4acedd38509
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_read_guard.out
@@ -0,0 +1,44 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before_declaration --
+1
+2
+3
+
+-- !plain_rows --
+1
+2
+3
+
+-- !plain_empty --
+0
+
+-- !plain_count --
+3
+
+-- !plain_count --
+3
+
+-- !plain_after_rejections --
+1
+2
+3
+
+-- !plain_rows --
+1
+2
+3
+
+-- !plain_empty --
+0
+
+-- !plain_count --
+3
+
+-- !plain_count --
+3
+
+-- !plain_after_rejections --
+1
+2
+3
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_read_guard.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_read_guard.groovy
new file mode 100644
index 00000000000..dbd07e0e81d
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_read_guard.groovy
@@ -0,0 +1,141 @@
+// 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.
+
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+
+suite("test_iceberg_encryption_read_guard", "p0,external") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ return
+ }
+ String host = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String uri = "http://${host}:${restPort}";
+ String db = "iceberg_encryption_read_guard_db"
+
+ def restRequest = { String method, String path, Map body, List<Integer>
statuses ->
+ def connection = (HttpURLConnection) new
URL("${uri}${path}").openConnection()
+ connection.setRequestMethod(method)
+ connection.setConnectTimeout(10000)
+ connection.setReadTimeout(30000)
+ try {
+ if (body != null) {
+ connection.setRequestProperty("Content-Type",
"application/json")
+ connection.setDoOutput(true)
+ connection.outputStream.withCloseable {
+ it.write(JsonOutput.toJson(body).getBytes("UTF-8"))
+ }
+ }
+ int status = connection.responseCode
+ assertTrue(statuses.contains(status), "Iceberg fixture ${method}
${path} returned ${status}")
+ // Released REST fixtures may return 200 with no body for DELETE
instead of 204.
+ if (status == 200 && method != "DELETE") {
+ return connection.inputStream.withCloseable { new
JsonSlurper().parse(it) }
+ }
+ return null
+ } finally {
+ connection.disconnect()
+ }
+ }
+
+ def createCatalog = { String name, String cacheEnabled ->
+ sql "DROP CATALOG IF EXISTS ${name}"
+ sql """CREATE CATALOG ${name} PROPERTIES (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'${uri}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${host}:${minioPort}', 's3.region' =
'us-east-1',
+ 'use_path_style' = 'true', 'meta.cache.iceberg.table.enable' =
'${cacheEnabled}',
+ 'enable_encryption_read' = 'true'
+ )"""
+ }
+
+ // Admission is determined by SDK metadata, before file decryption. Keep
the files plaintext so
+ // this regression needs no KMS and cannot mistake a late decryption error
for the admission error.
+ createCatalog("encryption_guard_seed", "false")
+ sql "CREATE DATABASE IF NOT EXISTS encryption_guard_seed.${db}"
+ for (String table : ["plain_data", "plain_empty", "declared_data",
"declared_empty"]) {
+ // Remove previous declarations through the fixture, since Doris
correctly refuses to bind them.
+ // Only this suite's tables are touched; keep all tables after the run
for failure inspection.
+ restRequest("DELETE", "/v1/namespaces/${db}/tables/${table}", null,
[200, 204, 404])
+ }
+ sql """CREATE TABLE encryption_guard_seed.${db}.plain_data (id INT)
+ PROPERTIES ('format-version' = '3')"""
+ sql """CREATE TABLE encryption_guard_seed.${db}.plain_empty (id INT)
+ PROPERTIES ('format-version' = '3')"""
+ sql """CREATE TABLE encryption_guard_seed.${db}.declared_data (id INT)
+ PROPERTIES ('format-version' = '3')"""
+ sql """CREATE TABLE encryption_guard_seed.${db}.declared_empty (id INT)
+ PROPERTIES ('format-version' = '3')"""
+ sql "INSERT INTO encryption_guard_seed.${db}.plain_data VALUES (1), (2),
(3)"
+ sql "INSERT INTO encryption_guard_seed.${db}.declared_data VALUES (1),
(2), (3)"
+ order_qt_before_declaration "SELECT id FROM
encryption_guard_seed.${db}.declared_data ORDER BY id"
+
+ for (String table : ["declared_data", "declared_empty"]) {
+ String path = "/v1/namespaces/${db}/tables/${table}"
+ def loaded = restRequest("GET", path, null, [200])
+ restRequest("POST", path, [
+ requirements: [[type: "assert-table-uuid", uuid:
loaded.metadata["table-uuid"]]],
+ updates: [[action: "set-properties", updates:
["encryption.key-id": "test-key"]]]
+ ], [200])
+ def declared = restRequest("GET", path, null, [200])
+ assertEquals("test-key",
declared.metadata.properties["encryption.key-id"],
+ "The remote table must actually declare encryption")
+ }
+
+ try {
+ for (String cacheEnabled : ["false", "true"]) {
+ String catalog = "encryption_guard_cache_${cacheEnabled}"
+ // New catalogs isolate read admission from the pre-declaration
binding and its schema cache.
+ // The catalog property above must never override the plugin's
deployment-level switch.
+ createCatalog(catalog, cacheEnabled)
+ order_qt_plain_rows "SELECT id FROM ${catalog}.${db}.plain_data
ORDER BY id"
+ qt_plain_empty "SELECT COUNT(*) FROM ${catalog}.${db}.plain_empty"
+ for (String countPushdown : ["true", "false"]) {
+ sql "SET
enable_count_push_down_for_external_table=${countPushdown}"
+ qt_plain_count "SELECT COUNT(*) FROM
${catalog}.${db}.plain_data"
+ // Repeated failures exercise a warmed table cache as well as
the initial acquisition.
+ for (int attempt = 0; attempt < 2; attempt++) {
+ for (String table : ["declared_data", "declared_empty"]) {
+ // Assert the default-off deployment contract even
when the catalog requests encryption.
+ test {
+ sql "SELECT id FROM ${catalog}.${db}.${table}
ORDER BY id"
+ exception "Iceberg encryption [READ_DISABLED]"
+ }
+ test {
+ sql "SELECT COUNT(*) FROM
${catalog}.${db}.${table}"
+ exception "Iceberg encryption [READ_DISABLED]"
+ }
+ test {
+ sql "DESC ${catalog}.${db}.${table}"
+ exception "Iceberg encryption [READ_DISABLED]"
+ }
+ for (String systemTable : ["files", "snapshots",
"position_deletes"]) {
+ test {
+ sql "SELECT COUNT(*) FROM
${catalog}.${db}.`${table}\$${systemTable}`"
+ exception "Iceberg encryption [READ_DISABLED]"
+ }
+ }
+ }
+ }
+ }
+ order_qt_plain_after_rejections "SELECT id FROM
${catalog}.${db}.plain_data ORDER BY id"
+ }
+ } finally {
+ sql "UNSET VARIABLE enable_count_push_down_for_external_table"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
