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 5fcf30f319a [fix](paimon) Authenticate deferred metadata reads and
normalized HDFS storage (#69247)
5fcf30f319a is described below
commit 5fcf30f319a24068c05dc6977a2f2160af7ef50e
Author: Gabriel <[email protected]>
AuthorDate: Tue Oct 20 09:38:26 2026 +0800
[fix](paimon) Authenticate deferred metadata reads and normalized HDFS
storage (#69247)
### What problem does this PR solve?
Problem Summary:
Queries against Kerberos-protected Paimon storage can fail with `Failed
to find latest snapshot id` and `Client cannot authenticate via: [TOKEN,
KERBEROS]`. Resolving a Table under authentication does not cover
snapshot I/O performed after the resolver returns. In addition, storage
configured with `hdfs.authentication.*` can initialize without a
plugin-side Kerberos authenticator because the selection checks only the
raw legacy Hadoop key.
Authenticate actual snapshot-cache loads, partition-cache key loads, and
the complete time-travel resolution. Reuse the metadata-read wrapper and
select the storage authenticator from the effective Hadoop
configuration, including normalized storage properties. HMS and storage
identities remain separate; cache behavior and empty-table snapshot
semantics are preserved.
The CHAR-padding regression asserts SQL rejection of complex MAP keys,
matching the upstream analyzer guard, while retaining primitive-key
padding/collision and DML coverage. The Variant NULL fixture correction
is already supplied by master through #69261.
### Release note
Fix Kerberos-authenticated Paimon snapshot reads and recognize
normalized HDFS authentication settings when selecting the plugin
storage identity.
### Check List (For Author)
- [x] Unit Test
- [x] Regression test
- [x] Manual test
- Behavior changed: authenticated Paimon metadata reads retain the
storage identity through deferred file I/O.
- Documentation needed: No; no configuration key or public API added.
Validation:
- External CI follow-up: reproduced the full
`test_paimon_write_variant_errors` suite failure on the original
`c5d8569a085` DEBUG/ASAN artifact with isolated Spark 4/Paimon 1.3.1 and
object storage. Regenerated the output with the regression runner and
reran the full suite: one passed, zero failures/fatal errors/skips.
- Reproduced 11 snapshot authentication scenarios before the fix. Added
a negative authentication/cache recovery control; all 12 pass after the
fix.
- Reproduced the normalized-storage-authenticator assertion failure
before fixing the legacy-only gate.
- Revalidated after rebasing onto master `1eb427f19eb`: Paimon module
`install` with `Paimon*` tests: 823 tests, zero failures/errors, one
skipped live-connectivity test. Checkstyle and plugin packaging passed.
- Local KDC/HMS/HDFS using the original failing DEBUG/ASAN runtime: both
Paimon cases failed at the first snapshot read before the patch.
Replacing the Paimon plugin with the built artifact made both local
suites pass (two suites, zero failed/fatal/skipped), including all four
Kerberos HMS property variants, JNI/native reads, incremental reads,
time travel, system tables, and the added REFRESH assertion.
- Local adaptations replace endpoints/keytab paths and exclude unrelated
object-store variants. The migration fixture has matching columns and
four rows, created as local Paimon tables; it does not reproduce the
original Hive-migrated physical files.
- Ten concurrent FE plans across two Kerberos identities succeeded; a
mismatched keytab was rejected. A separate intermittent BE credential
error observed with a warm backend is not claimed fixed here; the final
SQL run used a fresh backend.
Rebase validation (master `41a58afba2e`):
- Resolved the conflict by retaining master's Variant fix from #69261
and removing this PR's duplicate Variant change. The output is
identical; only the explanatory comment uses the upstream wording.
- Original product fixes and the CHAR-padding regression fix are
preserved. Range-diff confirms the unchanged patches; the Iceberg
product patch is also unchanged where applicable.
- Paimon connector `install`: 832 tests, zero failures/errors, one
live-connectivity skip; Checkstyle and plugin packaging passed.
- The end-to-end results listed below were obtained before this rebase.
Full integration validation of the new head is pending CI.
CHAR-padding regression follow-up:
- The upstream MAP-key guard (#69101) rejects STRUCT keys before the
Paimon sink, while the CHAR-padding suite (#69083) still expected those
writes to succeed or reach writer collision checks.
- All four complex-key groups now assert the specific analysis error for
both former success/collision inputs, and verify their tables remain
empty. Primitive CHAR-key padding/collision checks and partition, ORC,
bucket-routing, rowkind, UPDATE/DELETE, and MERGE coverage are retained.
Direct writer normalization tests remain unchanged.
- Reproduced the original full-suite failure on the CI runtime built
from `86cd8732c93`. Regenerated expected output with the regression
runner, then independently ran the complete
`test_paimon_write_char_padding` and `test_paimon_write_variant_errors`
suites: two passed, zero failures/fatal errors/skips. Only local
endpoints were adapted; product binaries were unchanged.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/connector/paimon/PaimonConnector.java | 8 +-
.../connector/paimon/PaimonConnectorMetadata.java | 29 +++--
.../PaimonConnectorPluginAuthenticatorTest.java | 15 +++
.../paimon/PaimonSnapshotAuthenticationTest.java | 127 +++++++++++++++++++++
.../paimon/RecordingPaimonCatalogOps.java | 13 +++
.../write/test_paimon_write_char_padding.out | 16 +--
.../write/test_paimon_write_char_padding.groovy | 85 ++++++--------
.../paimon/test_paimon_hms_catalog.groovy | 4 +
8 files changed, 228 insertions(+), 69 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
index 094883afb03..a5154b81f86 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
@@ -216,9 +216,11 @@ public class PaimonConnector implements Connector {
*/
static HadoopAuthenticator buildPluginAuthenticator(Map<String, String>
properties,
Map<String, String> storageHadoopConfig) {
- if
("kerberos".equalsIgnoreCase(properties.get(HADOOP_SECURITY_AUTHENTICATION))) {
- return HadoopAuthenticator.getHadoopAuthenticator(
- PaimonCatalogFactory.buildHadoopConfiguration(properties,
storageHadoopConfig));
+ // Storage bindings normalize hdfs.authentication.* and XML settings
into Hadoop keys.
+ // Gating on the raw legacy key would leave FileIO using the plugin's
unauthenticated UGI.
+ Configuration conf =
PaimonCatalogFactory.buildHadoopConfiguration(properties, storageHadoopConfig);
+ if
("kerberos".equalsIgnoreCase(conf.get(HADOOP_SECURITY_AUTHENTICATION))) {
+ return HadoopAuthenticator.getHadoopAuthenticator(conf);
}
return null;
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index b6dd5993eb2..245da688dd8 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -311,7 +311,7 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// are not DataTable). Sharing buildTableSchema with the at-snapshot
path keeps the two from drifting.
if (!paimonHandle.isSystemTable()) {
Optional<PaimonCatalogOps.PaimonSchemaSnapshot> latest =
- readSchemaAuthenticated(() ->
catalogOps.latestSchema(table));
+ readMetadataAuthenticated(() ->
catalogOps.latestSchema(table));
if (latest.isPresent()) {
PaimonCatalogOps.PaimonSchemaSnapshot schema = latest.get();
return buildTableSchema(
@@ -634,7 +634,8 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// on a miss; when caching is disabled (ttl-second<=0, the no-cache
catalog) it runs every call.
Identifier identifier =
Identifier.create(paimonHandle.getDatabaseName(), paimonHandle.getTableName());
long id = latestSnapshotCache.getOrLoad(identifier,
- () ->
catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L));
+ () -> readMetadataAuthenticated(
+ () ->
catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L)));
StatementPin pin = statementPin(paimonHandle,
resolveTable(paimonHandle), id);
Map<String, String> coordinates = new HashMap<>(pin.coordinates);
if (!coordinates.isEmpty()) {
@@ -644,19 +645,19 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
.retainSchema(pin.schemaId() >=
0).properties(coordinates).build());
}
- private <T> T readSchemaAuthenticated(Supplier<T> read) {
- // Cached tables do not cache schema files: exact/latest schema reads
still need plugin UGI and TCCL.
+ private <T> T readMetadataAuthenticated(Supplier<T> read) {
+ // Table resolution can return before lazy schema/snapshot I/O;
authenticate the actual read as well.
try {
return context.executeAuthenticated(read::get);
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
- throw new RuntimeException("Failed to read Paimon schema", e);
+ throw new RuntimeException("Failed to read Paimon metadata", e);
}
}
private StatementPin statementPin(PaimonTableHandle handle, Table table,
long snapshotId) {
- return statementPins.computeIfAbsent(handle, ignored ->
readSchemaAuthenticated(() -> {
+ return statementPins.computeIfAbsent(handle, ignored ->
readMetadataAuthenticated(() -> {
Optional<PaimonCatalogOps.PaimonSchemaSnapshot> schema =
catalogOps.latestSchema(table);
long schemaId =
schema.map(PaimonCatalogOps.PaimonSchemaSnapshot::schemaId).orElse(-1L);
Map<String, String> coordinates = PaimonSchemaPin.capture(table,
schemaId, snapshotId);
@@ -669,9 +670,9 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
private PaimonCatalogOps.PaimonSchemaSnapshot
schemaForPin(PaimonTableHandle handle, Table table, long schemaId) {
if (!PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) {
return schemaAtMemo.getOrLoad(handle, schemaId,
- () -> readSchemaAuthenticated(() ->
catalogOps.schemaAt(table, schemaId)));
+ () -> readMetadataAuthenticated(() ->
catalogOps.schemaAt(table, schemaId)));
}
- return readSchemaAuthenticated(() -> {
+ return readMetadataAuthenticated(() -> {
PaimonSchemaPin.validate(table, handle.getScanOptions());
// INSERT replanning can replace this scope while retaining the
source pin. Rehydrate
// the exact schema in the new scope without using the catalog's
historical memo.
@@ -752,6 +753,11 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
if (paimonHandle.isSystemTable()) {
return Optional.empty();
}
+ return readMetadataAuthenticated(() ->
resolveTimeTravelAuthenticated(session, paimonHandle, spec));
+ }
+
+ private Optional<ConnectorMvccSnapshot> resolveTimeTravelAuthenticated(
+ ConnectorSession session, PaimonTableHandle paimonHandle,
ConnectorTimeTravelSpec spec) {
Table table = resolveTable(paimonHandle);
switch (spec.getKind()) {
case SNAPSHOT_ID: {
@@ -1559,7 +1565,7 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
// rowType(). System tables keep their synthetic rowType() (no
schema-version history).
if (!paimonHandle.isSystemTable()) {
Optional<PaimonCatalogOps.PaimonSchemaSnapshot> latest =
- readSchemaAuthenticated(() ->
catalogOps.latestSchema(table));
+ readMetadataAuthenticated(() ->
catalogOps.latestSchema(table));
if (latest.isPresent()) {
return buildColumnHandles(latest.get().fields(), true);
}
@@ -1740,7 +1746,8 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
private ConnectorTableKey partitionViewCacheKey(PaimonTableHandle
paimonHandle) {
Identifier identifier =
Identifier.create(paimonHandle.getDatabaseName(), paimonHandle.getTableName());
long snapshotId = latestSnapshotCache.getOrLoad(identifier,
- () ->
catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L));
+ () -> readMetadataAuthenticated(
+ () ->
catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L)));
return new ConnectorTableKey(
paimonHandle.getDatabaseName(), paimonHandle.getTableName(),
snapshotId, -1L);
}
@@ -1991,7 +1998,7 @@ public class PaimonConnectorMetadata implements
ConnectorMetadata {
}
Table dataTable = PaimonTableResolver.resolveSystemSource(catalogOps,
handle, context);
if (PaimonScanParams.preservesBoundSchema(scanOptions)) {
- return readSchemaAuthenticated(() ->
PaimonReaderOptions.runtimeSafeSystemTable(
+ return readMetadataAuthenticated(() ->
PaimonReaderOptions.runtimeSafeSystemTable(
handle.getSysTableName(), systemTable, dataTable,
scanOptions));
}
return PaimonReaderOptions.runtimeSafeSystemTable(
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorPluginAuthenticatorTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorPluginAuthenticatorTest.java
index d965ef84453..759b7e30297 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorPluginAuthenticatorTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorPluginAuthenticatorTest.java
@@ -18,6 +18,7 @@
package org.apache.doris.connector.paimon;
import org.apache.doris.kerberos.HadoopAuthenticator;
+import org.apache.doris.kerberos.HadoopKerberosAuthenticator;
import org.apache.hadoop.conf.Configuration;
import org.apache.paimon.hive.pool.CachedClientPool;
@@ -63,6 +64,20 @@ public class PaimonConnectorPluginAuthenticatorTest {
Assertions.assertNotNull(auth, "storage kerberos must yield a plugin
authenticator");
}
+ @Test
+ public void canonicalStorageKerberosBuildsPluginAuthenticator() {
+ HadoopAuthenticator auth = PaimonConnector.buildPluginAuthenticator(
+ props("paimon.catalog.type", "hms", "warehouse",
"hdfs://ns/warehouse",
+ "hdfs.authentication.type", "kerberos",
+ "hdfs.authentication.kerberos.principal",
"[email protected]",
+ "hdfs.authentication.kerberos.keytab",
"/etc/security/storage.keytab"),
+ props("hadoop.security.authentication", "kerberos",
+ "hadoop.kerberos.principal", "[email protected]",
+ "hadoop.kerberos.keytab",
"/etc/security/storage.keytab"));
+ Assertions.assertTrue(auth instanceof HadoopKerberosAuthenticator,
+ "Canonical storage configuration must select the plugin-side
Kerberos UGI");
+ }
+
/**
* THE S6 GAP: a Kerberized HMS whose data storage is simple. Storage auth
is unset, so the storage gate is
* off; the connector must fall back to the HMS client-principal/keytab
facts and still build a plugin
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSnapshotAuthenticationTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSnapshotAuthenticationTest.java
new file mode 100644
index 00000000000..38f566cbc8e
--- /dev/null
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSnapshotAuthenticationTest.java
@@ -0,0 +1,127 @@
+// 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.paimon;
+
+import org.apache.doris.connector.cache.ConnectorMetadataCache;
+import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.stream.Stream;
+
+public class PaimonSnapshotAuthenticationTest {
+ private final RecordingConnectorContext context = new
RecordingConnectorContext();
+ private final RecordingPaimonCatalogOps ops = new
RecordingPaimonCatalogOps();
+
+ private PaimonTableHandle handle() {
+ FakePaimonTable table = new FakePaimonTable("t",
+ RowType.builder().field("region", DataTypes.STRING()).build(),
+ Collections.singletonList("region"), Collections.emptyList());
+ ops.table = table;
+ ops.latestSnapshotId = OptionalLong.of(7L);
+ ops.snapshotReadAuthentication = () -> context.authDepth > 0;
+ PaimonTableHandle handle = new PaimonTableHandle("db", "t",
+ Collections.singletonList("region"), Collections.emptyList());
+ handle.setPaimonTable(table);
+ return handle;
+ }
+
+ private PaimonConnectorMetadata metadata(PaimonLatestSnapshotCache cache) {
+ return new PaimonConnectorMetadata(ops,
PaimonCatalogProperties.of(Collections.emptyMap()), context,
+ new PaimonSchemaAtMemo(10), cache,
+ new ConnectorMetadataCache<>("paimon", "partition_view",
Collections.emptyMap()));
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {0L, 60L})
+ public void snapshotCacheLoadsInsideAuthentication(long ttl) {
+ PaimonTableHandle handle = handle();
+ PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(ttl,
10);
+ Assertions.assertEquals(7L, metadata(cache).beginQuerySnapshot(null,
handle).get().getSnapshotId());
+ ops.latestSnapshotId = OptionalLong.of(8L);
+ Assertions.assertEquals(ttl == 0 ? 8L : 7L,
+ metadata(cache).beginQuerySnapshot(null,
handle).get().getSnapshotId());
+ cache.invalidate(Identifier.create("db", "t"));
+ Assertions.assertEquals(8L, metadata(cache).beginQuerySnapshot(null,
handle).get().getSnapshotId());
+ Assertions.assertEquals(0, context.authDepth);
+ }
+
+ @Test
+ public void partitionCacheKeyLoadsSnapshotInsideAuthentication() {
+ PaimonTableHandle handle = handle();
+ Assertions.assertTrue(metadata(new PaimonLatestSnapshotCache(0, 10))
+ .listPartitionNames(null, handle).isEmpty());
+ Assertions.assertTrue(ops.log.contains("latestSnapshotId"));
+ Assertions.assertEquals(0, context.authDepth);
+ }
+
+ @Test
+ public void emptySnapshotRemainsAnEmptyPin() {
+ PaimonTableHandle handle = handle();
+ ops.latestSnapshotId = OptionalLong.empty();
+ Assertions.assertEquals(-1L, metadata(new PaimonLatestSnapshotCache(0,
10))
+ .beginQuerySnapshot(null, handle).get().getSnapshotId());
+ }
+
+ @Test
+ public void authenticationFailureDoesNotCacheAnEmptySnapshot() {
+ PaimonTableHandle handle = handle();
+ PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(60,
10);
+ context.failAuth = true;
+ Assertions.assertThrows(RuntimeException.class, () ->
metadata(cache).beginQuerySnapshot(null, handle));
+ Assertions.assertFalse(ops.log.contains("latestSnapshotId"));
+ context.failAuth = false;
+ Assertions.assertEquals(7L, metadata(cache).beginQuerySnapshot(null,
handle).get().getSnapshotId());
+ Assertions.assertEquals(0, context.authDepth);
+ }
+
+ static Stream<ConnectorTimeTravelSpec> timeTravelSpecs() {
+ return Stream.of(ConnectorTimeTravelSpec.snapshotId("7"),
+ ConnectorTimeTravelSpec.timestamp("1700000000000", true),
+ ConnectorTimeTravelSpec.tag("release"),
+ ConnectorTimeTravelSpec.versionRef("release"),
+ ConnectorTimeTravelSpec.branch("branch"),
+ ConnectorTimeTravelSpec.incremental(Map.of("startSnapshotId",
"1", "endSnapshotId", "7")),
+ ConnectorTimeTravelSpec.options(Collections.emptyMap()));
+ }
+
+ @ParameterizedTest
+ @MethodSource("timeTravelSpecs")
+ public void timeTravelReadsInsideAuthentication(ConnectorTimeTravelSpec
spec) {
+ PaimonTableHandle handle = handle();
+ ops.snapshotExists = true;
+ ops.branchExists = true;
+ ops.branchTable = ops.table;
+ ops.tagSnapshot = new PaimonCatalogOps.TagSnapshot(7L, 1L);
+ ops.snapshotIdAtOrBefore = OptionalLong.of(7L);
+ ops.snapshotSchemaId = OptionalLong.of(1L);
+ Assertions.assertEquals(7L, metadata(new PaimonLatestSnapshotCache(0,
10))
+ .resolveTimeTravel(null, handle, spec).get().getSnapshotId());
+ Assertions.assertEquals(0, context.authDepth);
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
index d5bf7252fc5..73147f40620 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java
@@ -95,6 +95,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
// ---- T20 E5 MVCC seam: configurable lookup results (no real
Snapshot/SnapshotManager) ----
OptionalLong latestSnapshotId = OptionalLong.empty();
BooleanSupplier latestSnapshotAuthentication;
+ BooleanSupplier snapshotReadAuthentication;
OptionalLong snapshotIdAtOrBefore = OptionalLong.empty();
boolean snapshotExists;
/** The table the metadata layer passed to the most recent MVCC seam call.
*/
@@ -252,8 +253,15 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
lastSchemaChanges = changes;
}
+ private void checkSnapshotAuthentication() {
+ if (snapshotReadAuthentication != null &&
!snapshotReadAuthentication.getAsBoolean()) {
+ throw new IllegalStateException("Snapshot I/O ran outside
authentication");
+ }
+ }
+
@Override
public OptionalLong latestSnapshotId(Table table) {
+ checkSnapshotAuthentication();
if (latestSnapshotAuthentication != null &&
!latestSnapshotAuthentication.getAsBoolean()) {
throw new IllegalStateException("latestSnapshotId called outside
authentication");
}
@@ -264,6 +272,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public OptionalLong snapshotIdAtOrBefore(Table table, long
timestampMillis) {
+ checkSnapshotAuthentication();
log.add("snapshotIdAtOrBefore:" + timestampMillis);
lastMvccTable = table;
snapshotIdAtOrBeforeArg = timestampMillis;
@@ -272,6 +281,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public boolean snapshotExists(Table table, long snapshotId) {
+ checkSnapshotAuthentication();
log.add("snapshotExists:" + snapshotId);
lastMvccTable = table;
return snapshotExists;
@@ -279,6 +289,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public OptionalLong snapshotSchemaId(Table table, long snapshotId) {
+ checkSnapshotAuthentication();
log.add("snapshotSchemaId:" + snapshotId);
lastMvccTable = table;
lastSnapshotSchemaIdArg = snapshotId;
@@ -287,6 +298,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public java.util.Optional<PaimonCatalogOps.TagSnapshot>
getSnapshotByTag(Table table, String tagName) {
+ checkSnapshotAuthentication();
log.add("getSnapshotByTag:" + tagName);
lastMvccTable = table;
lastTagNameArg = tagName;
@@ -310,6 +322,7 @@ final class RecordingPaimonCatalogOps implements
PaimonCatalogOps {
@Override
public boolean branchExists(Table table, String branchName) {
+ checkSnapshotAuthentication();
log.add("branchExists:" + branchName);
// Capture which table validation ran against (must be the BASE table,
mirroring legacy
// resolvePaimonBranch which validates the branch on the base table's
branchManager).
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
index dba5910049d..e1a699d7a60 100644
---
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_char_padding.out
@@ -12,17 +12,17 @@
2 2 [k |kk] 1|2
3 2 [k|k ] 1|2
--- !map_row_key_padded --
-1 1 [k ] 7
+-- !map_row_key_rejected --
+0
--- !map_nested_map_key_padded --
-2 1 [k ] 10
+-- !map_nested_map_key_rejected --
+0
--- !map_nan_key_padded --
-2 1 [k ] NaN 10
+-- !map_nan_key_rejected --
+0
--- !map_nested_nan_key_padded --
-2 2 [k ]|[k ] 10|20
+-- !map_nested_nan_key_rejected --
+0
-- !partition_padded --
1 [x ]
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
index f0631de6f2b..cdb8602d454 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_char_padding.groovy
@@ -265,41 +265,37 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
FROM t_char_map_avro
"""
- // The writer takes a map key from the Arrow batch BE sends, where a
ROW key holding a ROW is a columnar row
- // that cannot be hashed, so keys are compared by their bytes: such a
key is padded, and two keys that
- // padding makes equal fail the row as above.
- sql """INSERT INTO t_char_map_row_key VALUES (1, map(named_struct('c',
'k', 'r', named_struct('i', 1)), 7))"""
+ // SQL rejects complex MAP keys before the Paimon writer runs, even
when padding would keep them distinct.
+ // PaimonWriteSchemaTest covers the writer's complex-key normalization
directly; these statements check
+ // the SQL boundary and that rejected writes leave each table empty.
+ test {
+ sql """INSERT INTO t_char_map_row_key VALUES (1,
map(named_struct('c', 'k', 'r', named_struct('i', 1)), 7))"""
+ exception "MAP key type must be a primitive type"
+ }
test {
sql """
INSERT INTO t_char_map_row_key VALUES (2,
map(named_struct('c', 'k', 'r', named_struct('i', 1)), 1,
named_struct('c', 'k ', 'r', named_struct('i', 1)), 2))
"""
- exception "into the same key"
+ exception "MAP key type must be a primitive type"
}
- order_qt_map_row_key_padded """
- SELECT id, map_size(m), concat('[', struct_element(map_keys(m)[1],
'c'), ']'), map_values(m)[1]
- FROM t_char_map_row_key
- """
+ order_qt_map_row_key_rejected """SELECT count(*) FROM
t_char_map_row_key"""
- // Paimon compares maps regardless of the order of their entries, and
its Avro reader keeps one of two keys
- // whose nested maps differ only in that order, so such keys that
padding makes equal fail the row too. A
- // key holding a map is padded.
+ // A STRUCT containing a MAP is still a complex key, regardless of the
nested entry order.
test {
sql """
INSERT INTO t_char_map_nested_map_key VALUES (1,
map(named_struct('c', 'k', 'm', map('a', 1, 'b', 2)),
10, named_struct('c', 'k ', 'm', map('b', 2, 'a', 1)), 20))
"""
- exception "into the same key"
+ exception "MAP key type must be a primitive type"
}
- sql """INSERT INTO t_char_map_nested_map_key VALUES (2,
map(named_struct('c', 'k', 'm', map('a', 1)), 10))"""
- order_qt_map_nested_map_key_padded """
- SELECT id, map_size(m), concat('[', struct_element(map_keys(m)[1],
'c'), ']'), map_values(m)[1]
- FROM t_char_map_nested_map_key
- """
+ test {
+ sql """INSERT INTO t_char_map_nested_map_key VALUES (2,
map(named_struct('c', 'k', 'm', map('a', 1)), 10))"""
+ exception "MAP key type must be a primitive type"
+ }
+ order_qt_map_nested_map_key_rejected """SELECT count(*) FROM
t_char_map_nested_map_key"""
- // Paimon compares a DOUBLE regardless of the bits of a NaN, and its
Avro reader keeps one of two keys that
- // differ only so, while a NaN cast at run time keeps its sign. So the
keys (k, NaN) and (k , -NaN), which
- // padding makes equal, fail the row too. A key holding a NaN is
padded.
+ // Runtime NaN values do not bypass the key-type check in INSERT
SELECT.
sql """INSERT INTO internal.${dbName}.char_nan_source VALUES ('NaN',
'-NaN')"""
test {
sql """
@@ -308,21 +304,18 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
named_struct('c', 'k ', 'd', cast(neg AS DOUBLE)), 20)
FROM internal.${dbName}.char_nan_source
"""
- exception "into the same key"
+ exception "MAP key type must be a primitive type"
}
- sql """
- INSERT INTO t_char_map_nan_key
- SELECT 2, map(named_struct('c', 'k', 'd', cast(neg AS DOUBLE)),
10) FROM internal.${dbName}.char_nan_source
- """
- order_qt_map_nan_key_padded """
- SELECT id, map_size(m), concat('[', struct_element(map_keys(m)[1],
'c'), ']'),
- struct_element(map_keys(m)[1], 'd'), map_values(m)[1]
- FROM t_char_map_nan_key
- """
+ test {
+ sql """
+ INSERT INTO t_char_map_nan_key
+ SELECT 2, map(named_struct('c', 'k', 'd', cast(neg AS
DOUBLE)), 10) FROM internal.${dbName}.char_nan_source
+ """
+ exception "MAP key type must be a primitive type"
+ }
+ order_qt_map_nan_key_rejected """SELECT count(*) FROM
t_char_map_nan_key"""
- // Paimon's Avro reader keeps the last of the entries of a map whose
keys are equal, so a nested map holding
- // NaN -> 1 and -NaN -> 2 reads back as {NaN: 2}, and the keys (k,
{NaN: 1, -NaN: 2}) and (k , {NaN: 2}),
- // which padding then makes equal, fail the row too. Keys whose nested
maps keep other last values are padded.
+ // Nesting those NaNs inside another MAP also fails at the outer
STRUCT key.
test {
sql """
INSERT INTO t_char_map_nested_nan_key
@@ -330,20 +323,18 @@ suite("test_paimon_write_char_padding",
"p0,external,paimon") {
named_struct('c', 'k ', 'm', map(cast(pos AS DOUBLE), 2)),
20)
FROM internal.${dbName}.char_nan_source
"""
- exception "into the same key"
+ exception "MAP key type must be a primitive type"
}
- sql """
- INSERT INTO t_char_map_nested_nan_key
- SELECT 2, map(named_struct('c', 'k', 'm', map(cast(pos AS DOUBLE),
1, cast(neg AS DOUBLE), 2)), 10,
- named_struct('c', 'k ', 'm', map(cast(pos AS DOUBLE), 1)), 20)
- FROM internal.${dbName}.char_nan_source
- """
- order_qt_map_nested_nan_key_padded """
- SELECT id, map_size(m),
- array_join(array_sort(array_map(k -> concat('[',
struct_element(k, 'c'), ']'), map_keys(m))), '|'),
- array_join(array_sort(map_values(m)), '|')
- FROM t_char_map_nested_nan_key
- """
+ test {
+ sql """
+ INSERT INTO t_char_map_nested_nan_key
+ SELECT 2, map(named_struct('c', 'k', 'm', map(cast(pos AS
DOUBLE), 1, cast(neg AS DOUBLE), 2)), 10,
+ named_struct('c', 'k ', 'm', map(cast(pos AS DOUBLE), 1)),
20)
+ FROM internal.${dbName}.char_nan_source
+ """
+ exception "MAP key type must be a primitive type"
+ }
+ order_qt_map_nested_nan_key_rejected """SELECT count(*) FROM
t_char_map_nested_nan_key"""
// A partition value is padded whether it comes from the rows or from
a static PARTITION clause, so a
// second static overwrite of 'x' replaces the partition the first one
wrote.
diff --git
a/regression-test/suites/external_table_p2/paimon/test_paimon_hms_catalog.groovy
b/regression-test/suites/external_table_p2/paimon/test_paimon_hms_catalog.groovy
index 722b50531f5..1ef30d8a94e 100644
---
a/regression-test/suites/external_table_p2/paimon/test_paimon_hms_catalog.groovy
+++
b/regression-test/suites/external_table_p2/paimon/test_paimon_hms_catalog.groovy
@@ -54,6 +54,10 @@ suite("test_paimon_hms_catalog", "p2,external") {
sql """SELECT * FROM external_test_table FOR VERSION AS OF 1;"""
// branch/tag
// TODO(zgx): add branch/tag
+ // REFRESH must reload snapshots with the catalog identity, even after
a warm query.
+ def beforeRefresh = sql """SELECT * FROM external_test_table ORDER BY
1, 2"""
+ sql """REFRESH TABLE external_test_table"""
+ assertEquals(beforeRefresh, sql("""SELECT * FROM external_test_table
ORDER BY 1, 2"""))
// system table
sql """SELECT * FROM external_test_table\$snapshots;"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
