[rest] Propagate dependency read context in headers

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -150,6 +150,15 @@
 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";
 

paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java

@@ -131,7 +131,7 @@ public InnerTableRead newRead() {
                 ? new DataEvolutionTableRead(
                         providerFactories,
                         schema(),
-                        catalogEnvironment.catalogContext(),
+                        catalogEnvironment.dependencyReadContext(),
                         () -> new AppendTableRead(providerFactories, schema()))
                 : new AppendTableRead(providerFactories, schema());
     }

paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReaderFactory.java

@@ -20,6 +20,7 @@
 
 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 static UriReaderFactory create(FileStoreTable table) {
     }
 
     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();

paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java

@@ -30,8 +30,14 @@
 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.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 CatalogContext catalogContext() {
         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,

paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java

@@ -49,6 +49,8 @@
 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 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,36 @@ void testBaseHeadersInRequests() throws Exception {
         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));
+        assertThat(restCatalogServer.getReceivedHeaders())
+                .hasSizeGreaterThanOrEqualTo(2)
+                .allSatisfy(
+                        headers ->
+                                assertThat(headers)
+                                        .containsEntry(READ_VIA_HEADER.toLowerCase(), readVia));
+    }
+
     @Test
     void testCreateFormatTableWhenEnableDataToken() throws Exception {
         RESTCatalog restCatalog = initCatalog(true);

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);
+    }
+}