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 e8e1ff96c87b..12a368612c81 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 @@ 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. + * + *

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 4d35147c17a4..f5f59ee70c88 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 InnerTableRead newRead() { ? 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 a7c29ebfdbc8..bcbef2980451 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 @@ 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(); 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 cb40c4447e18..5f75b7d0e58d 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.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. + * + *

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 ff3736d88903..2b91f1311b41 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.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,37 @@ 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)); + 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 @@ private void checkHeader(String headerName, String headerValue) { 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 6796e1ddf79e..b6ae35561f15 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> receivedHeaders = new ArrayList<>(); + private final Map>> receivedHeadersByPath = new HashMap<>(); private volatile boolean partitionListingSupported = true; @@ -362,6 +363,9 @@ public MockResponse dispatch(RecordedRequest request) { 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 parameters = paths.length == 2 ? getParameters(paths[1]) : Collections.emptyMap(); String data = request.getBody().readUtf8(); @@ -3102,7 +3106,12 @@ public List> getReceivedHeaders() { return receivedHeaders; } + public List> 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 000000000000..77cc90788e47 --- /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); + } +}