From 4c27195882604187af62878cf49fe673b1418137 Mon Sep 17 00:00:00 2001 From: fanng Date: Sun, 16 Aug 2026 18:45:48 +0900 Subject: [PATCH 1/2] [fix](lance) Pass namespace-vended storage options through to the BE A Lance REST catalog discarded the storage options a namespace vended for a table, so any scan that relied on credential vending failed with: open Lance dataset failed: Failed to get AWS credentials: CredentialsNotLoaded("no providers in chain provided credentials") The options were re-encoded twice on the way to the BE, through a five-entry S3-only table used in both directions: namespace vends access_key_id -> FE LanceStorageOptions.forBackend aws_access_key_id -> AWS_ACCESS_KEY -> TFileScanRangeParams.properties -> BE kStorageKeys AWS_ACCESS_KEY -> aws_access_key_id -> lance-c That table was written to emit one canonical spelling, which is right for the outbound direction, but reading it backwards makes it a parser that only accepts the spelling it happens to emit. Anything else was dropped: credentials under any other accepted alias, and every non-S3 provider's keys, which left the catalog unable to use credential vending outside S3 at all. The failure was also split across the two halves - the FE passed the vended map to the Lance Java SDK untouched and resolved schema fine, so only the scan failed. The namespace specification describes storage_options as configuration "passed directly to Lance", so the protocol defines no key vocabulary and a client cannot assume one. Stop re-encoding them: - Add TFileScanRangeParams.lance_storage_options, carrying the options in Lance's own vocabulary. It is set at ScanNode level, like paimon_options, so credentials are not serialized once per fragment split. - The FE now builds a single option map for both readers, so the FE SDK and lance-c can no longer disagree about how a dataset is reached. - The BE hands that map to lance-c as it arrives. Two details worth calling out: Vended aliases are renamed to the spelling this class emits before merging. object_store resolves an alias and its canonical name to one config key and keeps only one of the two values, chosen by hash order, so letting both through would leave the effective credentials and addressing style up to chance. Only the aws_-prefixed aliases are renamed; bare names such as "token" mean different things across providers. Catalog options now use the unprefixed spelling. object_store accepts both, but it is also the field name the OpenDAL backend uses, and that one performs no alias normalization. Note for a rolling upgrade: an FE upgraded ahead of the BEs no longer puts vended credentials into TFileScanRangeParams.properties, so a REST catalog with no static credentials cannot be scanned until the BEs are upgraded too. Verified end to end against Apache Gravitino 1.3.0's lance-rest service, which vends the unprefixed spelling, using a catalog with no s3.access_key or s3.secret_key: full scan, predicate pushdown and IVF_FLAT vector search all return correct results where the scan previously failed. The docker stub only ever vended aws_-prefixed keys, so the alias path had no coverage. It now serves a second table under the unprefixed spelling, and LanceStorageOptionsTest covers the merge directly - LanceStorageOptions had no test of its own before. Claude-Session: https://claude.ai/code/session_01M3mYXBKShBonG6Lg3br4Ld --- be/src/format_v2/table/lance_reader.cpp | 36 +-- .../docker-compose/iceberg/iceberg.yaml.tpl | 5 +- .../iceberg/scripts/lance_rest_server.py | 54 ++++- .../lance/LanceExternalCatalog.java | 33 +-- .../datasource/lance/LanceMetadataLoader.java | 28 +-- .../datasource/lance/LanceStorageOptions.java | 131 +++++++++-- .../datasource/lance/LanceTableMetadata.java | 11 +- .../lance/source/LanceScanNode.java | 9 +- .../datasource/tvf/source/TVFScanNode.java | 15 ++ .../ExternalFileTableValuedFunction.java | 6 + .../datasource/LanceThriftContractTest.java | 43 ++++ .../lance/LanceFilesystemCatalogTest.java | 20 -- .../datasource/lance/LanceSnapshotTest.java | 4 +- .../lance/LanceStorageOptionsTest.java | 213 ++++++++++++++++++ gensrc/thrift/PlanNodes.thrift | 6 + .../lance/test_lance_rest_catalog.out | 4 + .../lance/test_lance_rest_catalog.groovy | 8 + 17 files changed, 500 insertions(+), 126 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java diff --git a/be/src/format_v2/table/lance_reader.cpp b/be/src/format_v2/table/lance_reader.cpp index 142c35ac4d42bb..24e8f77c2b950f 100644 --- a/be/src/format_v2/table/lance_reader.cpp +++ b/be/src/format_v2/table/lance_reader.cpp @@ -894,36 +894,22 @@ Status LanceTableReader::_fill_block_from_arrow(LanceBatch* batch, Block* block, return Status::OK(); } +// The FE sends these already in Lance's own vocabulary, merged from the catalog properties and +// from whatever the namespace vended. Re-encoding them here would drop every option this list did +// not anticipate, so they are handed to lance-c as they arrive. std::vector LanceTableReader::_storage_options( const TFileScanRangeParams* scan_params) { - if (scan_params == nullptr || !scan_params->__isset.properties) { + if (scan_params == nullptr || !scan_params->__isset.lance_storage_options) { return {}; } - static constexpr std::array, 5> kStorageKeys = { - {{"AWS_ACCESS_KEY", "aws_access_key_id"}, - {"AWS_SECRET_KEY", "aws_secret_access_key"}, - {"AWS_TOKEN", "aws_session_token"}, - {"AWS_ENDPOINT", "aws_endpoint"}, - {"AWS_REGION", "aws_region"}}}; std::vector options; - options.reserve(kStorageKeys.size() * 2); - for (const auto& [doris_key, lance_key] : kStorageKeys) { - const auto it = scan_params->properties.find(std::string(doris_key)); - if (it != scan_params->properties.end() && !it->second.empty()) { - options.emplace_back(lance_key); - options.emplace_back(it->second); - } - } - const auto endpoint = scan_params->properties.find("AWS_ENDPOINT"); - if (endpoint != scan_params->properties.end() && endpoint->second.rfind("http://", 0) == 0) { - options.emplace_back("allow_http"); - options.emplace_back("true"); - } - const auto path_style = scan_params->properties.find("use_path_style"); - if (path_style != scan_params->properties.end() && !path_style->second.empty()) { - const bool use_path_style = path_style->second == "true" || path_style->second == "1"; - options.emplace_back("aws_virtual_hosted_style_request"); - options.emplace_back(use_path_style ? "false" : "true"); + options.reserve(scan_params->lance_storage_options.size() * 2); + for (const auto& [key, value] : scan_params->lance_storage_options) { + if (value.empty()) { + continue; + } + options.emplace_back(key); + options.emplace_back(value); } return options; } diff --git a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl index 5fc6af63e774b2..90b20a444eacf4 100644 --- a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl +++ b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl @@ -118,7 +118,10 @@ services: - ./scripts/lance_rest_server.py:/opt/lance-rest/server.py:ro environment: LANCE_REST_BEARER_TOKEN: doris-lance-rest-test-token - LANCE_REST_TABLES_JSON: '{"all_types":"s3://warehouse/lance/all_types.lance"}' + LANCE_REST_TABLES_JSON: '{"all_types":"s3://warehouse/lance/all_types.lance","all_types_unprefixed":"s3://warehouse/lance/all_types.lance"}' + # all_types_unprefixed serves the same dataset but vends its credentials under the + # unprefixed object-store spelling, which is what real namespace servers emit. + LANCE_REST_UNPREFIXED_TABLES_JSON: '["all_types_unprefixed"]' LANCE_S3_ACCESS_KEY: admin LANCE_S3_SECRET_KEY: password LANCE_S3_REGION: us-east-1 diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py index 8aef278c024f6d..3f614a34c03e49 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py @@ -65,6 +65,49 @@ def _load_tables() -> dict[tuple[str, ...], str]: TABLES = _load_tables() +def _load_unprefixed_tables() -> set[tuple[str, ...]]: + """Tables whose vended credentials use the unprefixed object-store spelling. + + A namespace may spell credentials with any alias Lance accepts, and real servers do use the + unprefixed one, so at least one table has to exercise it. + """ + raw = os.environ.get("LANCE_REST_UNPREFIXED_TABLES_JSON", "[]") + identifiers = json.loads(raw) + if not isinstance(identifiers, list): + raise ValueError("LANCE_REST_UNPREFIXED_TABLES_JSON must be a JSON array") + return { + tuple(part for part in identifier.split(DELIMITER) if part) + for identifier in identifiers + } + + +UNPREFIXED_TABLES = _load_unprefixed_tables() + + +def _storage_options(identifier: tuple[str, ...]) -> dict[str, str]: + access_key = os.environ.get("LANCE_S3_ACCESS_KEY", "admin") + secret_key = os.environ.get("LANCE_S3_SECRET_KEY", "password") + region = os.environ.get("LANCE_S3_REGION", "us-east-1") + if identifier in UNPREFIXED_TABLES: + return { + "access_key_id": access_key, + "secret_access_key": secret_key, + "region": region, + "virtual_hosted_style_request": "false", + # Lance refreshes expiring credentials against the namespace itself, so a client has + # to carry this through rather than drop it. + "expires_at_millis": os.environ.get( + "LANCE_S3_EXPIRES_AT_MILLIS", "4102444800000" + ), + } + return { + "aws_access_key_id": access_key, + "aws_secret_access_key": secret_key, + "aws_region": region, + "aws_virtual_hosted_style_request": "false", + } + + def _decode_identifier(identifier: str) -> tuple[str, ...]: identifier = unquote(identifier) if identifier == DELIMITER: @@ -141,16 +184,7 @@ def do_POST(self) -> None: "namespace": list(identifier[:-1]), "location": table_uri, "table_uri": table_uri, - "storage_options": { - "aws_access_key_id": os.environ.get( - "LANCE_S3_ACCESS_KEY", "admin" - ), - "aws_secret_access_key": os.environ.get( - "LANCE_S3_SECRET_KEY", "password" - ), - "aws_region": os.environ.get("LANCE_S3_REGION", "us-east-1"), - "aws_virtual_hosted_style_request": "false", - }, + "storage_options": _storage_options(identifier), "managed_versioning": False, "is_only_declared": False, }, diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java index 4339a863993a55..b94b3d4c403cff 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java @@ -46,7 +46,6 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; @@ -79,8 +78,7 @@ public class LanceExternalCatalog extends ExternalCatalog { private transient List parentNamespace = Collections.emptyList(); private transient String catalogType; private transient String rootDatabase; - private transient Map javaStorageOptions = Collections.emptyMap(); - private transient Map backendStorageOptions = Collections.emptyMap(); + private transient Map lanceStorageOptions = Collections.emptyMap(); private transient Object namespaceLock = new Object(); public LanceExternalCatalog(long catalogId, String name, String resource, Map props, @@ -98,11 +96,11 @@ protected void initLocalObjectsImpl() { rootDatabase = properties.getRootDatabase(); parentNamespace = LanceNamespaceName.parseParentNamespace( properties.getNamespaceParent(), properties.getNamespaceDelimiter()); - backendStorageOptions = catalogProperty.getBackendStorageProperties(); - javaStorageOptions = LanceStorageOptions.forJavaSdk(backendStorageOptions); + lanceStorageOptions = LanceStorageOptions.toLanceOptions( + catalogProperty.getBackendStorageProperties()); allocator = new RootAllocator(ALLOCATOR_LIMIT); - namespace = properties.createNamespace(allocator, javaStorageOptions); + namespace = properties.createNamespace(allocator, lanceStorageOptions); } catch (Exception e) { closeLanceObjects(); throw new RuntimeException("Failed to initialize Lance catalog '" + getName() @@ -120,7 +118,7 @@ public void checkWhenCreating() throws DdlException { } AbstractLanceProperties properties = getLanceProperties(); - Map storageOptions = LanceStorageOptions.forJavaSdk( + Map storageOptions = LanceStorageOptions.toLanceOptions( catalogProperty.getBackendStorageProperties()); List parent = LanceNamespaceName.parseParentNamespace( properties.getNamespaceParent(), properties.getNamespaceDelimiter()); @@ -285,12 +283,10 @@ public LanceTableMetadata loadTableMetadata(String dbName, String tableName, throw new RuntimeException("Lance namespace returned no table URI for " + dbName + "." + tableName); } - Map storageOptions = new HashMap<>(javaStorageOptions); - if (table.getStorageOptions() != null) { - storageOptions.putAll(table.getStorageOptions()); - } - Map tableBackendStorageOptions = LanceStorageOptions.forBackend( - backendStorageOptions, table.getStorageOptions()); + // One option map serves both readers: the FE opens the dataset through the Lance Java SDK + // and the BE through lance-c, so neither can end up with credentials the other lacks. + Map storageOptions = LanceStorageOptions.mergeVended( + lanceStorageOptions, table.getStorageOptions()); try { if (tableSnapshot.isPresent()) { TableSnapshot snapshot = tableSnapshot.get(); @@ -306,11 +302,9 @@ public LanceTableMetadata loadTableMetadata(String dbName, String tableName, version = LanceSnapshotResolver.getVersionAtOrBefore( datasetUri, storageOptions, timestamp, allocator); } - return LanceMetadataLoader.loadVersion(datasetUri, storageOptions, - tableBackendStorageOptions, version, allocator); + return LanceMetadataLoader.loadVersion(datasetUri, storageOptions, version, allocator); } - return LanceMetadataLoader.loadLatest(datasetUri, storageOptions, - tableBackendStorageOptions, allocator); + return LanceMetadataLoader.loadLatest(datasetUri, storageOptions, allocator); } catch (Exception e) { throw new RuntimeException("Failed to load Lance table metadata for " + dbName + "." + tableName + ": " + sanitizedRootCauseMessage(e), safeCause(e)); @@ -348,11 +342,6 @@ private List buildFullNamespace(List relativeNamespace) { return result; } - public Map getBackendStorageOptions() { - makeSureInitialized(); - return backendStorageOptions; - } - public String getLanceCatalogType() { makeSureInitialized(); return catalogType; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataLoader.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataLoader.java index a2bc6d9a60de37..226cfd050fde22 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataLoader.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataLoader.java @@ -41,11 +41,10 @@ private LanceMetadataLoader() { * Lance dataset through an S3 TVF. */ public static LanceTableMetadata loadLatestForTvf( - String datasetUri, Map backendStorageOptions) + String datasetUri, Map backendProperties) throws Exception { try (BufferAllocator allocator = new RootAllocator(ALLOCATOR_LIMIT)) { - return loadLatest(datasetUri, LanceStorageOptions.forJavaSdk(backendStorageOptions), - backendStorageOptions, allocator); + return loadLatest(datasetUri, LanceStorageOptions.toLanceOptions(backendProperties), allocator); } } @@ -57,10 +56,9 @@ public static LanceTableMetadata loadLatestForTvf( * time-travel version is requested. Schema, version, and fragments are read from the same * opened dataset snapshot. */ - public static LanceTableMetadata loadLatest(String datasetUri, Map javaStorageOptions, - Map backendStorageOptions, BufferAllocator allocator) throws Exception { - return loadInternal( - datasetUri, javaStorageOptions, backendStorageOptions, OptionalLong.empty(), allocator); + public static LanceTableMetadata loadLatest(String datasetUri, Map lanceStorageOptions, + BufferAllocator allocator) throws Exception { + return loadInternal(datasetUri, lanceStorageOptions, OptionalLong.empty(), allocator); } /** @@ -70,18 +68,16 @@ public static LanceTableMetadata loadLatest(String datasetUri, Map javaStorageOptions, - Map backendStorageOptions, long version, BufferAllocator allocator) throws Exception { - return loadInternal( - datasetUri, javaStorageOptions, backendStorageOptions, OptionalLong.of(version), allocator); + public static LanceTableMetadata loadVersion(String datasetUri, Map lanceStorageOptions, + long version, BufferAllocator allocator) throws Exception { + return loadInternal(datasetUri, lanceStorageOptions, OptionalLong.of(version), allocator); } /** Shared implementation for the latest-version and explicit-version public entry points. */ - private static LanceTableMetadata loadInternal(String datasetUri, Map javaStorageOptions, - Map backendStorageOptions, OptionalLong version, - BufferAllocator allocator) throws Exception { + private static LanceTableMetadata loadInternal(String datasetUri, Map lanceStorageOptions, + OptionalLong version, BufferAllocator allocator) throws Exception { try (Dataset dataset = Dataset.open().allocator(allocator).uri(datasetUri) - .readOptions(LanceReadOptions.build(javaStorageOptions, version)).build()) { + .readOptions(LanceReadOptions.build(lanceStorageOptions, version)).build()) { long resolvedVersion = dataset.version(); List fragments = new ArrayList<>(); for (Fragment fragment : dataset.getFragments()) { @@ -90,7 +86,7 @@ private static LanceTableMetadata loadInternal(String datasetUri, MapBoth the FE, which opens the dataset through the Lance Java SDK, and the BE, which opens it + * through lance-c, consume the map produced here, so the two cannot disagree about how a dataset + * is accessed. + * + *

