This is an automated email from the ASF dual-hosted git repository.
leaves12138 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1a67a4e104 [rest] Propagate dependency read context in headers (#8823)
1a67a4e104 is described below
commit 1a67a4e104624eb642727bc322e530b637ea037a
Author: YeJunHao <[email protected]>
AuthorDate: Wed Jul 29 14:17:01 2026 +0800
[rest] Propagate dependency read context in headers (#8823)
---
.../main/java/org/apache/paimon/rest/RESTApi.java | 9 ++
.../paimon/table/AppendOnlyFileStoreTable.java | 2 +-
.../paimon/table/BlobDescriptorReaderFactory.java | 10 +-
.../apache/paimon/table/CatalogEnvironment.java | 46 +++++++
.../apache/paimon/rest/MockRESTCatalogTest.java | 43 ++++++
.../org/apache/paimon/rest/RESTCatalogServer.java | 9 ++
.../paimon/table/CatalogEnvironmentTest.java | 146 +++++++++++++++++++++
7 files changed, 262 insertions(+), 3 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index a49982d680..8da06988c3 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -150,6 +150,15 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public class RESTApi {
public static final String HEADER_PREFIX = "header.";
+ /**
+ * Optional header carrying the URL-encoded {@link Identifier} JSON of the
table which initiated
+ * a dependency read.
+ *
+ * <p>This header only provides request context. Servers must not treat it
as authorization
+ * proof.
+ */
+ public static final String READ_VIA_HEADER = "X-Paimon-Read-Via";
+
public static final String MAX_RESULTS = "maxResults";
public static final String PAGE_TOKEN = "pageToken";
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
index 4d35147c17..f5f59ee70c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
@@ -131,7 +131,7 @@ public class AppendOnlyFileStoreTable extends
AbstractFileStoreTable {
? new DataEvolutionTableRead(
providerFactories,
schema(),
- catalogEnvironment.catalogContext(),
+ catalogEnvironment.dependencyReadContext(),
() -> new AppendTableRead(providerFactories, schema()))
: new AppendTableRead(providerFactories, schema());
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
index a7c29ebfdb..bcbef29804 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
@@ -49,14 +50,19 @@ public final class BlobDescriptorReaderFactory {
}
private static UriReaderFactory fromSourceTable(FileStoreTable table,
String sourceTable) {
+ CatalogEnvironment catalogEnvironment = table.catalogEnvironment();
CatalogLoader catalogLoader =
checkNotNull(
- table.catalogEnvironment().catalogLoader(),
+ catalogEnvironment.catalogLoader(),
"Option '%s' is not supported for tables without a
catalog loader, "
+ "including external tables in REST
catalogs.",
BLOB_DESCRIPTOR_SOURCE_TABLE.key());
Identifier sourceIdentifier = Identifier.fromString(sourceTable);
- try (Catalog catalog = catalogLoader.load()) {
+ CatalogContext dependencyContext =
catalogEnvironment.dependencyReadContext();
+ try (Catalog catalog =
+ dependencyContext == catalogEnvironment.catalogContext()
+ ? catalogLoader.load()
+ : CatalogFactory.createCatalog(dependencyContext)) {
FileIO sourceFileIO = catalog.getTable(sourceIdentifier).fileIO();
// Initialize lazy credentials before serializing FileIO to
distributed workers.
sourceFileIO.isObjectStore();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
index cb40c4447e..5f75b7d0e5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -30,8 +30,14 @@ import org.apache.paimon.catalog.RenamingSnapshotCommit;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.catalog.TableRollback;
import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.RESTApi;
+import org.apache.paimon.rest.RESTCatalogFactory;
+import org.apache.paimon.rest.RESTCatalogLoader;
+import org.apache.paimon.rest.RESTUtil;
import org.apache.paimon.table.source.TableQueryAuth;
import org.apache.paimon.tag.SnapshotLoaderImpl;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.SnapshotLoader;
import org.apache.paimon.utils.SnapshotManager;
@@ -41,10 +47,13 @@ import java.io.Serializable;
import java.util.Optional;
import java.util.function.LongConsumer;
+import static org.apache.paimon.options.CatalogOptions.METASTORE;
+
/** Catalog environment in table which contains log factory, metastore client
factory. */
public class CatalogEnvironment implements Serializable {
private static final long serialVersionUID = 2L;
+ private static final String READ_VIA_OPTION = RESTApi.HEADER_PREFIX +
RESTApi.READ_VIA_HEADER;
@Nullable private final Identifier identifier;
@Nullable private final String uuid;
@@ -196,6 +205,43 @@ public class CatalogEnvironment implements Serializable {
return catalogContext;
}
+ /**
+ * Returns a context for loading tables referenced while reading this
table.
+ *
+ * <p>For REST catalogs, the outermost table identifier is attached as an
optional request
+ * header. A context which already carries the header is returned
unchanged so nested
+ * dependencies preserve the original table.
+ */
+ @Nullable
+ CatalogContext dependencyReadContext() {
+ if (identifier == null || catalogContext == null) {
+ return catalogContext;
+ }
+
+ boolean restCatalog =
+ catalogLoader instanceof RESTCatalogLoader
+ || RESTCatalogFactory.IDENTIFIER.equals(
+ catalogContext.options().get(METASTORE));
+ if (!restCatalog) {
+ return catalogContext;
+ }
+
+ Options options = catalogContext.options();
+ if (options.containsKey(READ_VIA_OPTION)) {
+ return catalogContext;
+ }
+
+ Options dependencyOptions = new Options(options.toMap());
+ dependencyOptions.set(METASTORE, RESTCatalogFactory.IDENTIFIER);
+ dependencyOptions.set(
+ READ_VIA_OPTION,
RESTUtil.encodeString(JsonSerdeUtil.toFlatJson(identifier)));
+ return CatalogContext.create(
+ dependencyOptions,
+ catalogContext.hadoopConf(),
+ catalogContext.preferIO(),
+ catalogContext.fallbackIO());
+ }
+
public CatalogEnvironment copy(Identifier identifier) {
return new CatalogEnvironment(
identifier,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index ff3736d889..2b91f1311b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -49,6 +49,8 @@ import
org.apache.paimon.rest.exceptions.NotAuthorizedException;
import org.apache.paimon.rest.exceptions.NotImplementedException;
import org.apache.paimon.rest.responses.ConfigResponse;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.BlobDescriptorReaderFactory;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.format.FormatTablePartitionManager;
import org.apache.paimon.types.DataTypes;
@@ -73,6 +75,7 @@ import java.util.UUID;
import static org.apache.paimon.catalog.Catalog.TABLE_DEFAULT_OPTION_PREFIX;
import static org.apache.paimon.rest.RESTApi.HEADER_PREFIX;
+import static org.apache.paimon.rest.RESTApi.READ_VIA_HEADER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -420,6 +423,37 @@ class MockRESTCatalogTest extends RESTCatalogTest {
checkHeader(customHeaderName, customHeaderValue);
}
+ @Test
+ void testReadViaHeaderOnDependencyTableAndDataTokenRequests() throws
Exception {
+ Identifier root = Identifier.create("db", "root");
+ Identifier target = Identifier.create("db", "target");
+ RESTCatalog restCatalog = initCatalog(true);
+ restCatalog.createDatabase(target.getDatabaseName(), true);
+ restCatalog.createTable(target, DEFAULT_TABLE_SCHEMA, false);
+ restCatalog.createTable(
+ root,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(
+ CoreOptions.BLOB_DESCRIPTOR_SOURCE_TABLE.key(),
+ target.getFullName())
+ .build(),
+ false);
+ FileStoreTable rootTable = (FileStoreTable) restCatalog.getTable(root);
+
+ restCatalogServer.clearReceivedHeaders();
+ BlobDescriptorReaderFactory.create(rootTable);
+
+ String readVia = RESTUtil.encodeString(JsonSerdeUtil.toFlatJson(root));
+ ResourcePaths resourcePaths =
+
ResourcePaths.forCatalogProperties(restCatalog.api().options());
+ assertReadViaHeader(
+ resourcePaths.table(target.getDatabaseName(),
target.getObjectName()), readVia);
+ assertReadViaHeader(
+ resourcePaths.tableToken(target.getDatabaseName(),
target.getObjectName()),
+ readVia);
+ }
+
@Test
void testCreateFormatTableWhenEnableDataToken() throws Exception {
RESTCatalog restCatalog = initCatalog(true);
@@ -495,6 +529,15 @@ class MockRESTCatalogTest extends RESTCatalogTest {
assert foundCustomHeader : "Header was not found in any request";
}
+ private void assertReadViaHeader(String resourcePath, String readVia) {
+ assertThat(restCatalogServer.getReceivedHeaders(resourcePath))
+ .singleElement()
+ .satisfies(
+ headers ->
+ assertThat(headers)
+
.containsEntry(READ_VIA_HEADER.toLowerCase(), readVia));
+ }
+
private void testDlfAuth(RESTCatalog restCatalog) throws Exception {
String databaseName = "db1";
restCatalog.createDatabase(databaseName, true);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 6796e1ddf7..b6ae35561f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -226,6 +226,7 @@ public class RESTCatalogServer {
private final ResourcePaths resourcePaths;
private final List<Map<String, String>> receivedHeaders = new
ArrayList<>();
+ private final Map<String, List<Map<String, String>>> receivedHeadersByPath
= new HashMap<>();
private volatile boolean partitionListingSupported = true;
@@ -362,6 +363,9 @@ public class RESTCatalogServer {
receivedHeaders.add(new HashMap<>(headers));
String[] paths = request.getPath().split("\\?");
String resourcePath = paths[0];
+ receivedHeadersByPath
+ .computeIfAbsent(resourcePath, ignored -> new
ArrayList<>())
+ .add(new HashMap<>(headers));
Map<String, String> parameters =
paths.length == 2 ? getParameters(paths[1]) :
Collections.emptyMap();
String data = request.getBody().readUtf8();
@@ -3102,7 +3106,12 @@ public class RESTCatalogServer {
return receivedHeaders;
}
+ public List<Map<String, String>> getReceivedHeaders(String resourcePath) {
+ return receivedHeadersByPath.getOrDefault(resourcePath,
Collections.emptyList());
+ }
+
public void clearReceivedHeaders() {
receivedHeaders.clear();
+ receivedHeadersByPath.clear();
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
new file mode 100644
index 0000000000..77cc90788e
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.RESTApi;
+import org.apache.paimon.rest.RESTCatalogFactory;
+import org.apache.paimon.rest.RESTCatalogLoader;
+import org.apache.paimon.rest.RESTUtil;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.apache.paimon.options.CatalogOptions.METASTORE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link CatalogEnvironment}. */
+class CatalogEnvironmentTest {
+
+ private static final String READ_VIA_OPTION = RESTApi.HEADER_PREFIX +
RESTApi.READ_VIA_HEADER;
+
+ @Test
+ void testDependencyReadContextForRestCatalog() {
+ Identifier root = Identifier.create("db", "root$branch_dev");
+ Options options = new Options();
+ options.set("other-option", "value");
+ CatalogContext context = CatalogContext.create(options);
+ CatalogEnvironment environment = restEnvironment(root, context);
+
+ CatalogContext dependencyContext = environment.dependencyReadContext();
+
+ assertThat(dependencyContext).isNotSameAs(context);
+ assertThat(context.options().containsKey(READ_VIA_OPTION)).isFalse();
+ assertThat(dependencyContext.options().get(METASTORE))
+ .isEqualTo(RESTCatalogFactory.IDENTIFIER);
+
assertThat(dependencyContext.options().get("other-option")).isEqualTo("value");
+ Identifier readVia =
+ JsonSerdeUtil.fromJson(
+
RESTUtil.decodeString(dependencyContext.options().get(READ_VIA_OPTION)),
+ Identifier.class);
+ assertThat(readVia).isEqualTo(root);
+ }
+
+ @Test
+ void testDependencyReadContextPreservesOutermostTable() {
+ Identifier outermost = Identifier.create("db", "outermost");
+ Options options = new Options();
+ options.set(READ_VIA_OPTION,
RESTUtil.encodeString(JsonSerdeUtil.toFlatJson(outermost)));
+ CatalogContext context = CatalogContext.create(options);
+ CatalogEnvironment environment =
+ restEnvironment(Identifier.create("db", "intermediate"),
context);
+
+ assertThat(environment.dependencyReadContext()).isSameAs(context);
+ assertThat(context.options().get(READ_VIA_OPTION))
+
.isEqualTo(RESTUtil.encodeString(JsonSerdeUtil.toFlatJson(outermost)));
+ }
+
+ @Test
+ void testDependencyReadContextDoesNotAffectOtherCatalogs() {
+ CatalogContext context = CatalogContext.create(new Options());
+ CatalogEnvironment environment = environment(Identifier.create("db",
"table"), context);
+
+ assertThat(environment.dependencyReadContext()).isSameAs(context);
+ assertThat(context.options().containsKey(READ_VIA_OPTION)).isFalse();
+ }
+
+ @Test
+ void testDependencyReadContextForExternalRestTable() {
+ Options options = new Options();
+ options.set(METASTORE, RESTCatalogFactory.IDENTIFIER);
+ CatalogContext context = CatalogContext.create(options);
+ CatalogEnvironment environment = environment(Identifier.create("db",
"external"), context);
+
+ assertThat(environment.dependencyReadContext()).isNotSameAs(context);
+ }
+
+ @Test
+ void testAppendTableUsesDependencyReadContext() {
+ CatalogEnvironment environment = mock(CatalogEnvironment.class);
+
when(environment.dependencyReadContext()).thenReturn(CatalogContext.create(new
Options()));
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(new DataField(0, "id",
DataTypes.INT())),
+ 0,
+ Collections.emptyList(),
+ Collections.emptyList(),
+
Collections.singletonMap(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"),
+ null);
+ AppendOnlyFileStoreTable table =
+ new AppendOnlyFileStoreTable(
+ mock(FileIO.class), new Path("file:/tmp/table"),
schema, environment);
+
+ table.newRead();
+
+ verify(environment).dependencyReadContext();
+ }
+
+ private static CatalogEnvironment environment(
+ Identifier identifier, CatalogContext catalogContext) {
+ return new CatalogEnvironment(
+ identifier, null, null, null, null, catalogContext, false,
false);
+ }
+
+ private static CatalogEnvironment restEnvironment(
+ Identifier identifier, CatalogContext catalogContext) {
+ return new CatalogEnvironment(
+ identifier,
+ null,
+ new RESTCatalogLoader(catalogContext),
+ null,
+ null,
+ catalogContext,
+ false,
+ false);
+ }
+}
