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 913532ce167 [feature](fe) Add Iceberg encrypted scan wire contract
(#70213)
913532ce167 is described below
commit 913532ce1673595cd552bd6f582e5b2640f45dd9
Author: Gabriel <[email protected]>
AuthorDate: Thu Nov 5 12:03:04 2026 +0800
[feature](fe) Add Iceberg encrypted scan wire contract (#70213)
### What problem does this PR solve?
Related PRs: #70159, #70192
Prepare the file metadata and diagnostic protection required for native
Iceberg encrypted reads. Data files and each delete file need
independent binary keys and AAD. Generated Thrift strings and JSON
diagnostics must not expose those values.
- Define `TFileEncryptionInfo` directly in `PlanNodes.thrift`, alongside
its data/delete consumers. The optional carrier contains a payload
version, binary file key and exact AAD bytes; malformed payloads are
rejected. There is no separate encryption IDL file.
- Assume FE and BE are deployed together. This PR adds no encryption
heartbeat fields, receiver capability negotiation, capability
persistence or downgrade/replay handling.
- Validate assigned fragment parameters/ranges and every lazy split
response. FE and BE uniformly reject encrypted execution with
`READ_UNAVAILABLE` before readers, caches or metadata COUNT shortcuts. A
denied final split batch stops producers and remains an error on
repeated fetches.
- Keep the BE file-encryption validator inside
`file_scan_range_utils.h`, alongside its only production caller, with no
separate validator header.
- Redact generated Java/C++ strings, nested output, ThriftDebugString,
Gson and Thrift JSON diagnostics. Binary/Compact retain the real
payload. The generation transform targets only the secret carrier inside
`PlanNodes_types.cpp`, leaves other plan diagnostics unchanged, is
idempotent and fails on compiler-shape drift. FE, regression-framework
and C++ generation use the same protection.
- Keep encrypted SQL reads disabled until native decryption and
authenticated, confidential transport for both fragment dispatch and
lazy split responses are implemented. The Iceberg option cannot bypass
these requirements.
The read and wire contracts document this scope. File-key extraction and
native decryption are subsequent roadmap work.
### Validation
- Full FE build and FE Checkstyle passed with Maven build cache
disabled.
- Official focused FE tests: 55 passed (10 protocol tests and 45 FE
tests), zero failures/errors/skips. Covers independent data/delete keys,
diagnostics, malformed payloads, plain scans, direct/lazy dispatch and
terminal last-batch retries.
- SQL normal comparisons passed: the wire guard suite executed 80
expected rejections with plaintext rows/counts; the existing D1
admission suite executed 96 expected rejections with plaintext
comparisons. Covers v2/v3, actual lazy planning, COUNT pushdown,
repeated queries, malformed versions and key lengths. Existing generated
result files remain valid and unchanged.
- Compiled the actual regenerated `PlanNodes_types.cpp` with the
existing C++ contract suite: 4 tests passed. Generator self-tests: 4
passed, including idempotence, compiler drift and preservation of other
plan diagnostics. Regression-framework generation also passed with the
pinned Thrift compiler.
- Clang-format 16, clang-tidy and BE build hygiene passed on affected
files. Packaged heartbeat classes no longer contain either removed field
or the removed backend state class.
- The updated guard expectations failed against the old implementation,
then passed after simplification.
- Reproduced all three `PluginDrivenScanNodePartitionBatchTest`
failures: its fixture omitted the required `scan_range`, so lazy
dispatch validation dereferenced a missing range. Completed the fixture
with the same plaintext file-scan envelope used by production and
neighboring streaming tests, preserving split identities and all
existing assertions. The official expanded FE run then passed 383 tests
across 46 classes, with zero failures/errors/skips, including all nine
partition-batch tests, plugin scan, split
producer/consumer/assignment/compaction/concurrency, and encryption
protocol/guard coverage. FE Checkstyle passed with zero violations. This
follow-up changes only the unit-test fixture.
- Inlined the unchanged BE validator and documented build-time redaction
versus runtime wire/diagnostic serialization. Four standalone ASAN C++
contract tests passed against freshly regenerated `PlanNodes` code, and
all four Python generator tests passed. Three relevant BE test
translation units passed syntax checks using the existing ASAN compile
database. Clang-format 16, clang-tidy and header hygiene passed. Python
transform logic and Thrift field definitions are unchanged.
The SQL tests use the updated FE and an existing ASAN BE. Native
encrypted SQL interoperability is outside this protocol-preparation
stage. Full BE ASAN unit execution was not rerun in this update: the
earlier attempt stopped in the unchanged baseline
`vparquet_writer_test.cpp` because its `WriteInfo` initializer still
uses removed path members. The standalone C++ tests do not replace a
full ASAN run.
### Release note
Prepare per-file Iceberg encryption metadata and redacted diagnostics.
Encrypted SQL reads remain disabled.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test (full FE build and local external stack)
- Behavior changed:
- [x] Yes. Encrypted file payloads are validated and uniformly rejected
until execution is implemented.
- Does this need documentation?
- [x] Yes. Read and wire contracts are included.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/scan/file_scan_range_utils.h | 41 ++++++
be/test/exec/scan/file_scanner_v2_test.cpp | 29 +++++
be/test/format/file_encryption_info_test.cpp | 110 ++++++++++++++++
fe/fe-connector/docs/iceberg-encryption-read.md | 35 +++---
fe/fe-connector/docs/iceberg-encryption-wire.md | 121 ++++++++++++++++++
.../datasource/scan/FileScanEncryptionGuard.java | 120 ++++++++++++++++++
.../apache/doris/datasource/split/SplitSource.java | 16 +++
.../qe/runtime/PipelineExecutionTaskBuilder.java | 2 +
.../org/apache/doris/rpc/BackendServiceProxy.java | 3 +
.../scan/FileScanEncryptionGuardTest.java | 100 +++++++++++++++
.../PluginDrivenScanNodePartitionBatchTest.java | 17 ++-
.../doris/datasource/split/SplitSourceTest.java | 20 +++
fe/fe-thrift/pom.xml | 21 ++++
.../org/apache/doris/rpc/FileScanEncryption.java | 66 ++++++++++
.../doris/thrift/FileEncryptionInfoTest.java | 115 +++++++++++++++++
.../doris/thrift/FileScanEncryptionTest.java | 68 ++++++++++
gensrc/script/redact_file_encryption.py | 140 +++++++++++++++++++++
gensrc/script/test_redact_file_encryption.py | 92 ++++++++++++++
gensrc/thrift/Makefile | 5 +
gensrc/thrift/PlanNodes.thrift | 17 +++
.../iceberg/test_iceberg_encryption_wire_guard.out | 27 ++++
regression-test/framework/pom.xml | 16 +++
.../test_iceberg_encryption_wire_guard.groovy | 97 ++++++++++++++
23 files changed, 1256 insertions(+), 22 deletions(-)
diff --git a/be/src/exec/scan/file_scan_range_utils.h
b/be/src/exec/scan/file_scan_range_utils.h
index 9b9e91fc221..a9bf6cbcedf 100644
--- a/be/src/exec/scan/file_scan_range_utils.h
+++ b/be/src/exec/scan/file_scan_range_utils.h
@@ -22,6 +22,22 @@
namespace doris {
+// Return only static classifications: malformed wire payloads must not enter
error messages.
+inline const char* file_encryption_validation_error(const TFileEncryptionInfo&
info) {
+ if (!info.__isset.version) {
+ return "File encryption [INVALID_DESCRIPTOR]";
+ }
+ if (info.version != 1) {
+ return "File encryption [UNSUPPORTED_VERSION]";
+ }
+ const auto length = info.key.size();
+ if (!info.__isset.key || (length != 16 && length != 24 && length != 32) ||
+ !info.__isset.aad_prefix) {
+ return "File encryption [INVALID_DESCRIPTOR]";
+ }
+ return nullptr;
+}
+
inline Status validate_file_scan_support(const TFileScanRangeParams& params,
const TFileRangeDesc* range =
nullptr) {
// Deprecated table markers must never fall through to plain file reads:
that would lose
@@ -33,6 +49,31 @@ inline Status validate_file_scan_support(const
TFileScanRangeParams& params,
if (table.table_format_type == "lakesoul") {
return Status::NotSupported("LakeSoul reads are no longer
supported");
}
+ if (table.__isset.iceberg_params) {
+ const auto& iceberg = table.iceberg_params;
+ bool encrypted = iceberg.__isset.encryption_info;
+ if (encrypted) {
+ if (const auto* error =
file_encryption_validation_error(iceberg.encryption_info)) {
+ return Status::InvalidArgument(error);
+ }
+ }
+ if (iceberg.__isset.delete_files) {
+ for (const auto& file : iceberg.delete_files) {
+ if (file.__isset.encryption_info) {
+ if (const auto* error =
+
file_encryption_validation_error(file.encryption_info)) {
+ return Status::InvalidArgument(error);
+ }
+ encrypted = true;
+ }
+ }
+ }
+ // A plain data file can still have encrypted deletes. Reject
before reader selection,
+ // caches or metadata COUNT shortcuts until native decryption is
implemented.
+ if (encrypted) {
+ return Status::NotSupported("File encryption
[READ_UNAVAILABLE]");
+ }
+ }
return Status::OK();
};
if (params.__isset.table_format_params) {
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index b29daffcb62..ee0a7c42fd6 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -41,6 +41,7 @@
#include "exec/operator/file_scan_operator.h"
#include "exec/runtime_filter/runtime_filter_definitions.h"
#include "exec/scan/file_scan_io_context.h"
+#include "exec/scan/file_scan_range_utils.h"
#include "exec/scan/split_source_connector.h"
#include "exprs/create_predicate_function.h"
#include "exprs/runtime_filter_expr.h"
@@ -68,6 +69,34 @@ TFileRangeDesc range_with_format(std::string table_format,
TFileFormatType::type
return range;
}
+TEST(FileScannerV2Test, EncryptionNeverFallsThroughToPlainReadersOrCounts) {
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ EXPECT_TRUE(validate_file_scan_support(params).ok());
+ for (int content : {0, 1, 2, 3}) {
+ auto range = range_with_format("iceberg",
TFileFormatType::FORMAT_PARQUET);
+ TFileEncryptionInfo info;
+ info.__set_version(1);
+ info.__set_key(std::string(16, 'k'));
+ info.__set_aad_prefix("");
+ TIcebergFileDesc iceberg;
+ if (content == 0) {
+ iceberg.__set_encryption_info(info);
+ } else {
+ TIcebergDeleteFileDesc file;
+ file.__set_content(content);
+ file.__set_encryption_info(info);
+ iceberg.__set_delete_files({file});
+ }
+ range.table_format_params.__set_iceberg_params(iceberg);
+ EXPECT_NE(validate_file_scan_support(params,
&range).to_string().find("READ_UNAVAILABLE"),
+ std::string::npos);
+ params.__set_table_format_params(range.table_format_params);
+ EXPECT_FALSE(validate_file_scan_support(params).ok());
+ params.__isset.table_format_params = false;
+ }
+}
+
TEST(FileScannerV2Test, CountStarPlaceholderIsNotASemanticProjection) {
EXPECT_TRUE(ScanLocalStateBase::is_count_star_pushdown(TPushAggOp::type::COUNT,
std::vector<int32_t> {}));
diff --git a/be/test/format/file_encryption_info_test.cpp
b/be/test/format/file_encryption_info_test.cpp
new file mode 100644
index 00000000000..a62b69a3ce7
--- /dev/null
+++ b/be/test/format/file_encryption_info_test.cpp
@@ -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.
+
+#include <gtest/gtest.h>
+#include <thrift/TToString.h>
+#include <thrift/protocol/TBinaryProtocol.h>
+#include <thrift/protocol/TCompactProtocol.h>
+#include <thrift/protocol/TDebugProtocol.h>
+#include <thrift/protocol/TJSONProtocol.h>
+#include <thrift/transport/TBufferTransports.h>
+
+#include <memory>
+#include <sstream>
+#include <string>
+#include <vector>
+
+#include "exec/scan/file_scan_range_utils.h"
+
+namespace doris {
+namespace {
+
+TFileEncryptionInfo encryption_info() {
+ TFileEncryptionInfo info;
+ info.__set_version(1);
+ std::string key(16, 'k');
+ key[0] = '\0';
+ key[1] = static_cast<char>(255);
+ info.__set_key(key);
+ info.__set_aad_prefix(std::string("\0aad\xff", 5));
+ return info;
+}
+
+TEST(FileEncryptionInfoTest, ValidatePresenceVersionAndKeySizes) {
+ auto info = encryption_info();
+ EXPECT_EQ(file_encryption_validation_error(info), nullptr);
+ info.__set_aad_prefix("");
+ for (size_t length : {16, 24, 32}) {
+ info.__set_key(std::string(length, 'k'));
+ EXPECT_EQ(file_encryption_validation_error(info), nullptr);
+ }
+ info.__isset.aad_prefix = false;
+ EXPECT_STREQ(file_encryption_validation_error(info), "File encryption
[INVALID_DESCRIPTOR]");
+ info = encryption_info();
+ info.__set_key("");
+ EXPECT_STREQ(file_encryption_validation_error(info), "File encryption
[INVALID_DESCRIPTOR]");
+ info = encryption_info();
+ info.__set_version(9);
+ EXPECT_STREQ(file_encryption_validation_error(info), "File encryption
[UNSUPPORTED_VERSION]");
+ EXPECT_STREQ(file_encryption_validation_error(TFileEncryptionInfo()),
+ "File encryption [INVALID_DESCRIPTOR]");
+}
+
+template <typename Protocol>
+void round_trip() {
+ auto buffer = std::make_shared<apache::thrift::transport::TMemoryBuffer>();
+ Protocol protocol(buffer);
+ const auto original = encryption_info();
+ original.write(&protocol);
+ TFileEncryptionInfo decoded;
+ decoded.read(&protocol);
+ EXPECT_EQ(decoded, original);
+}
+
+TEST(FileEncryptionInfoTest, BinaryAndCompactPreserveEveryByte) {
+ using namespace apache::thrift::protocol;
+ using apache::thrift::transport::TMemoryBuffer;
+ round_trip<TBinaryProtocol>();
+ round_trip<TCompactProtocol>();
+ round_trip<TBinaryProtocolT<TMemoryBuffer>>();
+ round_trip<TCompactProtocolT<TMemoryBuffer>>();
+}
+
+TEST(FileEncryptionInfoTest, NestedPrintAndThriftDebugStringAreRedacted) {
+ const auto info = encryption_info();
+ std::ostringstream stream;
+ stream << info;
+ EXPECT_EQ(stream.str(), "TFileEncryptionInfo(<redacted>)");
+ const auto nested =
apache::thrift::to_string(std::vector<TFileEncryptionInfo> {info});
+ EXPECT_NE(nested.find("<redacted>"), std::string::npos);
+ EXPECT_EQ(nested.find("kkkk"), std::string::npos);
+ const auto debug = apache::thrift::ThriftDebugString(info);
+ EXPECT_NE(debug.find("redacted"), std::string::npos);
+ EXPECT_EQ(debug.find("kkkk"), std::string::npos);
+}
+
+TEST(FileEncryptionInfoTest, JsonIsDiagnosticOnly) {
+ auto buffer = std::make_shared<apache::thrift::transport::TMemoryBuffer>();
+ apache::thrift::protocol::TJSONProtocol protocol(buffer);
+ encryption_info().write(&protocol);
+ const auto diagnostic = buffer->getBufferAsString();
+ EXPECT_NE(diagnostic.find("redacted"), std::string::npos);
+ EXPECT_EQ(diagnostic.find("a2tr"), std::string::npos);
+}
+
+} // namespace
+} // namespace doris
diff --git a/fe/fe-connector/docs/iceberg-encryption-read.md
b/fe/fe-connector/docs/iceberg-encryption-read.md
index eefa46a372d..4c63c7b0ef1 100644
--- a/fe/fe-connector/docs/iceberg-encryption-read.md
+++ b/fe/fe-connector/docs/iceberg-encryption-read.md
@@ -27,7 +27,7 @@ The initial target matrix is:
| 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 |
+| Backend | Native encrypted reader and protected FE/BE key transport | Both
remain unavailable, 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
@@ -63,25 +63,20 @@ 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.
+## Requirements before enabling execution
+
+The scan protocol carries each data/delete file's binary key and AAD
independently.
+FE and BE are deployed together for this feature; no heartbeat capability
+negotiation is required. FE must reject encrypted dispatch and lazy split
+responses until the native reader and protected internal key transport are
ready.
+BE must also reject encrypted data and deletes before reader, cache or COUNT
+paths. Repeated split fetches must preserve an existing rejection.
+
+The connector currently reports `BACKEND_CAPABILITY_UNAVAILABLE` before file
+descriptors are created. The [scan wire contract](iceberg-encryption-wire.md)
+defines per-file payloads, diagnostic redaction and the later dispatch guard.
+File keys remain query-scoped, with bounded lifetimes; they must not enter
EXPLAIN
+output, profiles or persistent scan caches.
See the upstream [Iceberg encryption
documentation](https://iceberg.apache.org/docs/nightly/encryption/)
and [table specification](https://iceberg.apache.org/spec/#encryption-keys) for
diff --git a/fe/fe-connector/docs/iceberg-encryption-wire.md
b/fe/fe-connector/docs/iceberg-encryption-wire.md
new file mode 100644
index 00000000000..4cb62e08635
--- /dev/null
+++ b/fe/fe-connector/docs/iceberg-encryption-wire.md
@@ -0,0 +1,121 @@
+# Iceberg encrypted scan wire contract
+
+This is protocol preparation. It does not enable encrypted SQL reads, implement
+native decryption, or establish a protected RPC channel. D1 admission remains
in
+place; D4 also rejects secret-bearing ranges before dispatch and on the BE.
+
+## File payload
+
+`PlanNodes.thrift:TFileEncryptionInfo` is a query-scoped secret carrier.
+Its optional fields are `1: version`, `2: binary key`, and `3: binary
aad_prefix`.
+Version 1 means Parquet AES_GCM_V1 with the SDK-unwrapped file key and exact
AAD
+prefix bytes. AAD is binary; do not convert it to text or reconstruct it from
the
+file path. Native reader initialization belongs to subsequent work.
+
+Each `TIcebergFileDesc` has optional field **14** for the data-file payload.
+Each `TIcebergDeleteFileDesc` has optional field **12** for its own payload.
+Data and delete files never inherit each other's keys. This also preserves a
+plaintext data file with encrypted deletes, and encrypted data with plaintext
+deletes. These numbers were checked against master and branches 4.2, 4.1, 4.0
+and 3.1; data-file field 13 remains reserved for historical equality-delete
schemas.
+
+Absence of the entire payload retains the existing plaintext wire semantics.
+Presence requires an explicitly supported version, a 16/24/32-byte key and a
+present AAD field, including an empty prefix. Empty structures, missing fields,
+invalid key sizes and future versions fail with static classifications:
+`INVALID_DESCRIPTOR` or `UNSUPPORTED_VERSION`. They never fall back to
plaintext.
+Table admission still detects encrypted declarations before file descriptors
exist.
+
+Binary and Compact Thrift preserve every byte, including NUL and non-UTF-8
bytes.
+Java's existing object serialization also uses explicit Compact Thrift
encoding.
+These encodings are transport representations, not permission to persist keys.
+Never put them in durable plan dumps, persistent caches or audit records.
+
+## Dispatch
+
+FE and BE are deployed together for this feature. This change does not
introduce
+heartbeat capability fields, version negotiation, downgrade handling or backend
+capability persistence. Native encrypted execution remains uniformly disabled
+until the reader and protected key transport are implemented.
+
+The pipeline task builder validates assigned file parameters and ranges before
+serialization. The direct typed fragment RPC entry has the same restriction.
+Each lazy `SplitSource` batch is checked before the FE response is populated.
+A rejected lazy batch stops its producers and keeps a terminal failure,
including
+when the final batch was already drained. Replanned or reassigned attempts
repeat
+the checks. Errors contain classifications, not payloads. The BE checks both
data
+and delete payloads before reader selection,
+I/O, cache lookup and metadata COUNT handling, even if FE sent an unsupported
range.
+
+## Transport prerequisite
+
+Before native reads can open, **both** fragment dispatch from FE to BE and lazy
+split responses from FE to BE must have authenticated, confidential transport.
+The latter uses the separate BE-to-FE Thrift connection. Every selected
endpoint,
+including retry targets, must meet this requirement. HTTPS for SQL, catalog or
+object storage access does not establish protection of either internal channel.
+
+The OSS fragment channel provider currently selects plaintext gRPC. D4
therefore
+rejects every encrypted dispatch with `READ_UNAVAILABLE`. There is no
configuration
+flag that asserts the transport is secure. A subsequent change must implement
the
+native reader and verify actual protection of both channels before lifting this
+restriction. Enabling the Iceberg plugin option cannot bypass it.
+
+## Diagnostic redaction
+
+Thrift 0.24 does not enforce sensitive annotations. The normal FE generation,
+regression-framework generation and C++ generation apply the same narrowly
scoped
+`gensrc/script/redact_file_encryption.py` transform. A compiler shape mismatch
+fails generation. The transform is idempotent and leaves wire field IDs
intact. C++ redaction
+targets only `TFileEncryptionInfo` inside the generated `PlanNodes_types.cpp`;
+other plan descriptors retain their ordinary diagnostic output.
+
+- Generated Java `toString()` and C++ `printTo()` redact the entire carrier,
+ including when nested inside scan descriptors and plan parameters.
+- C++ `ThriftDebugString` serializes through a debug protocol, so changing
+ `printTo()` alone is insufficient. Only supported Binary/Compact wire
protocol
+ implementations receive the actual fields; diagnostic protocols receive a
+ redacted placeholder. Java JSON Thrift protocols behave the same way.
+- Generated Java instance fields are transient for reflective diagnostic Gson.
+ Thrift explicitly reads/writes the fields, so RPC and existing Java object
+ serialization are unchanged. Default Gson and FE's Gson configuration omit
the
+ carrier fields. Non-wire diagnostic JSON is deliberately unable to restore
keys.
+
+Do not log serialized RPC byte strings or manually format key/AAD buffers.
+This change does not promise memory zeroization, protected crash dumps or a
+key-scoped cache lifecycle; those remain required before final read enablement.
+
+## Validation
+
+`FileEncryptionInfoTest` covers Java Binary/Compact round trips, distinct data
+and delete values, absent payloads, object serialization, nested strings, Gson
+and both JSON Thrift protocols. `FileScanEncryptionTest` checks descriptor
shape.
+FE guard tests cover independent delete kinds, unknown payload versions,
repeated
+dispatch and rejection before execution. SplitSource tests preserve a terminal
+error when a rejected last batch is fetched again.
+`FileEncryptionInfoTest` in BE covers binary bytes, key sizes, stream printing,
+nested containers, `ThriftDebugString` and JSON; FileScannerV2 has a pre-reader
+rejection test. Generator self-tests cover idempotence and compiler drift.
+
+Run the generator checks with:
+
+```sh
+python3 gensrc/script/test_redact_file_encryption.py
+```
+
+`test_iceberg_encryption_wire_guard` is an external SQL regression. It uses
+plaintext v2/v3 tables, then a path-scoped debug point injects synthetic
descriptors
+at the dispatch/fetch guard. Eighty expected rejections cover data/delete
payloads,
+invalid/future descriptors, synchronous/streaming scans, count pushdown on/off
and
+repeated queries. EXPLAIN verifies that v2 actually takes the lazy path while
+current v3 planning remains synchronous. Plaintext rows/counts are compared
before
+and after. This tests protocol enforcement; the real encrypted HMS fixture is
separate metadata work.
+
+```sh
+bash run-regression-test.sh --run -d external_table_p0/iceberg \
+ -s test_iceberg_encryption_wire_guard
+```
+
+The standard Iceberg stack and FE debug points must be enabled. No separate
+feature-mode variable is needed. These tests do not replace a future complete
+native encrypted SQL interoperability run.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanEncryptionGuard.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanEncryptionGuard.java
new file mode 100644
index 00000000000..c9b7a5220e2
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileScanEncryptionGuard.java
@@ -0,0 +1,120 @@
+// 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.datasource.scan;
+
+import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.common.util.DebugPointUtil.DebugPoint;
+import org.apache.doris.rpc.FileScanEncryption;
+import org.apache.doris.thrift.TFileEncryptionInfo;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TFileScanRangeParams;
+import org.apache.doris.thrift.TIcebergDeleteFileDesc;
+import org.apache.doris.thrift.TIcebergFileDesc;
+import org.apache.doris.thrift.TPipelineFragmentParams;
+import org.apache.doris.thrift.TPipelineFragmentParamsList;
+import org.apache.doris.thrift.TPipelineInstanceParams;
+import org.apache.doris.thrift.TScanRange;
+import org.apache.doris.thrift.TScanRangeLocations;
+import org.apache.doris.thrift.TScanRangeParams;
+import org.apache.doris.thrift.TTableFormatFileDesc;
+
+import java.util.List;
+
+/** Last checks before serializing file keys, including lazy split responses
and replanned retries. */
+public final class FileScanEncryptionGuard {
+ private FileScanEncryptionGuard() {
+ }
+
+ private static void rejectEncryptedRead() {
+ // Native decryption and protected transport for file keys are not
implemented yet.
+ throw new IllegalStateException("File encryption [READ_UNAVAILABLE]");
+ }
+
+ public static void checkFragments(TPipelineFragmentParamsList fragments) {
+ if (!fragments.isSetParamsList()) {
+ return;
+ }
+ for (TPipelineFragmentParams fragment : fragments.getParamsList()) {
+ if (fragment.isSetFileScanParams()) {
+ for (TFileScanRangeParams params :
fragment.getFileScanParams().values()) {
+ checkParams(params);
+ }
+ }
+ if (fragment.isSetLocalParams()) {
+ for (TPipelineInstanceParams instance :
fragment.getLocalParams()) {
+ if (instance.isSetPerNodeScanRanges()) {
+ for (List<TScanRangeParams> ranges :
instance.getPerNodeScanRanges().values()) {
+ for (TScanRangeParams range : ranges) {
+ checkRange(range.getScanRange());
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ public static void checkBatch(List<TScanRangeLocations> ranges) {
+ for (TScanRangeLocations range : ranges) {
+ checkRange(range.getScanRange());
+ }
+ }
+
+ private static void checkParams(TFileScanRangeParams params) {
+ if (params.isSetTableFormatParams() &&
FileScanEncryption.validate(params.getTableFormatParams())) {
+ rejectEncryptedRead();
+ }
+ }
+
+ private static void checkRange(TScanRange range) {
+ if (!range.isSetExtScanRange() ||
!range.getExtScanRange().isSetFileScanRange()) {
+ return;
+ }
+ TFileScanRange fileScan = range.getExtScanRange().getFileScanRange();
+ if (fileScan.isSetParams()) {
+ checkParams(fileScan.getParams());
+ }
+ if (fileScan.isSetRanges()) {
+ DebugPoint injection =
DebugPointUtil.getDebugPoint("FileScanEncryptionGuard.inject");
+ for (TFileRangeDesc file : fileScan.getRanges()) {
+ TTableFormatFileDesc table = file.getTableFormatParams();
+ if (injection != null && file.isSetPath()
+ &&
file.getPath().contains(injection.param("path_contains",
"iceberg_encryption_wire_db"))) {
+ // Synthetic descriptors let SQL regressions exercise
dispatch without opening encrypted reads.
+ table = injectedDescriptor(injection);
+ }
+ if (table != null && FileScanEncryption.validate(table)) {
+ rejectEncryptedRead();
+ }
+ }
+ }
+ }
+
+ private static TTableFormatFileDesc injectedDescriptor(DebugPoint point) {
+ TFileEncryptionInfo info = new
TFileEncryptionInfo().setVersion(point.param("version", 1))
+ .setKey(new byte[point.param("invalid_key", false) ? 0 :
16]).setAadPrefix(new byte[] {0, (byte) 255});
+ TIcebergFileDesc file = new TIcebergFileDesc();
+ if (point.param("delete", false)) {
+ file.setDeleteFiles(List.of(new
TIcebergDeleteFileDesc().setEncryptionInfo(info)));
+ } else {
+ file.setEncryptionInfo(info);
+ }
+ return new TTableFormatFileDesc().setIcebergParams(file);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
index 1488bfd25ed..e5ff2dcaa9d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
@@ -18,6 +18,7 @@
package org.apache.doris.datasource.split;
import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.scan.FileScanEncryptionGuard;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TScanRangeLocations;
@@ -58,6 +59,8 @@ public class SplitSource {
private final AtomicBoolean firstBatchRequested = new AtomicBoolean();
private final long maxWaitTime;
private final int initialBatchSize;
+ // Accessed only under getNextBatch's monitor; a denied final batch must
not turn into EOF on retry.
+ private UserException dispatchFailure;
public SplitSource(Backend backend, SplitAssignment splitAssignment, long
maxWaitTime) {
this(backend, splitAssignment, maxWaitTime, Integer.MAX_VALUE);
@@ -84,6 +87,9 @@ public class SplitSource {
*/
public synchronized List<TScanRangeLocations> getNextBatch(int
maxBatchSize) throws UserException {
Preconditions.checkArgument(maxBatchSize > 0, "Requested split batch
size must be positive");
+ if (dispatchFailure != null) {
+ throw dispatchFailure;
+ }
if (isLastBatch.get()) {
return Collections.emptyList();
}
@@ -158,6 +164,16 @@ public class SplitSource {
throw new UserException(e.getMessage(), e);
}
}
+ // Lazy splits leave FE after initial dispatch; apply the same checks
before each RPC response.
+ try {
+ FileScanEncryptionGuard.checkBatch(scanRanges);
+ } catch (IllegalArgumentException | IllegalStateException e) {
+ // The rejected ranges were already drained. Stop every producer
and preserve the error,
+ // including when this source had marked its last batch before
checking the descriptors.
+ dispatchFailure = new UserException(e.getMessage(), e);
+ splitAssignment.setException(dispatchFailure);
+ throw dispatchFailure;
+ }
return FileScanRangeCompactor.compactSplitBatch(scanRanges);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTaskBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTaskBuilder.java
index 03f6f923dc2..60d7925506b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTaskBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTaskBuilder.java
@@ -20,6 +20,7 @@ package org.apache.doris.qe.runtime;
import org.apache.doris.common.Pair;
import org.apache.doris.common.profile.ExecutionProfile;
import org.apache.doris.common.profile.SummaryProfile;
+import org.apache.doris.datasource.scan.FileScanEncryptionGuard;
import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
import org.apache.doris.qe.CoordinatorContext;
@@ -118,6 +119,7 @@ public class PipelineExecutionTaskBuilder {
.parallelStream()
.map(kv -> {
try {
+ FileScanEncryptionGuard.checkFragments(kv.getValue());
ByteString serializeString =
new TFastSerializer(1024, new
Factory()).serialize(kv.getValue());
return Pair.of(kv.getKey(), serializeString);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
index fd9ffa9a19c..4b687c18590 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.Status;
import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.datasource.scan.FileScanEncryptionGuard;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.proto.InternalService;
import org.apache.doris.proto.InternalService.PAlterVaultSyncRequest;
@@ -169,6 +170,8 @@ public class BackendServiceProxy {
public Future<InternalService.PExecPlanFragmentResult>
execPlanFragmentsAsync(TNetworkAddress address,
TPipelineFragmentParamsList params, boolean twoPhaseExecution)
throws TException, RpcException {
+ // This direct entry bypasses the pipeline task builder, but has the
same transport restriction.
+ FileScanEncryptionGuard.checkFragments(params);
InternalService.PExecPlanFragmentRequest.Builder builder =
InternalService.PExecPlanFragmentRequest.newBuilder();
if (Config.use_compact_thrift_rpc) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileScanEncryptionGuardTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileScanEncryptionGuardTest.java
new file mode 100644
index 00000000000..675bab25980
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileScanEncryptionGuardTest.java
@@ -0,0 +1,100 @@
+// 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.datasource.scan;
+
+import org.apache.doris.thrift.TExternalScanRange;
+import org.apache.doris.thrift.TFileEncryptionInfo;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TFileScanRangeParams;
+import org.apache.doris.thrift.TIcebergDeleteFileDesc;
+import org.apache.doris.thrift.TIcebergFileDesc;
+import org.apache.doris.thrift.TPipelineFragmentParams;
+import org.apache.doris.thrift.TPipelineFragmentParamsList;
+import org.apache.doris.thrift.TPipelineInstanceParams;
+import org.apache.doris.thrift.TScanRange;
+import org.apache.doris.thrift.TScanRangeLocations;
+import org.apache.doris.thrift.TScanRangeParams;
+import org.apache.doris.thrift.TTableFormatFileDesc;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+class FileScanEncryptionGuardTest {
+ @Test
+ void plaintextScansRemainAvailable() {
+ FileScanEncryptionGuard.checkBatch(List.of(location(new
TIcebergFileDesc())));
+ FileScanEncryptionGuard.checkFragments(new
TPipelineFragmentParamsList());
+ }
+
+ @Test
+ void dataAndEachDeleteKindAreRejectedBeforeDispatchOrLazyResponse() {
+ for (int content : List.of(0, 1, 2, 3)) {
+ TIcebergFileDesc iceberg = content == 0 ? new
TIcebergFileDesc().setEncryptionInfo(info())
+ : new TIcebergFileDesc().setDeleteFiles(List.of(new
TIcebergDeleteFileDesc()
+ .setContent(content).setEncryptionInfo(info())));
+ TScanRangeLocations location = location(iceberg);
+ assertUnavailable(() ->
FileScanEncryptionGuard.checkBatch(List.of(location)));
+ TPipelineFragmentParamsList fragments = fragments(location);
+ assertUnavailable(() ->
FileScanEncryptionGuard.checkFragments(fragments));
+ // A repeated dispatch must remain rejected until encrypted
execution is implemented.
+ assertUnavailable(() ->
FileScanEncryptionGuard.checkFragments(fragments));
+ }
+ }
+
+ @Test
+ void queryLevelFileParamsCannotBypassTheCheck() {
+ TPipelineFragmentParams fragment = new
TPipelineFragmentParams().setFileScanParams(Map.of(1,
+ new TFileScanRangeParams().setTableFormatParams(new
TTableFormatFileDesc()
+ .setIcebergParams(new
TIcebergFileDesc().setEncryptionInfo(info())))));
+ assertUnavailable(() -> FileScanEncryptionGuard.checkFragments(
+ new
TPipelineFragmentParamsList().setParamsList(List.of(fragment))));
+ }
+
+ @Test
+ void malformedAndFutureDescriptorsDoNotBecomePlaintext() {
+ for (TFileEncryptionInfo value : List.of(new TFileEncryptionInfo(),
info().setVersion(9))) {
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
FileScanEncryptionGuard.checkBatch(
+ List.of(location(new
TIcebergFileDesc().setEncryptionInfo(value)))));
+ }
+ }
+
+ private static TFileEncryptionInfo info() {
+ return new TFileEncryptionInfo().setVersion(1).setKey(new
byte[16]).setAadPrefix(new byte[] {0, (byte) 255});
+ }
+
+ private static TScanRangeLocations location(TIcebergFileDesc iceberg) {
+ return new TScanRangeLocations().setScanRange(new
TScanRange().setExtScanRange(new TExternalScanRange()
+ .setFileScanRange(new TFileScanRange().setRanges(List.of(new
TFileRangeDesc()
+ .setTableFormatParams(new
TTableFormatFileDesc().setIcebergParams(iceberg)))))));
+ }
+
+ private static TPipelineFragmentParamsList fragments(TScanRangeLocations
location) {
+ return new TPipelineFragmentParamsList().setParamsList(List.of(new
TPipelineFragmentParams()
+ .setLocalParams(List.of(new
TPipelineInstanceParams().setPerNodeScanRanges(Map.of(1,
+ List.of(new
TScanRangeParams().setScanRange(location.getScanRange()))))))));
+ }
+
+ private static void assertUnavailable(Runnable action) {
+ IllegalStateException error =
Assertions.assertThrows(IllegalStateException.class, action::run);
+ Assertions.assertEquals("File encryption [READ_UNAVAILABLE]",
error.getMessage());
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
index 1ee8cf67cee..a14a3fe76e6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
@@ -32,6 +32,11 @@ import org.apache.doris.datasource.split.SplitSource;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.spi.Split;
import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TExternalScanRange;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TFileScanRangeParams;
+import org.apache.doris.thrift.TScanRange;
import org.apache.doris.thrift.TScanRangeLocation;
import org.apache.doris.thrift.TScanRangeLocations;
@@ -248,8 +253,16 @@ class PluginDrivenScanNodePartitionBatchTest {
return result;
});
assignment = new SplitAssignment(policy,
Mockito.mock(SplitGenerator.class),
- (be, properties, split, keys, admission) -> new
TScanRangeLocations().setLocations(
- Collections.singletonList(new
TScanRangeLocation().setBackendId(split.getStart()))),
+ (be, properties, split, keys, admission) -> {
+ // Lazy dispatch validates file envelopes even when
tests only observe backend IDs.
+ TFileScanRange files = new
TFileScanRange().setParams(new TFileScanRangeParams())
+ .setRanges(Collections.singletonList(
+ new TFileRangeDesc().setPath("file-" +
split.getStart())));
+ return new TScanRangeLocations().setScanRange(new
TScanRange().setExtScanRange(
+ new
TExternalScanRange().setFileScanRange(files)))
+ .setLocations(Collections.singletonList(
+ new
TScanRangeLocation().setBackendId(split.getStart())));
+ },
Collections.emptyMap(), Collections.emptyList(), true,
batchSize, queueCapacity);
node = Mockito.mock(PluginDrivenScanNode.class,
Mockito.CALLS_REAL_METHODS);
Deencapsulation.setField(node, "splitAssignment", assignment);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
index 9b089eaf2ec..61137ccdbb5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
@@ -24,11 +24,14 @@ import org.apache.doris.qe.VarAttrDef;
import org.apache.doris.spi.Split;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TExternalScanRange;
+import org.apache.doris.thrift.TFileEncryptionInfo;
import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TFileScanRange;
import org.apache.doris.thrift.TFileScanRangeParams;
+import org.apache.doris.thrift.TIcebergFileDesc;
import org.apache.doris.thrift.TScanRange;
import org.apache.doris.thrift.TScanRangeLocations;
+import org.apache.doris.thrift.TTableFormatFileDesc;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
@@ -440,6 +443,23 @@ public class SplitSourceTest {
Assertions.assertTrue(second.getNextBatch(1).isEmpty());
}
+ @Test
+ void encryptionRejectionRemainsTerminalAfterTheLastBatchWasDrained()
throws Exception {
+ TFileRangeDesc file =
ranges.get(0).getScanRange().getExtScanRange().getFileScanRange().getRanges().get(0);
+ file.setTableFormatParams(new
TTableFormatFileDesc().setIcebergParams(new TIcebergFileDesc()
+ .setEncryptionInfo(new TFileEncryptionInfo().setVersion(1)
+ .setKey(new byte[16]).setAadPrefix(new byte[0]))));
+ assignment.addToQueue(splits.subList(0, 2));
+ assignment.finishSchedule();
+ SplitSource source = new SplitSource(backend, assignment, 1000);
+ UserException failure = Assertions.assertThrows(UserException.class,
() -> source.getNextBatch(10));
+
Assertions.assertTrue(failure.getMessage().contains("READ_UNAVAILABLE"));
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class, () -> source.getNextBatch(10)),
+ "A denied last batch must not become a successful EOF on
retry");
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class,
+ () -> new SplitSource(backend, assignment,
1000).getNextBatch(10)));
+ }
+
@Test
void producerFailureAfterPendingFetchIsStillAnError() throws Exception {
SplitSource source = new SplitSource(backend, assignment, 20);
diff --git a/fe/fe-thrift/pom.xml b/fe/fe-thrift/pom.xml
index 7eba4ca2576..e9f1192e7ad 100644
--- a/fe/fe-thrift/pom.xml
+++ b/fe/fe-thrift/pom.xml
@@ -78,6 +78,11 @@ under the License.
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.code.gson</groupId>
+ <artifactId>gson</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -112,6 +117,22 @@ under the License.
</target>
</configuration>
</execution>
+ <execution>
+ <id>redact-file-encryption-diagnostics</id>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <exec executable="python3" failonerror="true">
+ <arg
value="${project.basedir}/../../gensrc/script/redact_file_encryption.py"/>
+ <arg value="--java"/>
+ <arg
value="${project.build.directory}/generated-sources/thrift/org/apache/doris/thrift/TFileEncryptionInfo.java"/>
+ </exec>
+ </target>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
diff --git
a/fe/fe-thrift/src/main/java/org/apache/doris/rpc/FileScanEncryption.java
b/fe/fe-thrift/src/main/java/org/apache/doris/rpc/FileScanEncryption.java
new file mode 100644
index 00000000000..9feb9ed81b4
--- /dev/null
+++ b/fe/fe-thrift/src/main/java/org/apache/doris/rpc/FileScanEncryption.java
@@ -0,0 +1,66 @@
+// 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.rpc;
+
+import org.apache.doris.thrift.TFileEncryptionInfo;
+import org.apache.doris.thrift.TIcebergDeleteFileDesc;
+import org.apache.doris.thrift.TIcebergFileDesc;
+import org.apache.doris.thrift.TTableFormatFileDesc;
+
+/** Wire validation shared by scan planning and dispatch; never formats secret
payloads. */
+public final class FileScanEncryption {
+ public static final int VERSION = 1;
+
+ private FileScanEncryption() {
+ }
+
+ public static void validateInfo(TFileEncryptionInfo info) {
+ if (!info.isSetVersion()) {
+ throw new IllegalArgumentException("File encryption
[INVALID_DESCRIPTOR]");
+ }
+ if (info.getVersion() != VERSION) {
+ throw new IllegalArgumentException("File encryption
[UNSUPPORTED_VERSION]");
+ }
+ // Generated binary accessors copy key bytes; validation only needs
the existing buffer length.
+ int keyLength = info.isSetKey() ? info.key.remaining() : 0;
+ if ((keyLength != 16 && keyLength != 24 && keyLength != 32) ||
!info.isSetAadPrefix()) {
+ throw new IllegalArgumentException("File encryption
[INVALID_DESCRIPTOR]");
+ }
+ }
+
+ /** Returns true for data encryption or any independently encrypted delete
file. */
+ public static boolean validate(TTableFormatFileDesc table) {
+ if (!table.isSetIcebergParams()) {
+ return false;
+ }
+ TIcebergFileDesc iceberg = table.getIcebergParams();
+ boolean encrypted = iceberg.isSetEncryptionInfo();
+ if (encrypted) {
+ validateInfo(iceberg.getEncryptionInfo());
+ }
+ if (iceberg.isSetDeleteFiles()) {
+ for (TIcebergDeleteFileDesc delete : iceberg.getDeleteFiles()) {
+ if (delete.isSetEncryptionInfo()) {
+ validateInfo(delete.getEncryptionInfo());
+ encrypted = true;
+ }
+ }
+ }
+ return encrypted;
+ }
+}
diff --git
a/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileEncryptionInfoTest.java
b/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileEncryptionInfoTest.java
new file mode 100644
index 00000000000..98c37132067
--- /dev/null
+++
b/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileEncryptionInfoTest.java
@@ -0,0 +1,115 @@
+// 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.thrift;
+
+import com.google.gson.Gson;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TJSONProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
+import org.apache.thrift.protocol.TSimpleJSONProtocol;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+
+class FileEncryptionInfoTest {
+ private static final byte[] KEY =
"synthetic-key-12".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] AAD = new byte[] {0, (byte) 255, 7, 0};
+ private static final byte[] DELETE_KEY = new byte[] {0, (byte) 255, 7, 0,
1, 2, 3, 4,
+ 5, 6, 7, 8, 9, 10, 11, 12};
+
+ @Test
+ void binaryAndCompactKeepIndependentDataAndDeleteKeys() throws Exception {
+ for (TProtocolFactory protocol : List.of(new
TBinaryProtocol.Factory(), new TCompactProtocol.Factory())) {
+ TIcebergFileDesc original = descriptor();
+ TIcebergFileDesc decoded = new TIcebergFileDesc();
+ new TDeserializer(protocol).deserialize(decoded, new
TSerializer(protocol).serialize(original));
+ Assertions.assertEquals(original, decoded);
+ Assertions.assertArrayEquals(KEY,
decoded.getEncryptionInfo().getKey());
+ Assertions.assertArrayEquals(AAD,
decoded.getEncryptionInfo().getAadPrefix());
+ Assertions.assertArrayEquals(DELETE_KEY,
decoded.getDeleteFiles().get(0).getEncryptionInfo().getKey());
+
Assertions.assertFalse(decoded.getDeleteFiles().get(1).isSetEncryptionInfo());
+ }
+ }
+
+ @Test
+ void javaObjectSerializationStillUsesTheExplicitThriftEncoding() throws
Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
+ output.writeObject(descriptor());
+ }
+ try (ObjectInputStream input = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray()))) {
+ Assertions.assertEquals(descriptor(), input.readObject());
+ }
+ }
+
+ @Test
+ void absentFieldsRoundTripAsPlaintext() throws Exception {
+ TIcebergFileDesc plain = new TIcebergFileDesc().setFormatVersion(3);
+ TIcebergFileDesc decoded = new TIcebergFileDesc();
+ new TDeserializer().deserialize(decoded, new
TSerializer().serialize(plain));
+ Assertions.assertFalse(decoded.isSetEncryptionInfo());
+ }
+
+ @Test
+ void nestedGeneratedStringsNeverPrintSecrets() {
+ String diagnostic = descriptor().toString();
+ Assertions.assertTrue(diagnostic.contains("<redacted>"), diagnostic);
+ Assertions.assertFalse(diagnostic.contains("73 79 6E 74"), diagnostic);
+ Assertions.assertFalse(diagnostic.contains("00 FF 07 00"), diagnostic);
+ }
+
+ @Test
+ void nestedGsonNeverReflectsSecretFields() {
+ String diagnostic = new Gson().toJson(Map.of("data",
descriptor().getEncryptionInfo(),
+ "deletes",
List.of(descriptor().getDeleteFiles().get(0).getEncryptionInfo())));
+ Assertions.assertFalse(diagnostic.contains("synthetic-key"),
diagnostic);
+ Assertions.assertFalse(diagnostic.contains("key"), diagnostic);
+ Assertions.assertFalse(diagnostic.contains("aad_prefix"), diagnostic);
+ }
+
+ @Test
+ void diagnosticThriftProtocolsAreLossyAndRedacted() throws Exception {
+ for (TProtocolFactory protocol : List.of(new TJSONProtocol.Factory(),
new TSimpleJSONProtocol.Factory())) {
+ String diagnostic = new String(new
TSerializer(protocol).serialize(descriptor()), StandardCharsets.UTF_8);
+ Assertions.assertTrue(diagnostic.contains("redacted"), diagnostic);
+ for (byte[] secret : List.of(KEY, AAD, DELETE_KEY)) {
+ Assertions.assertFalse(diagnostic.contains(
+
Base64.getEncoder().withoutPadding().encodeToString(secret)), diagnostic);
+ }
+ }
+ }
+
+ private static TIcebergFileDesc descriptor() {
+ return new TIcebergFileDesc().setFormatVersion(3)
+ .setEncryptionInfo(new
TFileEncryptionInfo().setVersion(1).setKey(KEY).setAadPrefix(AAD))
+ .setDeleteFiles(List.of(new
TIcebergDeleteFileDesc().setEncryptionInfo(
+ new
TFileEncryptionInfo().setVersion(1).setKey(DELETE_KEY).setAadPrefix(KEY)),
+ new TIcebergDeleteFileDesc()));
+ }
+}
diff --git
a/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileScanEncryptionTest.java
b/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileScanEncryptionTest.java
new file mode 100644
index 00000000000..5b41731e667
--- /dev/null
+++
b/fe/fe-thrift/src/test/java/org/apache/doris/thrift/FileScanEncryptionTest.java
@@ -0,0 +1,68 @@
+// 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.thrift;
+
+import org.apache.doris.rpc.FileScanEncryption;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+class FileScanEncryptionTest {
+ @Test
+ void absentPayloadMeansPlaintextAndDeletePayloadIsIndependent() {
+ TTableFormatFileDesc table = new
TTableFormatFileDesc().setIcebergParams(new TIcebergFileDesc());
+ Assertions.assertFalse(FileScanEncryption.validate(table));
+ table.getIcebergParams().setDeleteFiles(List.of(new
TIcebergDeleteFileDesc().setEncryptionInfo(valid())));
+ Assertions.assertTrue(FileScanEncryption.validate(table));
+ table.getIcebergParams().setEncryptionInfo(valid());
+ Assertions.assertTrue(FileScanEncryption.validate(table));
+ }
+
+ @Test
+ void emptyOrMalformedPayloadNeverBecomesPlaintext() {
+ TFileEncryptionInfo missingVersion = valid();
+ missingVersion.unsetVersion();
+ for (TFileEncryptionInfo info : List.of(new TFileEncryptionInfo(),
missingVersion,
+ new TFileEncryptionInfo().setVersion(1).setKey(new byte[16]),
+ valid().setKey(new byte[15]), valid().setKey(new byte[0]))) {
+ IllegalArgumentException error =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> FileScanEncryption.validateInfo(info));
+
Assertions.assertTrue(error.getMessage().contains("INVALID_DESCRIPTOR"));
+ }
+ }
+
+ @Test
+ void unknownVersionsAreRejectedWithoutPrintingTheDescriptor() {
+ IllegalArgumentException error =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> FileScanEncryption.validateInfo(valid().setVersion(9)));
+ Assertions.assertEquals("File encryption [UNSUPPORTED_VERSION]",
error.getMessage());
+ }
+
+ @Test
+ void emptyAadIsPresentAndValidForAllParquetKeySizes() {
+ for (int length : List.of(16, 24, 32)) {
+ FileScanEncryption.validateInfo(valid().setKey(new byte[length]));
+ }
+ }
+
+ private static TFileEncryptionInfo valid() {
+ return new TFileEncryptionInfo().setVersion(1).setKey(new
byte[16]).setAadPrefix(new byte[0]);
+ }
+}
diff --git a/gensrc/script/redact_file_encryption.py
b/gensrc/script/redact_file_encryption.py
new file mode 100644
index 00000000000..fd728bd845a
--- /dev/null
+++ b/gensrc/script/redact_file_encryption.py
@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Protect file-key diagnostics by transforming generated Thrift code at build
time.
+
+The FE Thrift module and regression framework invoke this in Maven
process-sources,
+after Java generation; gensrc/thrift/Makefile invokes it after C++ generation.
+Queries execute the transformed code, without invoking this Python script.
+
+TFileEncryptionInfo intentionally carries an unwrapped per-file key and AAD
from
+FE to BE for reader setup. Default generated strings and JSON can expose those
+bytes in diagnostics, so redact them while preserving the actual RPC payload.
+
+Thrift 0.24 ignores sensitive annotations. Keep this transform narrow and fail
+on generator drift rather than silently compiling an unredacted representation.
+Only binary/compact protocols (and Java tuple encoding) carry the real payload.
+"""
+
+import argparse
+import pathlib
+import re
+
+MARKER = "DORIS_FILE_ENCRYPTION_REDACTED"
+
+
+def replace_body(source, signature, body):
+ matches = list(re.finditer(signature + r"\s*\{", source))
+ if len(matches) != 1:
+ raise ValueError("Unexpected generated secret-carrier method shape")
+ start = matches[0].end()
+ depth = 1
+ end = start
+ while depth and end < len(source):
+ depth += (source[end] == "{") - (source[end] == "}")
+ end += 1
+ if depth:
+ raise ValueError("Unterminated generated secret-carrier method")
+ return source[:start] + "\n" + body + "\n" + source[end - 1:]
+
+
+def redact_java(source):
+ source = replace_body(source, r"public java\.lang\.String toString\(\)",
+ ' return "TFileEncryptionInfo(<redacted>)";')
+ signature = r"public void write\(org\.apache\.thrift\.protocol\.TProtocol
oprot\) throws org\.apache\.thrift\.TException"
+ guard = """ // Diagnostic protocols must never serialize query-scoped
secrets.
+ if (!(oprot instanceof org.apache.thrift.protocol.TBinaryProtocol
+ || oprot instanceof org.apache.thrift.protocol.TCompactProtocol)) {
+ oprot.writeStructBegin(new
org.apache.thrift.protocol.TStruct("TFileEncryptionInfo"));
+ oprot.writeFieldBegin(new org.apache.thrift.protocol.TField(
+ "redacted", org.apache.thrift.protocol.TType.STRING, (short) 1000));
+ oprot.writeString("<redacted>");
+ oprot.writeFieldEnd();
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ return;
+ }
+ scheme(oprot).write(oprot, this);"""
+ source = replace_body(source, signature, guard)
+ # Gson omits transient fields. Thrift reads/writes fields explicitly, so
wire bytes are unchanged.
+ source = re.sub(r"(?m)^( (?:public|private|protected))
(?!static\b|transient\b)"
+ r"([^\n(){};]*;)", r"\1 transient \2", source)
+ for field in ("version", "key", "aad_prefix"):
+ if not re.search(r"(?m)^ public transient [^\n;]*\b" + field + r";",
source):
+ raise ValueError("Unexpected generated secret-carrier fields")
+ return source + "\n// " + MARKER + "\n"
+
+
+def redact_cpp(source):
+ includes = """#include <thrift/protocol/TBinaryProtocol.h>
+#include <thrift/protocol/TCompactProtocol.h>
+#include <thrift/transport/TBufferTransports.h>
+"""
+ if source.startswith(includes):
+ source = source[len(includes):]
+ source = re.sub(r"\n // DORIS_FILE_ENCRYPTION_WIRE_GUARD_BEGIN.*?"
+ r" // DORIS_FILE_ENCRYPTION_WIRE_GUARD_END\n", "",
source, flags=re.S)
+ source = includes + source
+ source = replace_body(source, r"void
TFileEncryptionInfo::printTo\(std::ostream& out\) const",
+ ' out << "TFileEncryptionInfo(<redacted>)";')
+ pattern = r"(uint32_t
TFileEncryptionInfo::write\(::apache::thrift::protocol::TProtocol\* oprot\)
const \{)"
+ guard = """
+ // DORIS_FILE_ENCRYPTION_WIRE_GUARD_BEGIN
+ // ThriftDebugString uses write(), not printTo(). Only wire protocols may
see keys.
+ using namespace ::apache::thrift::protocol;
+ using ::apache::thrift::transport::TMemoryBuffer;
+ if (dynamic_cast<TBinaryProtocol*>(oprot) == nullptr
+ && dynamic_cast<TCompactProtocol*>(oprot) == nullptr
+ && dynamic_cast<TBinaryProtocolT<TMemoryBuffer>*>(oprot) == nullptr
+ && dynamic_cast<TCompactProtocolT<TMemoryBuffer>*>(oprot) == nullptr) {
+ uint32_t xfer = oprot->writeStructBegin("TFileEncryptionInfo");
+ xfer += oprot->writeFieldBegin("redacted", T_STRING, 1000);
+ xfer += oprot->writeString("<redacted>");
+ xfer += oprot->writeFieldEnd();
+ xfer += oprot->writeFieldStop();
+ xfer += oprot->writeStructEnd();
+ return xfer;
+ }
+ // DORIS_FILE_ENCRYPTION_WIRE_GUARD_END
+"""
+ matches = list(re.finditer(pattern, source))
+ if len(matches) != 1:
+ raise ValueError("Unexpected generated secret-carrier writer shape")
+ point = matches[0].end()
+ return source[:point] + guard + source[point:] + "\n// " + MARKER + "\n"
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--java", type=pathlib.Path)
+ parser.add_argument("--cpp", type=pathlib.Path)
+ args = parser.parse_args()
+ if not args.java and not args.cpp:
+ parser.error("Expected a generated Java or C++ file")
+ for path, transform in [(args.java, redact_java), (args.cpp, redact_cpp)]:
+ if path is None:
+ continue
+ source = path.read_text(encoding="utf-8")
+ source = source.replace("\n// " + MARKER + "\n", "")
+ transformed = transform(source)
+ if transformed != path.read_text(encoding="utf-8"):
+ path.write_text(transformed, encoding="utf-8")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/gensrc/script/test_redact_file_encryption.py
b/gensrc/script/test_redact_file_encryption.py
new file mode 100644
index 00000000000..59f83db05bf
--- /dev/null
+++ b/gensrc/script/test_redact_file_encryption.py
@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Test the build-time redaction transform with the repository's Thrift
compiler.
+
+Check generated-code shape, idempotence and preservation of unrelated
diagnostics.
+The Java and C++ contract tests separately exercise runtime protocol round
trips
+and diagnostic output from the transformed code.
+"""
+
+import os
+import pathlib
+import re
+import subprocess
+import tempfile
+import unittest
+
+import redact_file_encryption as redact
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+
+
+class RedactFileEncryptionTest(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.temporary = tempfile.TemporaryDirectory()
+ cls.output = pathlib.Path(cls.temporary.name)
+ thirdparty = pathlib.Path(os.environ.get("DORIS_THIRDPARTY", ROOT /
"thirdparty"))
+ for language in ("java", "cpp"):
+ destination = cls.output / language
+ destination.mkdir()
+ subprocess.run([str(thirdparty / "installed/bin/thrift"), "-r",
"--gen", language,
+ "-I", str(ROOT / "gensrc/thrift"), "-I", str(ROOT
/ "gensrc/build/thrift"),
+ "-out", str(destination), str(ROOT /
"gensrc/thrift/PlanNodes.thrift")],
+ check=True, capture_output=True)
+ cls.java = (cls.output /
"java/org/apache/doris/thrift/TFileEncryptionInfo.java").read_text()
+ cls.cpp = (cls.output / "cpp/PlanNodes_types.cpp").read_text()
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.temporary.cleanup()
+
+ def test_java_transform_is_idempotent(self):
+ transformed = redact.redact_java(self.java)
+ # CLI strips its single file-level marker before transforming again.
+ again = redact.redact_java(transformed.replace("\n// " + redact.MARKER
+ "\n", ""))
+ self.assertEqual(transformed, again)
+ self.assertIn("public transient", transformed)
+ self.assertNotIn("TBaseHelper.toString(this.key", transformed)
+
+ def test_cpp_transform_is_idempotent(self):
+ transformed = redact.redact_cpp(self.cpp)
+ again = redact.redact_cpp(transformed.replace("\n// " + redact.MARKER
+ "\n", ""))
+ self.assertEqual(transformed, again)
+ diagnostic = re.search(r"void
TFileEncryptionInfo::printTo\(std::ostream& out\) const \{.*?\n\}",
+ transformed, re.S).group()
+ self.assertNotIn("to_string(key)", diagnostic)
+ self.assertIn("TCompactProtocolT<TMemoryBuffer>", transformed)
+
+ def test_generator_drift_fails_closed(self):
+ with self.assertRaises(ValueError):
+ redact.redact_java(self.java.replace("toString()", "renamed()"))
+ with self.assertRaises(ValueError):
+ redact.redact_java(self.java.replace("aad_prefix;", "renamed;"))
+ with self.assertRaises(ValueError):
+ redact.redact_cpp(self.cpp.replace("::write(", "::renamed("))
+
+ def test_other_plan_descriptors_keep_their_diagnostics(self):
+ transformed = redact.redact_cpp(self.cpp)
+ signature = r"void TFileRangeDesc::printTo\(std::ostream& out\) const
\{.*?\n\}"
+ original = re.search(signature, self.cpp, re.S)
+ self.assertIsNotNone(original)
+ self.assertEqual(original.group(), re.search(signature, transformed,
re.S).group())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/gensrc/thrift/Makefile b/gensrc/thrift/Makefile
index f6a196390ee..6cde1d66f03 100644
--- a/gensrc/thrift/Makefile
+++ b/gensrc/thrift/Makefile
@@ -28,11 +28,16 @@ GEN_OBJECTS = $(patsubst ${BUILD_DIR}/thrift/%.thrift,
${BUILD_DIR}/gen_cpp/%_ty
all: ${GEN_OBJECTS} ${OBJECTS}
.PHONY: all
+.DELETE_ON_ERROR:
THRIFT_CPP_ARGS = -I ${CURDIR} -I ${BUILD_DIR}/thrift/ --gen
cpp:moveable_types,no_skeleton -out ${BUILD_DIR}/gen_cpp --allow-64bit-consts
-strict
${BUILD_DIR}/gen_cpp:
mkdir -p $@
# handwrite thrift
+${BUILD_DIR}/gen_cpp/PlanNodes_types.cpp: ${CURDIR}/PlanNodes.thrift
${CURDIR}/../script/redact_file_encryption.py | ${BUILD_DIR}/gen_cpp
+ ${THRIFT} ${THRIFT_CPP_ARGS} $<
+ python3 ${CURDIR}/../script/redact_file_encryption.py --cpp $@
+
${BUILD_DIR}/gen_cpp/%_types.cpp: ${CURDIR}/%.thrift | ${BUILD_DIR}/gen_cpp
${THRIFT} ${THRIFT_CPP_ARGS} $<
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index fefb20b140f..f755d53a202 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -322,6 +322,19 @@ struct TFileAttributes {
1001: optional bool ignore_csv_redundant_col;
}
+// Per-file decryption payload for FE-to-BE reader setup. Execution remains
blocked until
+// native decryption and protected transport are available.
+// redact_file_encryption.py runs after code generation to hide these fields
in diagnostics.
+// Only binary/compact RPC preserves this payload; diagnostic JSON is
deliberately lossy.
+struct TFileEncryptionInfo {
+ // Version 1: Parquet AES_GCM_V1, with an unwrapped file key and the exact
SDK AAD prefix.
+ 1: optional i32 version;
+ // Unwrapped file data-encryption key for the BE reader; never a KMS
master key.
+ 2: optional binary key;
+ // Presence is required by version 1, including when the prefix is empty.
+ 3: optional binary aad_prefix;
+}
+
struct TIcebergDeleteFileDesc {
1: optional string path;
2: optional i64 position_lower_bound;
@@ -339,6 +352,8 @@ struct TIcebergDeleteFileDesc {
// Referenced data file path. Required to materialize rows from deletion
vectors.
10: optional string referenced_data_file_path;
11: optional i64 file_size;
+ // Independent of the data file key. Absence retains plaintext semantics.
+ 12: optional TFileEncryptionInfo encryption_info;
}
struct TIcebergFileDesc {
@@ -365,6 +380,8 @@ struct TIcebergFileDesc {
// Reserved for branch-4.1 / branch-4.2 field `13: optional
ExternalTableSchema.TSchema
// equality_delete_schema` (the split-local schema of dropped
equality-delete keys). Master
// carries those schemas in TFileScanRangeParams.history_schema_info. Do
not reuse this id.
+ // Absence retains plaintext semantics; an empty or unsupported payload
must never do so.
+ 14: optional TFileEncryptionInfo encryption_info;
}
struct TPaimonDeletionFileDesc {
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.out
new file mode 100644
index 00000000000..be46ecfa4b0
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.out
@@ -0,0 +1,27 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !plain_before --
+1
+2
+3
+
+-- !plain_before --
+1
+2
+3
+
+-- !plain_after --
+1
+2
+3
+
+-- !plain_count --
+3
+
+-- !plain_after --
+1
+2
+3
+
+-- !plain_count --
+3
+
diff --git a/regression-test/framework/pom.xml
b/regression-test/framework/pom.xml
index cd48d8c1b22..b3d713528b4 100644
--- a/regression-test/framework/pom.xml
+++ b/regression-test/framework/pom.xml
@@ -204,6 +204,22 @@ under the License.
</target>
</configuration>
</execution>
+ <execution>
+ <id>redact-file-encryption-diagnostics</id>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <exec executable="python3" failonerror="true">
+ <arg
value="${doris.home}/gensrc/script/redact_file_encryption.py"/>
+ <arg value="--java"/>
+ <arg
value="${project.build.directory}/generated-sources/thrift/org/apache/doris/thrift/TFileEncryptionInfo.java"/>
+ </exec>
+ </target>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.groovy
new file mode 100644
index 00000000000..49f6cb96309
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_encryption_wire_guard.groovy
@@ -0,0 +1,97 @@
+// 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.
+
+suite("test_iceberg_encryption_wire_guard", "p0,external,nonConcurrent") {
+ 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")
+ sql "DROP CATALOG IF EXISTS encryption_wire_regression"
+ sql """CREATE CATALOG encryption_wire_regression PROPERTIES (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'http://${host}:${restPort}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${host}:${minioPort}', 's3.region' =
'us-east-1',
+ 'use_path_style' = 'true'
+ )"""
+ sql "CREATE DATABASE IF NOT EXISTS
encryption_wire_regression.iceberg_encryption_wire_db"
+ for (String version : ["2", "3"]) {
+ sql "DROP TABLE IF EXISTS
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}"
+ sql """CREATE TABLE
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
(id INT)
+ PROPERTIES ('format-version' = '${version}')"""
+ sql "INSERT INTO
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
VALUES (1), (2), (3)"
+ order_qt_plain_before """SELECT id FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
+ ORDER BY id"""
+ }
+ String point = "FileScanEncryptionGuard.inject"
+ try {
+ sql "SET num_files_in_batch_mode=1"
+ for (String version : ["2", "3"]) {
+ for (String batch : ["false", "true"]) {
+ sql "SET enable_external_table_batch_mode=${batch}"
+ // Current v3 planning is synchronous; v2 supplies the actual
lazy split path.
+ explain {
+ sql "SELECT id FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}"
+ if (version == "2" && batch == "true") {
+ contains "(approximate)inputSplitNum"
+ } else {
+ notContains "(approximate)inputSplitNum"
+ }
+ }
+ for (String pushdown : ["false", "true"]) {
+ sql "SET
enable_count_push_down_for_external_table=${pushdown}"
+ for (boolean delete : [false, true]) {
+ GetDebugPoint().enableDebugPointForAllFEs(point,
[delete: delete,
+ path_contains: "iceberg_encryption_wire_db"])
+ // Injection is after Iceberg admission. These queries
exercise real dispatch and lazy fetch guards.
+ for (int attempt = 0; attempt < 2; attempt++) {
+ test {
+ sql "SELECT id FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
ORDER BY id"
+ exception "File encryption [READ_UNAVAILABLE]"
+ }
+ test {
+ sql "SELECT COUNT(*) FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}"
+ exception "File encryption [READ_UNAVAILABLE]"
+ }
+ }
+ GetDebugPoint().disableDebugPointForAllFEs(point)
+ }
+ for (Map malformed : [[version: 9, expected:
"UNSUPPORTED_VERSION"],
+ [invalid_key: true, expected:
"INVALID_DESCRIPTOR"]]) {
+ GetDebugPoint().enableDebugPointForAllFEs(point,
malformed.findAll { it.key != "expected" })
+ test {
+ sql "SELECT id FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
ORDER BY id"
+ exception "File encryption [${malformed.expected}]"
+ }
+ GetDebugPoint().disableDebugPointForAllFEs(point)
+ }
+ }
+ }
+ }
+ } finally {
+ GetDebugPoint().disableDebugPointForAllFEs(point)
+ sql "UNSET VARIABLE enable_external_table_batch_mode"
+ sql "UNSET VARIABLE num_files_in_batch_mode"
+ sql "UNSET VARIABLE enable_count_push_down_for_external_table"
+ }
+ for (String version : ["2", "3"]) {
+ order_qt_plain_after """SELECT id FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}
+ ORDER BY id"""
+ qt_plain_count "SELECT COUNT(*) FROM
encryption_wire_regression.iceberg_encryption_wire_db.guard_data_v${version}"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