Options vended by a namespace are merged in as they arrive. The Lance Namespace specification + * describes {@code storage_options} as configuration "passed directly to Lance", so the protocol + * defines no key vocabulary of its own and a client cannot assume one. Re-encoding those options + * into a fixed set of names would silently drop everything outside it, including credentials + * spelled with a different accepted alias and every non-S3 provider's keys. + */ public final class LanceStorageOptions { + private static final Logger LOG = LogManager.getLogger(LanceStorageOptions.class); + + /** + * Doris backend property to Lance object-store option. + * + *

Lance reaches S3 through object_store, which accepts both {@code access_key_id} and + * {@code aws_access_key_id}. The unprefixed spelling is chosen because it is also the field + * name used by the OpenDAL backend, which performs no alias normalization at all, so these + * options stay correct if that backend is ever selected. + */ private static final Map S3_KEYS = new HashMap<>(); static { - S3_KEYS.put("AWS_ACCESS_KEY", "aws_access_key_id"); - S3_KEYS.put("AWS_SECRET_KEY", "aws_secret_access_key"); - S3_KEYS.put("AWS_TOKEN", "aws_session_token"); - S3_KEYS.put("AWS_ENDPOINT", "aws_endpoint"); - S3_KEYS.put("AWS_REGION", "aws_region"); + S3_KEYS.put("AWS_ACCESS_KEY", "access_key_id"); + S3_KEYS.put("AWS_SECRET_KEY", "secret_access_key"); + S3_KEYS.put("AWS_TOKEN", "session_token"); + S3_KEYS.put("AWS_ENDPOINT", "endpoint"); + S3_KEYS.put("AWS_REGION", "region"); } + /** + * The {@code aws_}-prefixed aliases of the options above, mapped to the spelling this class + * emits. + * + *

object_store resolves an alias and its canonical name to one config key and keeps only one + * of the two values, chosen by hash order. So a namespace vending {@code aws_endpoint} while + * the catalog contributes {@code endpoint} does not override it - the two survive as separate + * entries and Lance later picks between them unpredictably. Renaming the vended aliases first + * makes the merge below decide, every time. + * + *

Only the prefixed aliases are renamed. Bare names such as {@code token} are ambiguous + * across providers - object_store reads it as an S3 session token but as a bearer token for + * Azure - and this class does not know which provider a dataset uses. + */ + private static final Map VENDED_ALIASES = ImmutableMap.builder() + .put("aws_access_key_id", "access_key_id") + .put("aws_secret_access_key", "secret_access_key") + .put("aws_session_token", "session_token") + .put("aws_token", "session_token") + .put("aws_endpoint", "endpoint") + .put("aws_endpoint_url", "endpoint") + .put("aws_region", "region") + .put("aws_virtual_hosted_style_request", "virtual_hosted_style_request") + .put("aws_allow_http", "allow_http") + .build(); + + /** + * Options a namespace may not override, because they decide which data is read rather than how + * it is accessed. Lance protects the same keys in the options it accepts from a namespace. + */ + private static final Set PROTECTED_KEYS = ImmutableSet.of( + "bucket", "aws_bucket", "aws_bucket_name", "bucket_name", "root"); + private LanceStorageOptions() { } - public static Map forJavaSdk(Map backendProperties) { + /** Converts normalized Doris storage properties to Lance object-store options. */ + public static Map toLanceOptions(Map backendProperties) { Map result = new HashMap<>(); S3_KEYS.forEach((dorisKey, lanceKey) -> putIfNotEmpty(result, lanceKey, backendProperties.get(dorisKey))); - String endpoint = backendProperties.get("AWS_ENDPOINT"); - if (endpoint != null && endpoint.startsWith("http://")) { - result.put("allow_http", "true"); - } String usePathStyle = backendProperties.get("use_path_style"); if (usePathStyle != null && !usePathStyle.isEmpty()) { - result.put("aws_virtual_hosted_style_request", + result.put("virtual_hosted_style_request", String.valueOf(!Boolean.parseBoolean(usePathStyle))); } - return result; + return withDerivedAllowHttp(result); } - /** Merge Lance storage options returned by a namespace into properties understood by Doris BE. */ - public static Map forBackend(Map staticBackendProperties, - Map lanceStorageOptions) { - Map result = new HashMap<>(staticBackendProperties); - if (lanceStorageOptions == null || lanceStorageOptions.isEmpty()) { + /** + * Merges the options a namespace vended for one table over the catalog's own options. + * + *

Options a namespace may not override are dropped; everything else replaces the catalog + * value, since the namespace decides how the table it just described is reached. + */ + public static Map mergeVended(Map lanceOptions, + Map vendedOptions) { + Map result = new HashMap<>(lanceOptions); + if (vendedOptions == null || vendedOptions.isEmpty()) { return result; } - S3_KEYS.forEach((dorisKey, lanceKey) -> putIfNotEmpty(result, dorisKey, - lanceStorageOptions.get(lanceKey))); + vendedOptions.forEach((key, value) -> { + if (key == null) { + return; + } + String lowerCased = key.toLowerCase(Locale.ROOT); + if (PROTECTED_KEYS.contains(lowerCased)) { + LOG.warn("Ignoring Lance storage option '{}' vended by the namespace because it " + + "would change which data is read", key); + return; + } + putIfNotEmpty(result, VENDED_ALIASES.getOrDefault(lowerCased, key), value); + }); + return withDerivedAllowHttp(result); + } - String virtualHostedStyle = lanceStorageOptions.get("aws_virtual_hosted_style_request"); - if (virtualHostedStyle != null && !virtualHostedStyle.isEmpty()) { - result.put("use_path_style", String.valueOf(!Boolean.parseBoolean(virtualHostedStyle))); + /** + * Allows plain HTTP when the endpoint in use asks for it. + * + *

Applied after merging, because a namespace can replace the endpoint the catalog was + * configured with - or supply the only one there is. + */ + private static Map withDerivedAllowHttp(Map options) { + String endpoint = options.get("endpoint"); + if (endpoint != null && endpoint.startsWith("http://")) { + options.putIfAbsent("allow_http", "true"); } - return result; + return options; } private static void putIfNotEmpty(Map target, String key, String value) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTableMetadata.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTableMetadata.java index 674b5b79d15be3..611f6db6855bea 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTableMetadata.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceTableMetadata.java @@ -30,15 +30,15 @@ public class LanceTableMetadata { private final long version; private final Schema schema; private final List fragments; - private final Map backendStorageOptions; + private final Map lanceStorageOptions; public LanceTableMetadata(String datasetUri, long version, Schema schema, - List fragments, Map backendStorageOptions) { + List fragments, Map lanceStorageOptions) { this.datasetUri = datasetUri; this.version = version; this.schema = schema; this.fragments = Collections.unmodifiableList(fragments); - this.backendStorageOptions = Collections.unmodifiableMap(new HashMap<>(backendStorageOptions)); + this.lanceStorageOptions = Collections.unmodifiableMap(new HashMap<>(lanceStorageOptions)); } public String getDatasetUri() { @@ -57,8 +57,9 @@ public List getFragments() { return fragments; } - public Map getBackendStorageOptions() { - return backendStorageOptions; + /** Lance object-store options for this dataset, understood as-is by both the FE SDK and lance-c. */ + public Map getLanceStorageOptions() { + return lanceStorageOptions; } public long getRowCount() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java index 4c433a30bce145..1fd71c58022f79 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java @@ -148,6 +148,11 @@ public void createScanRangeLocations() throws UserException { if (lanceSubstraitFilter.length > 0) { params.setLanceSubstraitFilter(ByteBuffer.wrap(lanceSubstraitFilter)); } + // Set at ScanNode level so credentials are not serialized once per fragment split. + Map lanceStorageOptions = plannedMetadata.getLanceStorageOptions(); + if (!lanceStorageOptions.isEmpty()) { + params.setLanceStorageOptions(lanceStorageOptions); + } } @Override @@ -249,7 +254,9 @@ protected TableIf getTargetTable() { @Override protected Map getLocationProperties() { - return plannedMetadata.getBackendStorageOptions(); + // lance-c reads the dataset itself and takes its configuration from lance_storage_options, + // so these serve only the shared file system layer and the file cache key. + return lanceTable.getCatalog().getCatalogProperty().getBackendStorageProperties(); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java index 3254b5235036a1..a3d6e2a1891e87 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java @@ -31,6 +31,7 @@ import org.apache.doris.datasource.FileSplit.FileSplitCreator; import org.apache.doris.datasource.FileSplitter; import org.apache.doris.datasource.TableFormatType; +import org.apache.doris.datasource.lance.LanceStorageOptions; import org.apache.doris.datasource.lance.LanceTableMetadata; import org.apache.doris.datasource.lance.source.LanceSplit; import org.apache.doris.planner.PlanNodeId; @@ -125,6 +126,20 @@ public Map getLocationProperties() { return tableValuedFunction.getBackendConnectProperties(); } + @Override + public void createScanRangeLocations() throws UserException { + super.createScanRangeLocations(); + if (tableValuedFunction.isLanceFormat()) { + // lance-c opens the dataset itself and needs the options in Lance's own vocabulary. + // Set at ScanNode level so credentials are not serialized once per fragment split. + Map lanceStorageOptions = LanceStorageOptions.toLanceOptions( + tableValuedFunction.getBackendConnectProperties()); + if (!lanceStorageOptions.isEmpty()) { + params.setLanceStorageOptions(lanceStorageOptions); + } + } + } + @Override public List getPathPartitionKeys() { return tableValuedFunction.getPathPartitionKeys(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index 8e06e8c3e3d3a3..a46a21dff28384 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -45,6 +45,7 @@ import org.apache.doris.common.util.S3Util; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.TableFormatType; +import org.apache.doris.datasource.lance.LanceStorageOptions; import org.apache.doris.datasource.lance.LanceTableMetadata; import org.apache.doris.datasource.lance.LanceTypeConverter; import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties; @@ -526,6 +527,11 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TExcepti Map beProperties = new HashMap<>(); beProperties.putAll(backendConnectProperties); fileScanRangeParams.setProperties(beProperties); + if (fileFormatProperties.getFileFormatType() == TFileFormatType.FORMAT_LANCE) { + // lance-c opens the dataset itself and needs the options in Lance's own vocabulary. + fileScanRangeParams.setLanceStorageOptions( + LanceStorageOptions.toLanceOptions(backendConnectProperties)); + } fileScanRangeParams.setFileAttributes(getFileAttributes()); ConnectContext ctx = ConnectContext.get(); fileScanRangeParams.setLoadId(ctx.queryId()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java index dc21fafa3148ab..c541af2504918c 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java @@ -18,6 +18,7 @@ package org.apache.doris.datasource; import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TFileScanRangeParams; import org.apache.doris.thrift.TLanceFileDesc; import org.apache.doris.thrift.TTableFormatFileDesc; @@ -28,6 +29,8 @@ import org.junit.Test; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; public class LanceThriftContractTest { @@ -77,4 +80,44 @@ public void testLanceDescriptorWithoutLimit() throws Exception { // A scan without a pushable LIMIT must leave the field unset so the BE reads all rows. Assert.assertFalse(restored.getLanceParams().isSetLimit()); } + + @Test + public void testLanceStorageOptionsSurviveRoundTripUntouched() throws Exception { + Map storageOptions = new HashMap<>(); + storageOptions.put("access_key_id", "ak"); + storageOptions.put("secret_access_key", "sk"); + storageOptions.put("endpoint", "http://127.0.0.1:9000"); + storageOptions.put("expires_at_millis", "1760000000000"); + storageOptions.put("azure_storage_sas_token", "sas"); + + TFileScanRangeParams source = new TFileScanRangeParams() + .setFormatType(TFileFormatType.FORMAT_LANCE) + .setLanceStorageOptions(storageOptions); + + TSerializer serializer = new TSerializer(new TCompactProtocol.Factory()); + byte[] bytes = serializer.serialize(source); + + TFileScanRangeParams restored = new TFileScanRangeParams(); + new TDeserializer(new TCompactProtocol.Factory()).deserialize(restored, bytes); + + // Whatever the namespace vended has to reach lance-c unchanged, including keys Doris + // itself assigns no meaning to. + Assert.assertTrue(restored.isSetLanceStorageOptions()); + Assert.assertEquals(storageOptions, restored.getLanceStorageOptions()); + } + + @Test + public void testLanceStorageOptionsAreOptional() throws Exception { + TFileScanRangeParams source = new TFileScanRangeParams() + .setFormatType(TFileFormatType.FORMAT_LANCE); + + TSerializer serializer = new TSerializer(new TCompactProtocol.Factory()); + byte[] bytes = serializer.serialize(source); + + TFileScanRangeParams restored = new TFileScanRangeParams(); + new TDeserializer(new TCompactProtocol.Factory()).deserialize(restored, bytes); + + // A local dataset needs no storage configuration at all. + Assert.assertFalse(restored.isSetLanceStorageOptions()); + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceFilesystemCatalogTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceFilesystemCatalogTest.java index 6b4a9f9c2bca7d..ce42386e2e603b 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceFilesystemCatalogTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceFilesystemCatalogTest.java @@ -21,29 +21,9 @@ import org.junit.Test; import java.util.Collections; -import java.util.HashMap; -import java.util.Map; public class LanceFilesystemCatalogTest { - @Test - public void testMinioStorageOptionMapping() { - Map backendProperties = new HashMap<>(); - backendProperties.put("AWS_ACCESS_KEY", "ak"); - backendProperties.put("AWS_SECRET_KEY", "sk"); - backendProperties.put("AWS_ENDPOINT", "http://minio:9000"); - backendProperties.put("AWS_REGION", "us-east-1"); - backendProperties.put("use_path_style", "true"); - - Map options = LanceStorageOptions.forJavaSdk(backendProperties); - Assert.assertEquals("ak", options.get("aws_access_key_id")); - Assert.assertEquals("sk", options.get("aws_secret_access_key")); - Assert.assertEquals("http://minio:9000", options.get("aws_endpoint")); - Assert.assertEquals("us-east-1", options.get("aws_region")); - Assert.assertEquals("true", options.get("allow_http")); - Assert.assertEquals("false", options.get("aws_virtual_hosted_style_request")); - } - @Test public void testNamespaceNameRoundTrip() throws Exception { Assert.assertEquals(Collections.emptyList(), LanceNamespaceName.dorisDatabaseNameToNamespace( diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceSnapshotTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceSnapshotTest.java index a48afa669ad771..a9723863eef4ca 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceSnapshotTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceSnapshotTest.java @@ -78,7 +78,7 @@ public void testBoundSnapshotCarriesItsOwnSchema() { Assertions.assertEquals(10, version10.getMetadata().getVersion()); Assertions.assertEquals(10, version10.getMetadata().getFragments().get(0).getId()); Assertions.assertEquals("http://minio:9000", - version10.getMetadata().getBackendStorageOptions().get("s3.endpoint")); + version10.getMetadata().getLanceStorageOptions().get("endpoint")); Assertions.assertTrue(version10.isSameSnapshot(new LanceMvccSnapshot(metadata(10, Field.nullable("value", new ArrowType.Int(32, true)))))); Assertions.assertFalse(version10.isSameSnapshot(new LanceMvccSnapshot(floatMetadata))); @@ -89,6 +89,6 @@ private static LanceTableMetadata metadata(long version, Field field) { return new LanceTableMetadata("s3://bucket/table.lance", version, new Schema(Collections.singletonList(field)), Collections.singletonList(new LanceTableMetadata.LanceFragmentInfo(version, 1, 1)), - Collections.singletonMap("s3.endpoint", "http://minio:9000")); + Collections.singletonMap("endpoint", "http://minio:9000")); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java new file mode 100644 index 00000000000000..ab12f6dac4d606 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java @@ -0,0 +1,213 @@ +// 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.lance; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +public class LanceStorageOptionsTest { + + private static Map minioCatalogProperties() { + Map backendProperties = new HashMap<>(); + backendProperties.put("AWS_ACCESS_KEY", "ak"); + backendProperties.put("AWS_SECRET_KEY", "sk"); + backendProperties.put("AWS_ENDPOINT", "http://minio:9000"); + backendProperties.put("AWS_REGION", "us-east-1"); + backendProperties.put("use_path_style", "true"); + return backendProperties; + } + + @Test + public void testMinioStorageOptionMapping() { + Map backendProperties = minioCatalogProperties(); + backendProperties.put("AWS_TOKEN", "token"); + + Map options = LanceStorageOptions.toLanceOptions(backendProperties); + Assertions.assertEquals("ak", options.get("access_key_id")); + Assertions.assertEquals("sk", options.get("secret_access_key")); + Assertions.assertEquals("token", options.get("session_token")); + Assertions.assertEquals("http://minio:9000", options.get("endpoint")); + Assertions.assertEquals("us-east-1", options.get("region")); + Assertions.assertEquals("true", options.get("allow_http")); + Assertions.assertEquals("false", options.get("virtual_hosted_style_request")); + } + + /** + * Lance reaches object storage through two backends. Only the unprefixed spellings are accepted + * by both, so emitting the prefixed ones would drop every credential on the backend that does + * not normalize aliases. + */ + @Test + public void testEmittedOptionsUseTheSpellingBothLanceBackendsAccept() { + Map options = LanceStorageOptions.toLanceOptions(minioCatalogProperties()); + Assertions.assertNull(options.get("aws_access_key_id")); + Assertions.assertNull(options.get("aws_secret_access_key")); + Assertions.assertNull(options.get("aws_endpoint")); + Assertions.assertNull(options.get("aws_region")); + } + + @Test + public void testOptionalPropertiesAreOmittedRatherThanEmpty() { + Map backendProperties = new HashMap<>(); + backendProperties.put("AWS_ACCESS_KEY", "ak"); + backendProperties.put("AWS_SECRET_KEY", ""); + backendProperties.put("AWS_ENDPOINT", "https://s3.amazonaws.com"); + + Map options = LanceStorageOptions.toLanceOptions(backendProperties); + Assertions.assertEquals("ak", options.get("access_key_id")); + Assertions.assertNull(options.get("secret_access_key")); + Assertions.assertNull(options.get("session_token")); + // allow_http only makes sense for a plain-HTTP endpoint. + Assertions.assertNull(options.get("allow_http")); + Assertions.assertNull(options.get("virtual_hosted_style_request")); + } + + /** + * object_store folds an alias and its canonical name onto one config key and keeps just one of + * the values, picked by hash order. Letting both spellings through would therefore leave the + * effective credentials and addressing style up to chance, so a vended alias has to replace the + * catalog's entry rather than sit beside it. + */ + @Test + public void testVendedPrefixedAliasReplacesTheCatalogEntry() { + Map vended = new HashMap<>(); + vended.put("aws_access_key_id", "vended-ak"); + vended.put("aws_secret_access_key", "vended-sk"); + vended.put("aws_endpoint", "http://127.0.0.1:9000"); + vended.put("aws_region", "eu-west-1"); + vended.put("aws_virtual_hosted_style_request", "true"); + vended.put("aws_session_token", "vended-token"); + + Map merged = LanceStorageOptions.mergeVended( + LanceStorageOptions.toLanceOptions(minioCatalogProperties()), vended); + + Assertions.assertEquals("vended-ak", merged.get("access_key_id")); + Assertions.assertEquals("vended-sk", merged.get("secret_access_key")); + Assertions.assertEquals("http://127.0.0.1:9000", merged.get("endpoint")); + Assertions.assertEquals("eu-west-1", merged.get("region")); + Assertions.assertEquals("true", merged.get("virtual_hosted_style_request")); + Assertions.assertEquals("vended-token", merged.get("session_token")); + + // No prefixed duplicate may survive alongside the value it replaced. + for (String alias : new String[] {"aws_access_key_id", "aws_secret_access_key", + "aws_endpoint", "aws_region", "aws_virtual_hosted_style_request", + "aws_session_token"}) { + Assertions.assertNull(merged.get(alias), alias + " must not survive the merge"); + } + } + + @Test + public void testVendedUnprefixedOptionsReplaceTheCatalogEntry() { + Map vended = new HashMap<>(); + vended.put("access_key_id", "vended-ak"); + vended.put("endpoint", "http://127.0.0.1:9000"); + + Map merged = LanceStorageOptions.mergeVended( + LanceStorageOptions.toLanceOptions(minioCatalogProperties()), vended); + Assertions.assertEquals("vended-ak", merged.get("access_key_id")); + Assertions.assertEquals("http://127.0.0.1:9000", merged.get("endpoint")); + Assertions.assertEquals("sk", merged.get("secret_access_key")); + } + + /** + * A namespace can supply the only endpoint there is, or replace the catalog's, so whether plain + * HTTP is allowed has to follow the endpoint that ends up in use. + */ + @Test + public void testAllowHttpFollowsTheEndpointActuallyUsed() { + Map catalogWithoutEndpoint = new HashMap<>(); + catalogWithoutEndpoint.put("AWS_ACCESS_KEY", "ak"); + + Map vended = new HashMap<>(); + vended.put("endpoint", "http://127.0.0.1:9000"); + + Map merged = LanceStorageOptions.mergeVended( + LanceStorageOptions.toLanceOptions(catalogWithoutEndpoint), vended); + Assertions.assertEquals("true", merged.get("allow_http")); + + // An explicitly vended value is respected rather than re-derived. + Map vendedWithFlag = new HashMap<>(); + vendedWithFlag.put("endpoint", "http://127.0.0.1:9000"); + vendedWithFlag.put("allow_http", "false"); + Assertions.assertEquals("false", LanceStorageOptions.mergeVended( + Collections.emptyMap(), vendedWithFlag).get("allow_http")); + } + + /** + * Lance recognizes this key on its namespace-backed refresh path. lance-c opens datasets with + * static options, so it has no effect on the BE today, but the option map is meant to reach + * Lance as the namespace wrote it. + */ + @Test + public void testUnrecognizedVendedOptionsArePassedThrough() { + Map vended = new HashMap<>(); + vended.put("access_key_id", "vended-ak"); + vended.put("expires_at_millis", "1760000000000"); + vended.put("refresh_offset_millis", "60000"); + + Map merged = LanceStorageOptions.mergeVended(Collections.emptyMap(), vended); + Assertions.assertEquals("1760000000000", merged.get("expires_at_millis")); + Assertions.assertEquals("60000", merged.get("refresh_offset_millis")); + } + + /** Options for other providers must survive too, or the catalog only ever works on S3. */ + @Test + public void testNonS3VendedOptionsArePassedThrough() { + Map vended = new HashMap<>(); + vended.put("azure_storage_sas_token", "sas"); + vended.put("google_storage_token", "gcp-token"); + + Map merged = LanceStorageOptions.mergeVended(Collections.emptyMap(), vended); + Assertions.assertEquals("sas", merged.get("azure_storage_sas_token")); + Assertions.assertEquals("gcp-token", merged.get("google_storage_token")); + } + + @Test + public void testVendedOptionsCannotRedirectWhichDataIsRead() { + Map vended = new HashMap<>(); + vended.put("access_key_id", "vended-ak"); + vended.put("bucket", "attacker-bucket"); + vended.put("aws_bucket_name", "attacker-bucket"); + vended.put("ROOT", "/elsewhere"); + + Map merged = LanceStorageOptions.mergeVended(Collections.emptyMap(), vended); + Assertions.assertEquals("vended-ak", merged.get("access_key_id")); + Assertions.assertNull(merged.get("bucket")); + Assertions.assertNull(merged.get("aws_bucket_name")); + Assertions.assertNull(merged.get("ROOT")); + } + + @Test + public void testAbsentVendedOptionsLeaveCatalogOptionsIntact() { + Map catalogOptions = LanceStorageOptions.toLanceOptions(minioCatalogProperties()); + Assertions.assertEquals(catalogOptions, + LanceStorageOptions.mergeVended(catalogOptions, null)); + Assertions.assertEquals(catalogOptions, + LanceStorageOptions.mergeVended(catalogOptions, new HashMap<>())); + + Map vended = new HashMap<>(); + vended.put("access_key_id", ""); + vended.put(null, "ignored"); + Assertions.assertEquals(catalogOptions, + LanceStorageOptions.mergeVended(catalogOptions, vended)); + } +} diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index d3f0583c2206a1..eecbc4690dde1f 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -635,6 +635,12 @@ struct TFileScanRangeParams { // Provider-independent search request. Set at ScanNode level so all ranges use the same logical // query. The first implementation uses one whole-dataset range for Lance vector search. 37: optional TExternalSearchRequest external_search_request + // Lance-native storage options, handed to lance-c untranslated. The namespace protocol treats + // storage_options as opaque configuration passed directly to Lance, so any key vocabulary the + // BE imposed here would drop options it does not happen to know - including credentials a + // namespace spelled with a different accepted alias, and every non-S3 provider's keys. + // Set at ScanNode level so credentials are not serialized once per fragment split. + 38: optional map lance_storage_options } struct TFileRangeDesc { diff --git a/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out b/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out index 62a570c8d16d63..cb8003936c2693 100644 --- a/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out +++ b/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out @@ -6,10 +6,14 @@ mysql -- !rest_tables -- all_types +all_types_unprefixed -- !rest_scan -- 12 12 1 12 78 +-- !rest_scan_unprefixed_credentials -- +12 12 1 12 78 + -- !rest_predicate_pushdown -- 8 9 diff --git a/regression-test/suites/external_table_p0/lance/test_lance_rest_catalog.groovy b/regression-test/suites/external_table_p0/lance/test_lance_rest_catalog.groovy index 82cd85858f82a6..75f2c100793ce6 100644 --- a/regression-test/suites/external_table_p0/lance/test_lance_rest_catalog.groovy +++ b/regression-test/suites/external_table_p0/lance/test_lance_rest_catalog.groovy @@ -70,6 +70,14 @@ suite("test_lance_rest_catalog", "p0,external") { FROM `${catalogName}`.`default`.`${tableName}` """ + // The same dataset, described by a namespace that spells the vended credentials without + // the aws_ prefix. Lance accepts either alias, so both have to reach the BE; a client that + // recognizes only one silently scans with no credentials at all. + qt_rest_scan_unprefixed_credentials """ + SELECT count(*), count(DISTINCT row_id), min(row_id), max(row_id), sum(row_id) + FROM `${catalogName}`.`default`.`all_types_unprefixed` + """ + String pushedQuery = """SELECT row_id FROM `${catalogName}`.`default`.`${tableName}` WHERE int32_col = 10 ORDER BY row_id""" explain { From 63a6dfcd293b21f88d2a0b79d41e8ef58091ddb9 Mon Sep 17 00:00:00 2001 From: fanng Date: Mon, 17 Aug 2026 11:54:10 +0900 Subject: [PATCH 2/2] [fix](lance) Collapse every accepted storage option alias onto one entry Review follow-up. The alias table only recognized the aws_-prefixed spellings, but object_store accepts four names for the endpoint and four for the session token. A namespace vending `endpoint_url`, or any spelling in a different case, still landed beside the catalog's `endpoint` as a second entry, and Lance picked between them by hash order - the same race the table was added to close, just moved to the spellings it missed. Recognize every accepted alias, keyed on the lower-cased name, and let a vended option displace whichever spelling the catalog used for it. `token` keeps the namespace's spelling: object_store reads it as an S3 session token but as a bearer token for Azure, so renaming it would corrupt the Azure reading; it is only used to decide which catalog entry it replaces. Also retract a derived allow_http when the namespace replaces a plain-HTTP endpoint with an HTTPS one. It was only ever added, so the merged options went on permitting plain HTTP for an endpoint that never asked for it. Two comments claimed Lance refreshes these credentials against the namespace. It does not on this path: lance-c opens datasets with a static option set, so `expires_at_millis` is carried but never acted on, and credentials that expire mid-scan are not re-vended. Renewal needs a channel of its own. Claude-Session: https://claude.ai/code/session_01M3mYXBKShBonG6Lg3br4Ld --- .../iceberg/scripts/lance_rest_server.py | 5 +- .../datasource/lance/LanceStorageOptions.java | 58 ++++++++++++++---- .../lance/LanceStorageOptionsTest.java | 61 +++++++++++++++++++ gensrc/thrift/PlanNodes.thrift | 3 + 4 files changed, 112 insertions(+), 15 deletions(-) diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py b/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py index 3f614a34c03e49..b8436fa686bb1c 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py +++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py @@ -94,8 +94,9 @@ def _storage_options(identifier: tuple[str, ...]) -> dict[str, str]: "secret_access_key": secret_key, "region": region, "virtual_hosted_style_request": "false", - # Lance refreshes expiring credentials against the namespace itself, so a client has - # to carry this through rather than drop it. + # A key Doris assigns no meaning to, kept here so the pass-through stays covered. + # The BE opens datasets with static options and never refreshes them, so this is + # only carried, not acted on; the expiry is far enough out that it never matters. "expires_at_millis": os.environ.get( "LANCE_S3_EXPIRES_AT_MILLIS", "4102444800000" ), diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageOptions.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageOptions.java index ef75cd45109a13..9f7faad1578030 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageOptions.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageOptions.java @@ -23,6 +23,7 @@ import org.apache.logging.log4j.Logger; import java.util.HashMap; +import java.util.HashSet; import java.util.Locale; import java.util.Map; import java.util.Set; @@ -62,31 +63,44 @@ public final class LanceStorageOptions { } /** - * The {@code aws_}-prefixed aliases of the options above, mapped to the spelling this class - * emits. + * Every spelling object_store accepts for the options above, mapped to the one this class emits. * *

object_store resolves an alias and its canonical name to one config key and keeps only one - * of the two values, chosen by hash order. So a namespace vending {@code aws_endpoint} while - * the catalog contributes {@code endpoint} does not override it - the two survive as separate - * entries and Lance later picks between them unpredictably. Renaming the vended aliases first - * makes the merge below decide, every time. - * - *

Only the prefixed aliases are renamed. Bare names such as {@code token} are ambiguous - * across providers - object_store reads it as an S3 session token but as a bearer token for - * Azure - and this class does not know which provider a dataset uses. + * of the two values, chosen by hash order. So a namespace vending {@code endpoint_url} while the + * catalog contributes {@code endpoint} does not override it - the two survive as separate + * entries, and the FE and the BE can each end up using a different one. Every accepted alias has + * to be recognized here, or that race simply moves to the spellings this table misses. */ - private static final Map VENDED_ALIASES = ImmutableMap.builder() + private static final Map CANONICAL_BY_ALIAS = ImmutableMap.builder() + .put("access_key_id", "access_key_id") .put("aws_access_key_id", "access_key_id") + .put("secret_access_key", "secret_access_key") .put("aws_secret_access_key", "secret_access_key") + .put("session_token", "session_token") .put("aws_session_token", "session_token") .put("aws_token", "session_token") + .put("token", "session_token") + .put("endpoint", "endpoint") + .put("endpoint_url", "endpoint") .put("aws_endpoint", "endpoint") .put("aws_endpoint_url", "endpoint") + .put("region", "region") .put("aws_region", "region") + .put("virtual_hosted_style_request", "virtual_hosted_style_request") .put("aws_virtual_hosted_style_request", "virtual_hosted_style_request") + .put("allow_http", "allow_http") .put("aws_allow_http", "allow_http") .build(); + /** + * Aliases that supersede the catalog's value but keep the spelling the namespace used. + * + *

{@code token} means an S3 session token to object_store's S3 parser but a bearer token to + * its Azure one, and this class does not know which provider a dataset uses. Renaming it would + * corrupt the Azure reading, so it is only used to decide which catalog entry it replaces. + */ + private static final Set AMBIGUOUS_ALIASES = ImmutableSet.of("token"); + /** * Options a namespace may not override, because they decide which data is read rather than how * it is accessed. Lance protects the same keys in the options it accepts from a namespace. @@ -123,8 +137,11 @@ public static Map mergeVended(Map lanceOptions, if (vendedOptions == null || vendedOptions.isEmpty()) { return result; } + + Map accepted = new HashMap<>(); + Set superseded = new HashSet<>(); vendedOptions.forEach((key, value) -> { - if (key == null) { + if (key == null || value == null || value.isEmpty()) { return; } String lowerCased = key.toLowerCase(Locale.ROOT); @@ -133,8 +150,23 @@ public static Map mergeVended(Map lanceOptions, + "would change which data is read", key); return; } - putIfNotEmpty(result, VENDED_ALIASES.getOrDefault(lowerCased, key), value); + String canonical = CANONICAL_BY_ALIAS.get(lowerCased); + if (canonical != null) { + superseded.add(canonical); + } + accepted.put(canonical != null && !AMBIGUOUS_ALIASES.contains(lowerCased) + ? canonical : key, value); }); + + // Drop the catalog's spelling of every option the namespace just supplied, so the two can + // never reach Lance as competing entries for one config key. + result.keySet().removeAll(superseded); + // allow_http describes the endpoint, so a vended endpoint invalidates a value derived from + // the catalog's. An explicitly vended allow_http is in `superseded` and survives. + if (superseded.contains("endpoint") && !superseded.contains("allow_http")) { + result.remove("allow_http"); + } + result.putAll(accepted); return withDerivedAllowHttp(result); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java index ab12f6dac4d606..1e7144414708e5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java @@ -115,6 +115,48 @@ public void testVendedPrefixedAliasReplacesTheCatalogEntry() { } } + /** + * object_store accepts four spellings of the endpoint and four of the session token. Any one + * this class fails to recognize reintroduces the race, so the whole equivalence class has to + * collapse onto a single entry. + */ + @Test + public void testEveryAcceptedAliasCollapsesOntoOneEntry() { + for (String alias : new String[] {"endpoint", "endpoint_url", "aws_endpoint", + "aws_endpoint_url", "ENDPOINT", "AWS_Endpoint_Url"}) { + Map vended = new HashMap<>(); + vended.put(alias, "http://127.0.0.1:9000"); + + Map merged = LanceStorageOptions.mergeVended( + LanceStorageOptions.toLanceOptions(minioCatalogProperties()), vended); + + long endpoints = merged.entrySet().stream() + .filter(e -> e.getKey().toLowerCase(java.util.Locale.ROOT).contains("endpoint")) + .count(); + Assertions.assertEquals(1, endpoints, "alias " + alias + " left a competing entry"); + Assertions.assertEquals("http://127.0.0.1:9000", merged.get("endpoint"), + "alias " + alias + " did not win"); + } + } + + /** + * {@code token} is an S3 session token to object_store but a bearer token to its Azure parser, + * so it keeps the namespace's spelling - it still has to displace the catalog's entry though. + */ + @Test + public void testAmbiguousAliasSupersedesWithoutBeingRenamed() { + Map catalogProperties = minioCatalogProperties(); + catalogProperties.put("AWS_TOKEN", "static-token"); + + Map vended = new HashMap<>(); + vended.put("token", "vended-token"); + + Map merged = LanceStorageOptions.mergeVended( + LanceStorageOptions.toLanceOptions(catalogProperties), vended); + Assertions.assertEquals("vended-token", merged.get("token")); + Assertions.assertNull(merged.get("session_token")); + } + @Test public void testVendedUnprefixedOptionsReplaceTheCatalogEntry() { Map vended = new HashMap<>(); @@ -152,6 +194,25 @@ public void testAllowHttpFollowsTheEndpointActuallyUsed() { Collections.emptyMap(), vendedWithFlag).get("allow_http")); } + /** + * The catalog's plain-HTTP endpoint derives allow_http. Replacing it with an HTTPS endpoint has + * to retract that, or the merged options keep permitting plain HTTP for an endpoint that never + * asked for it. + */ + @Test + public void testAllowHttpIsRetractedWhenTheEndpointBecomesHttps() { + Map catalogOptions = + LanceStorageOptions.toLanceOptions(minioCatalogProperties()); + Assertions.assertEquals("true", catalogOptions.get("allow_http")); + + Map vended = new HashMap<>(); + vended.put("endpoint", "https://s3.amazonaws.com"); + + Map merged = LanceStorageOptions.mergeVended(catalogOptions, vended); + Assertions.assertEquals("https://s3.amazonaws.com", merged.get("endpoint")); + Assertions.assertNull(merged.get("allow_http")); + } + /** * Lance recognizes this key on its namespace-backed refresh path. lance-c opens datasets with * static options, so it has no effect on the BE today, but the option map is meant to reach diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index eecbc4690dde1f..4a32901296cf6a 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -640,6 +640,9 @@ struct TFileScanRangeParams { // BE imposed here would drop options it does not happen to know - including credentials a // namespace spelled with a different accepted alias, and every non-S3 provider's keys. // Set at ScanNode level so credentials are not serialized once per fragment split. + // These are the initial options for the scan and are never refreshed: lance-c opens datasets + // with a static option set, so credentials that expire mid-scan are not re-vended. Renewal + // needs a refresh channel of its own, which this field is not. 38: optional map lance_storage_options }