This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 54423854c7e [fix](lance) Pass namespace-vended storage options through
to the BE (#66805)
54423854c7e is described below
commit 54423854c7e90e2c77f7fe6c5a4aa49935ab6d93
Author: FANNG <[email protected]>
AuthorDate: Tue Aug 25 09:48:20 2026 +0800
[fix](lance) Pass namespace-vended storage options through to the BE
(#66805)
### What problem does this PR solve?
Fixes #66772 (problem 1).
A Lance REST catalog discarded the `storage_options` a namespace vended
for a table, so any scan relying on credential vending failed:
```
open Lance dataset failed: LanceError(IO): ...
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 one
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
correct outbound. Reading it backwards turns it into a parser that
accepts only the spelling it happens to emit, so everything 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, which made it hard to
read: the FE passes the vended map to the Lance Java SDK untouched, so
`SHOW TABLES` and `DESC` succeeded and only the scan failed.
The Lance Namespace specification describes `storage_options` as
configuration *"passed directly to Lance"*, so the protocol defines no
key vocabulary of its own and a client cannot assume one.
### Release note
Fixed a Lance REST catalog failing to scan with `CredentialsNotLoaded`
when it relied on credentials vended by the namespace rather than static
`s3.access_key` / `s3.secret_key`.
### What is changed and how it works?
Stop re-encoding server-supplied options, and resolve the ones Doris
does contribute per storage provider.
**Transport.** `TFileScanRangeParams.lance_storage_options` (new, id 38)
carries the options in Lance's own vocabulary. Set at ScanNode level —
the same pattern as `paimon_options` — so credentials are not serialized
once per fragment split. `properties` is deliberately not reused: it is
copied wholesale into `io::FileSystemProperties` for the shared
filesystem layer. The BE hands that map to lance-c as it arrives, so
`kStorageKeys` is gone. The `s3()` TVF path populates the same field,
including the schema-fetch RPC.
**The option vocabulary belongs to a provider, not to this layer.**
Lance routes a dataset to a provider by URL scheme, and the accepted
spellings overlap without agreeing:
```rust
// object_store-0.13.2 src/azure/builder.rs:448 — note: no aws_endpoint
"azure_storage_endpoint" | "azure_endpoint" | "endpoint" =>
Ok(Self::Endpoint)
// lance-io/src/object_store/providers/oss.rs:71-73, :87
("endpoint", &["oss_endpoint"]), ("access_key_id", &["oss_access_key_id"]),
...
"OSS endpoint is required. Please provide 'oss_endpoint' ..."
```
Rewriting a vended `endpoint` onto `aws_endpoint` therefore loses the
Azure endpoint outright and makes OSS fail before it reads anything.
`LanceStorageProvider` picks by scheme the way lance-io's own registry
does, and each implementation owns both directions of its vocabulary:
- **`LanceS3StorageProvider`** — the `s3` and `s3+ddb` schemes lance-io
registers for its AWS provider. It converts the catalog's typed storage
properties into the canonical `aws_*` spellings, and resolves vended
spellings onto those same keys. `token` is decidable here too: it is a
bearer token to object_store's Azure parser, and only the provider
settles which it is.
- **`LancePassThroughStorageProvider`** — everything else. Both halves
inert, so such a dataset is reachable only through what its namespace
vends, named as the namespace named it.
**Why the two sides have to be resolved against each other.** A catalog
with static `s3.access_key` whose namespace also vends credentials
produces both spellings of one option:
```
aws_access_key_id = <catalog> access_key_id = <vended>
```
Both parse to `AmazonS3ConfigKey::AccessKeyId`, and `as_s3_options()`
collects them into a `HashMap`, so the winner is whichever the iteration
yields last — decided separately for the access key and the secret, and
separately in the FE and in the BE. Two of the four combinations pair
one side's key with the other's secret, which surfaces as
`SignatureDoesNotMatch` rather than as a missing credential. Putting
both sides in the provider's vocabulary first means one key, one value,
and the namespace wins because it just described the table.
The alias set covers only the options Doris itself contributes — those
are the only ones a vended option can collide with.
`aws_endpoint_url_s3` is deliberately left alone: object_store parses it
into a config key of its own and prefers it over the generic endpoint,
so a vended one already wins, and folding it in would replace a defined
precedence with map order.
**Doris's own configuration is read through its typed properties**, not
through the flattened `AWS_*` backend map, which is only a re-encoding
of them (`AbstractS3CompatibleProperties.doBuildS3Configuration`). That
map also mixes in backend-only knobs Lance has no use for, and flattens
every configured storage into one namespace where two S3-compatible ones
overwrite each other. The typed list keeps them apart; the one to read
is chosen the way Iceberg chooses it
(`AbstractIcebergProperties.toFileIOProperties`) — prefer a concrete
provider over the generic `S3Properties`. Selection filters by type
rather than index, because `StorageProperties.createAll` prepends a
default HDFS entry when no `fs.xx.support` flag is set.
**What is rejected rather than resolved.** A NUL truncates a key at
lance-c's `CStr::from_ptr` while the FE keeps reading the whole thing,
so the FE rejects it and the BE returns `InvalidArgument`; dropping it
silently would only move the divergence to an FE that predates the
check. A namespace vending two spellings of one option with different
values is rejected too — picking one would be the coin toss above.
### Not in scope
**Credentials are static for the life of a scan.** `lance-c` opens
datasets with a static option set, so `expires_at_millis` reaches the BE
but is never acted on, and credentials that expire mid-scan are not
re-vended. Renewal needs a channel of its own; this field is not one,
and the thrift comment says so.
**One map, two Lances.** The FE and the BE agree on the options, not on
the library that reads them: both pin `object_store` 0.13.2, but the
Java SDK (9.1.0-beta.3) carries OpenDAL 0.57 and `lance-c` 0.1.6 carries
0.56. An option only the newer one knows — `skip_signature`, say — takes
effect on the FE and is ignored on the BE. This only reaches the OpenDAL
backend, which a namespace has to opt into by vending
`use_opendal=true`, and closing it means aligning the two upstream pins.
**Non-S3 providers are unblocked, not enabled.** Their vended options
now survive instead of being dropped or rewritten, but Doris contributes
no static configuration for them — it models OSS and COS perfectly well,
so writing that translation is possible, but it would mean committing to
a vocabulary per provider with no backend here to test it against, which
is how the rewriting bug above got in. A filesystem catalog's
`warehouse` is still restricted to `file` and `s3`.
**A catalog and a namespace that disagree.** If both name the same
option in spellings the provider treats as distinct config keys, both
survive and Lance decides. Only the spellings Doris itself emits are
resolved.
### Rolling upgrade
An FE upgraded ahead of the BEs no longer puts vended credentials into
`TFileScanRangeParams.properties`, and an older BE reads only that. A
REST catalog with no static credentials therefore cannot be scanned
until the BEs are upgraded too. The Lance catalog is not in a release
yet, so this only affects development clusters.
### How was this patch tested?
**Unit tests.** `LanceStorageOptionsTest` is new — `LanceStorageOptions`
had no test of its own, and the deleted `forBackend` had zero call sites
in any test, which is why the original bug went unnoticed. Its fixtures
are built with `StorageProperties.createAll` from user-facing
properties, so they exercise Doris's real parsing and have to satisfy
its rules. It covers the canonical mapping and anonymous access, vended
credentials superseding the catalog's, the full endpoint alias class,
`token` resolving differently on `s3://` than on `az://`, six non-S3
schemes arriving untouched, S3 credentials not leaking onto another
provider, selection skipping the non-S3 entry Doris prepends, unknown
options passing through, conflicting spellings and embedded NULs being
rejected. `LanceThriftContractTest` gains round-trip coverage of the new
field.
**Regression.** The docker stub only ever vended `aws_`-prefixed keys,
so the alias path had no coverage anywhere. It now serves a second
table, `all_types_unprefixed`, backed by the same dataset but vending
the unprefixed spelling, and `test_lance_rest_catalog` scans it with no
static credentials configured. Both tables scan correctly on the current
revision.
**End to end**, against Apache Gravitino 1.3.0's `lance-rest` service,
which vends the unprefixed spelling. This run validated the original
transport fix, on the first revision of this PR; the docker suite above
is what has been re-run since. Catalog with no `s3.access_key` /
`s3.secret_key`:
```sql
CREATE CATALOG lance_novend PROPERTIES (
"type" = "lance",
"lance.catalog.type" = "rest",
"lance.rest.uri" = "http://127.0.0.1:9101/lance";,
"lance.namespace.parent" = "lance_catalog"
);
SELECT count(*), min(row_id), max(row_id), sum(row_id) FROM
lance_novend.doris_probe.rest_probe;
-- 1034 1 1034 535095
```
against a 1034-row, 2-fragment dataset with an IVF_FLAT index. Predicate
pushdown and vector search both return correct results on the same
catalog, and a catalog with static credentials is unaffected. Before
this change the same statement failed with `CredentialsNotLoaded`.
https://claude.ai/code/session_01M3mYXBKShBonG6Lg3br4Ld
---
be/src/format_v2/table/lance_reader.cpp | 70 +++--
be/src/format_v2/table/lance_reader.h | 5 +-
be/test/format_v2/table/lance_reader_test.cpp | 25 ++
.../docker-compose/iceberg/iceberg.yaml.tpl | 5 +-
.../iceberg/scripts/lance_rest_server.py | 55 +++-
.../datasource/lance/LanceExternalCatalog.java | 61 ++--
.../datasource/lance/LanceMetadataLoader.java | 35 +--
.../lance/LancePassThroughStorageProvider.java | 59 ++++
.../datasource/lance/LanceS3StorageProvider.java | 193 +++++++++++++
.../datasource/lance/LanceStorageOptions.java | 111 +++++---
.../datasource/lance/LanceStorageProvider.java | 84 ++++++
.../doris/datasource/lance/LanceTableMetadata.java | 19 +-
.../datasource/lance/source/LanceScanNode.java | 9 +-
.../metastore/AbstractLanceProperties.java | 9 +
.../LanceFileSystemMetastoreProperties.java | 6 +
.../doris/datasource/tvf/source/TVFScanNode.java | 16 ++
.../ExternalFileTableValuedFunction.java | 17 ++
.../doris/tablefunction/S3TableValuedFunction.java | 5 +-
.../doris/datasource/LanceThriftContractTest.java | 43 +++
.../lance/LanceFilesystemCatalogTest.java | 18 --
.../doris/datasource/lance/LanceSnapshotTest.java | 4 +-
.../datasource/lance/LanceStorageOptionsTest.java | 314 +++++++++++++++++++++
gensrc/thrift/PlanNodes.thrift | 9 +
.../lance/test_lance_rest_catalog.out | 4 +
.../lance/test_lance_rest_catalog.groovy | 8 +
25 files changed, 1012 insertions(+), 172 deletions(-)
diff --git a/be/src/format_v2/table/lance_reader.cpp
b/be/src/format_v2/table/lance_reader.cpp
index 7491f7a8c02..d909aac8e5e 100644
--- a/be/src/format_v2/table/lance_reader.cpp
+++ b/be/src/format_v2/table/lance_reader.cpp
@@ -261,7 +261,8 @@ Status LanceTableReader::fetch_schema(const TFileRangeDesc&
range,
return Status::InvalidArgument("Lance schema output must not be null");
}
const auto& params = range.table_format_params.lance_params;
- const auto storage_options = _storage_options(&scan_params);
+ std::vector<std::string> storage_options;
+ RETURN_IF_ERROR(_storage_options(&scan_params, &storage_options));
std::vector<const char*> storage_option_ptrs;
storage_option_ptrs.reserve(storage_options.size() + 1);
for (const auto& option : storage_options) {
@@ -624,7 +625,8 @@ Status
LanceTableReader::_validate_external_search_request() const {
}
Status LanceTableReader::_ensure_dataset_open(const TFileRangeDesc& range) {
- const auto key = _dataset_key(range);
+ DatasetKey key;
+ RETURN_IF_ERROR(_dataset_key(range, &key));
if (_dataset == nullptr) {
RETURN_IF_ERROR(_open_dataset(key));
_opened_dataset_key = key;
@@ -1025,47 +1027,39 @@ Status LanceTableReader::_fill_block_from_record_batch(
return Status::OK();
}
-std::vector<std::string> LanceTableReader::_storage_options(
- const TFileScanRangeParams* scan_params) {
- if (scan_params == nullptr || !scan_params->__isset.properties) {
- return {};
- }
- static constexpr std::array<std::pair<std::string_view, std::string_view>,
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<std::string> 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");
+// 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.
+Status LanceTableReader::_storage_options(const TFileScanRangeParams*
scan_params,
+ std::vector<std::string>* options) {
+ options->clear();
+ if (scan_params == nullptr || !scan_params->__isset.lance_storage_options)
{
+ return Status::OK();
}
- 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) {
+ // These become C strings below, so a NUL would truncate the option
here while the FE went
+ // on using the whole thing, and the two halves would open the dataset
with different
+ // configuration. The FE rejects these on both paths it builds options
from - its own
+ // storage configuration and what a namespace vends - so this is the
last line of defence,
+ // for an FE that predates those checks. Dropping one here instead of
failing would just
+ // recreate the divergence it exists to prevent.
+ if (key.find('\0') != std::string::npos || value.find('\0') !=
std::string::npos) {
+ return Status::InvalidArgument(
+ "Lance storage option '{}' contains a NUL and cannot reach
lance-c",
+ key.substr(0, key.find('\0')));
+ }
+ options->emplace_back(key);
+ options->emplace_back(value);
}
- return options;
+ return Status::OK();
}
-LanceTableReader::DatasetKey LanceTableReader::_dataset_key(const
TFileRangeDesc& range) const {
+Status LanceTableReader::_dataset_key(const TFileRangeDesc& range, DatasetKey*
key) const {
const auto& params = range.table_format_params.lance_params;
- return {
- .uri = params.dataset_uri,
- .version = params.version,
- .storage_options = _storage_options(_scan_params),
- };
+ key->uri = params.dataset_uri;
+ key->version = params.version;
+ return _storage_options(_scan_params, &key->storage_options);
}
Status LanceTableReader::_lance_error(std::string_view operation) {
diff --git a/be/src/format_v2/table/lance_reader.h
b/be/src/format_v2/table/lance_reader.h
index d16acf1c459..bda465185cb 100644
--- a/be/src/format_v2/table/lance_reader.h
+++ b/be/src/format_v2/table/lance_reader.h
@@ -95,8 +95,9 @@ private:
Block* block, size_t* rows);
Status _append_global_row_ids(const std::shared_ptr<arrow::Array>& row_ids,
MutableColumnPtr& output_column) const;
- static std::vector<std::string> _storage_options(const
TFileScanRangeParams* scan_params);
- DatasetKey _dataset_key(const TFileRangeDesc& range) const;
+ static Status _storage_options(const TFileScanRangeParams* scan_params,
+ std::vector<std::string>* options);
+ Status _dataset_key(const TFileRangeDesc& range, DatasetKey* key) const;
static Status _lance_error(std::string_view operation);
LanceDataset* _dataset = nullptr;
diff --git a/be/test/format_v2/table/lance_reader_test.cpp
b/be/test/format_v2/table/lance_reader_test.cpp
index 5d6e8f637b0..9b2943f0ad6 100644
--- a/be/test/format_v2/table/lance_reader_test.cpp
+++ b/be/test/format_v2/table/lance_reader_test.cpp
@@ -948,6 +948,31 @@ TEST(LanceTableReaderScanTest,
ReadsLatestSnapshotWithoutFragmentIds) {
EXPECT_TRUE(reader.close().ok());
}
+TEST(LanceTableReaderScanTest, RejectsStorageOptionWithEmbeddedNul) {
+ const std::filesystem::path dataset_uri =
+ "./be/test/format_v2/table/lance/data/all_types.lance";
+ const Columns columns {projected_column("row_id", TYPE_BIGINT, false)};
+ TQueryGlobals query_globals;
+ RuntimeState state(query_globals);
+ RuntimeProfile profile("lance_storage_option_embedded_nul");
+
+ // lance-c reads these as C strings, so a NUL truncates the option here
while the FE goes on
+ // using the whole thing, leaving the two halves opening the dataset with
different
+ // configuration. Dropping it instead of failing would only move that
divergence.
+ TFileScanRangeParams scan_params;
+ scan_params.__set_lance_storage_options(
+ {{std::string("aws_region\0ignored", 18), "us-east-1"}});
+
+ LanceTableReader reader;
+ ASSERT_TRUE(init_reader(&reader, columns, &state, &profile,
&scan_params).ok());
+
+ const auto status = prepare_range(&reader,
make_latest_lance_range(dataset_uri));
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("contains a NUL"), std::string::npos);
+ EXPECT_TRUE(reader.close().ok());
+}
+
TEST(LanceTableReaderTypeTest, ReadsNumericTypesFromAllTypesFixture) {
// The committed fixture contains four rows covering values, nulls, and
boundary cases.
const std::filesystem::path dataset_uri =
diff --git a/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
b/docker/thirdparties/docker-compose/iceberg/iceberg.yaml.tpl
index 5fc6af63e77..90b20a444ea 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 8aef278c024..b8436fa686b 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,50 @@ 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",
+ # 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"
+ ),
+ }
+ 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 +185,7 @@ class LanceRestHandler(BaseHTTPRequestHandler):
"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 7629d68f6cc..c2fa3d1a6ee 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
@@ -86,8 +86,7 @@ public class LanceExternalCatalog extends ExternalCatalog {
private transient List<String> parentNamespace = Collections.emptyList();
private transient String catalogType;
private transient String rootDatabase;
- private transient Map<String, String> javaStorageOptions =
Collections.emptyMap();
- private transient Map<String, String> backendStorageOptions =
Collections.emptyMap();
+ private transient Map<String, String> namespaceStorageOptions =
Collections.emptyMap();
private transient Object namespaceLock = new Object();
public LanceExternalCatalog(long catalogId, String name, String resource,
Map<String, String> props,
@@ -105,11 +104,12 @@ public class LanceExternalCatalog extends ExternalCatalog
{
rootDatabase = properties.getRootDatabase();
parentNamespace = LanceNamespaceName.parseParentNamespace(
properties.getNamespaceParent(),
properties.getNamespaceDelimiter());
- backendStorageOptions =
catalogProperty.getBackendStorageProperties();
- javaStorageOptions =
LanceStorageOptions.forJavaSdk(backendStorageOptions);
+ namespaceStorageOptions = LanceStorageOptions.forUri(
+ properties.getNamespaceStorageUri(),
+ catalogProperty.getOrderedStoragePropertiesList());
allocator = new RootAllocator(ALLOCATOR_LIMIT);
- namespace = properties.createNamespace(allocator,
javaStorageOptions);
+ namespace = properties.createNamespace(allocator,
namespaceStorageOptions);
} catch (Exception e) {
closeLanceObjects();
throw new RuntimeException("Failed to initialize Lance catalog '"
+ getName()
@@ -127,8 +127,9 @@ public class LanceExternalCatalog extends ExternalCatalog {
}
AbstractLanceProperties properties = getLanceProperties();
- Map<String, String> storageOptions = LanceStorageOptions.forJavaSdk(
- catalogProperty.getBackendStorageProperties());
+ Map<String, String> storageOptions = LanceStorageOptions.forUri(
+ properties.getNamespaceStorageUri(),
+ catalogProperty.getOrderedStoragePropertiesList());
List<String> parent = LanceNamespaceName.parseParentNamespace(
properties.getNamespaceParent(),
properties.getNamespaceDelimiter());
String type = properties.getLanceCatalogType();
@@ -316,16 +317,16 @@ public class LanceExternalCatalog extends ExternalCatalog
{
"Cannot parse Lance FOR TIME AS OF value '" +
snapshot.getValue() + "'");
}
version = LanceSnapshotResolver.getVersionAtOrBefore(
- tableAccess.datasetUri,
tableAccess.javaStorageOptions, timestamp, allocator);
+ tableAccess.datasetUri,
tableAccess.storageOptions, timestamp, allocator);
}
- return LanceMetadataLoader.loadVersion(tableAccess.datasetUri,
tableAccess.javaStorageOptions,
- tableAccess.backendStorageOptions, version, allocator);
+ return LanceMetadataLoader.loadVersion(
+ tableAccess.datasetUri, tableAccess.storageOptions,
version, allocator);
}
return loadIndexSegments
?
LanceMetadataLoader.loadLatestWithIndexSegments(tableAccess.datasetUri,
- tableAccess.javaStorageOptions,
tableAccess.backendStorageOptions, allocator)
- : LanceMetadataLoader.loadLatest(tableAccess.datasetUri,
tableAccess.javaStorageOptions,
- tableAccess.backendStorageOptions, allocator);
+ tableAccess.storageOptions, allocator)
+ : LanceMetadataLoader.loadLatest(
+ tableAccess.datasetUri,
tableAccess.storageOptions, allocator);
} catch (Exception e) {
throw new RuntimeException("Failed to load Lance table metadata
for " + dbName + "." + tableName
+ ": " + sanitizedRootCauseMessage(e), safeCause(e));
@@ -340,7 +341,8 @@ public class LanceExternalCatalog extends ExternalCatalog {
try {
makeSureInitialized();
} catch (Exception e) {
- throw indexMetadataLoadFailure(dbName, tableName, e, null,
javaStorageOptions);
+ throw indexMetadataLoadFailure(
+ dbName, tableName, e, null, namespaceStorageOptions);
}
ResolvedTableAccess tableAccess = null;
@@ -351,7 +353,7 @@ public class LanceExternalCatalog extends ExternalCatalog {
// The deadline below covers the Dataset/JNI index metadata read
itself.
tableAccess = resolveTableAccess(dbName, tableName);
String datasetUri = tableAccess.datasetUri;
- Map<String, String> storageOptions =
tableAccess.javaStorageOptions;
+ Map<String, String> storageOptions = tableAccess.storageOptions;
return LanceMetadataReadExecutor.execute(() -> {
// The caller may return on deadline while JNI is still
running. A task-owned
// allocator prevents catalog close from releasing native
resources prematurely.
@@ -362,7 +364,7 @@ public class LanceExternalCatalog extends ExternalCatalog {
} catch (Exception e) {
String datasetUri = tableAccess == null ? null :
tableAccess.datasetUri;
Map<String, String> runtimeStorageOptions = tableAccess == null
- ? javaStorageOptions : tableAccess.javaStorageOptions;
+ ? namespaceStorageOptions : tableAccess.storageOptions;
throw indexMetadataLoadFailure(
dbName, tableName, e, datasetUri, runtimeStorageOptions);
}
@@ -391,13 +393,12 @@ public class LanceExternalCatalog extends ExternalCatalog
{
throw new RuntimeException("Lance namespace returned no table URI
for " + dbName + "." + tableName);
}
- Map<String, String> tableJavaStorageOptions = new
HashMap<>(javaStorageOptions);
- if (table.getStorageOptions() != null) {
- tableJavaStorageOptions.putAll(table.getStorageOptions());
- }
- Map<String, String> tableBackendStorageOptions =
LanceStorageOptions.forBackend(
- backendStorageOptions, table.getStorageOptions());
- return new ResolvedTableAccess(datasetUri, tableJavaStorageOptions,
tableBackendStorageOptions);
+ // 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. The
+ // dataset URL picks the option vocabulary, the same way Lance picks a
provider from it.
+ Map<String, String> storageOptions =
LanceStorageOptions.forVendedTable(datasetUri,
+ catalogProperty.getOrderedStoragePropertiesList(),
table.getStorageOptions());
+ return new ResolvedTableAccess(datasetUri, storageOptions);
}
private DescribeTableResponse describeTable(String dbName, String
tableName) {
@@ -431,11 +432,6 @@ public class LanceExternalCatalog extends ExternalCatalog {
return result;
}
- public Map<String, String> getBackendStorageOptions() {
- makeSureInitialized();
- return backendStorageOptions;
- }
-
public String getLanceCatalogType() {
makeSureInitialized();
return catalogType;
@@ -538,14 +534,11 @@ public class LanceExternalCatalog extends ExternalCatalog
{
private static final class ResolvedTableAccess {
private final String datasetUri;
- private final Map<String, String> javaStorageOptions;
- private final Map<String, String> backendStorageOptions;
+ private final Map<String, String> storageOptions;
- private ResolvedTableAccess(String datasetUri, Map<String, String>
javaStorageOptions,
- Map<String, String> backendStorageOptions) {
+ private ResolvedTableAccess(String datasetUri, Map<String, String>
storageOptions) {
this.datasetUri = datasetUri;
- this.javaStorageOptions = Collections.unmodifiableMap(new
HashMap<>(javaStorageOptions));
- this.backendStorageOptions = Collections.unmodifiableMap(new
HashMap<>(backendStorageOptions));
+ this.storageOptions = Collections.unmodifiableMap(new
HashMap<>(storageOptions));
}
}
}
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 7c9faf2f0ec..9c4803af710 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
@@ -18,6 +18,7 @@
package org.apache.doris.datasource.lance;
import org.apache.doris.common.util.JsonUtil;
+import org.apache.doris.datasource.property.storage.StorageProperties;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.arrow.memory.BufferAllocator;
@@ -49,11 +50,11 @@ public final class LanceMetadataLoader {
* Lance dataset through an S3 TVF.
*/
public static LanceTableMetadata loadLatestForTvf(
- String datasetUri, Map<String, String> backendStorageOptions)
+ String datasetUri, List<StorageProperties> storageProperties)
throws Exception {
try (BufferAllocator allocator = new RootAllocator(ALLOCATOR_LIMIT)) {
- return loadLatest(datasetUri,
LanceStorageOptions.forJavaSdk(backendStorageOptions),
- backendStorageOptions, allocator);
+ return loadLatest(datasetUri,
+ LanceStorageOptions.forUri(datasetUri, storageProperties),
allocator);
}
}
@@ -65,18 +66,17 @@ public final class LanceMetadataLoader {
* time-travel version is requested. Schema, version, and fragments are
read from the same
* opened dataset snapshot.
*/
- public static LanceTableMetadata loadLatest(String datasetUri, Map<String,
String> javaStorageOptions,
- Map<String, String> backendStorageOptions, BufferAllocator
allocator) throws Exception {
+ public static LanceTableMetadata loadLatest(String datasetUri,
+ Map<String, String> lanceStorageOptions, BufferAllocator
allocator) throws Exception {
return loadInternal(
- datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.empty(), allocator, false);
+ datasetUri, lanceStorageOptions, OptionalLong.empty(),
allocator, false);
}
/** Loads the latest fixed snapshot together with vector index segment
coverage. */
public static LanceTableMetadata loadLatestWithIndexSegments(
- String datasetUri, Map<String, String> javaStorageOptions,
- Map<String, String> backendStorageOptions, BufferAllocator
allocator) throws Exception {
+ String datasetUri, Map<String, String> lanceStorageOptions,
BufferAllocator allocator) throws Exception {
return loadInternal(
- datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.empty(), allocator, true);
+ datasetUri, lanceStorageOptions, OptionalLong.empty(),
allocator, true);
}
/**
@@ -86,18 +86,19 @@ public final class LanceMetadataLoader {
* {@link LanceExternalCatalog#loadTableMetadata(String, String,
java.util.Optional)} for both
* {@code FOR VERSION AS OF} and the version resolved from {@code FOR TIME
AS OF}.
*/
- public static LanceTableMetadata loadVersion(String datasetUri,
Map<String, String> javaStorageOptions,
- Map<String, String> backendStorageOptions, long version,
BufferAllocator allocator) throws Exception {
+ public static LanceTableMetadata loadVersion(String datasetUri,
+ Map<String, String> lanceStorageOptions, long version,
BufferAllocator allocator)
+ throws Exception {
return loadInternal(
- datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.of(version), allocator, false);
+ datasetUri, lanceStorageOptions, OptionalLong.of(version),
allocator, false);
}
/** Shared implementation for the latest-version and explicit-version
public entry points. */
- private static LanceTableMetadata loadInternal(String datasetUri,
Map<String, String> javaStorageOptions,
- Map<String, String> backendStorageOptions, OptionalLong version,
+ private static LanceTableMetadata loadInternal(String datasetUri,
+ Map<String, String> lanceStorageOptions, OptionalLong version,
BufferAllocator allocator, boolean loadIndexSegments) 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<LanceFragmentInfo> fragments = new ArrayList<>();
for (Fragment fragment : dataset.getFragments()) {
@@ -112,9 +113,9 @@ public final class LanceMetadataLoader {
return loadIndexSegments
? LanceTableMetadata.withIndexSegments(datasetUri,
resolvedVersion,
dataset.getSchema(), fragments, lanceFieldIds,
- indexSegments, backendStorageOptions)
+ indexSegments, lanceStorageOptions)
: LanceTableMetadata.withoutIndexSegments(datasetUri,
resolvedVersion,
- dataset.getSchema(), fragments,
backendStorageOptions);
+ dataset.getSchema(), fragments,
lanceStorageOptions);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LancePassThroughStorageProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LancePassThroughStorageProvider.java
new file mode 100644
index 00000000000..78cc69ee9be
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LancePassThroughStorageProvider.java
@@ -0,0 +1,59 @@
+// 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.apache.doris.datasource.property.storage.StorageProperties;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Every provider Lance routes somewhere other than S3 - Azure, GCS, OSS,
Tencent COS, local files,
+ * and anything Lance adds later.
+ *
+ * <p>Both halves are deliberately inert, so such a dataset is reachable only
through what its
+ * namespace vends. Those options are the namespace's to name: Lance's OSS
provider reads
+ * {@code access_key_id} and requires {@code endpoint}, object_store's Azure
parser reads
+ * {@code endpoint} and takes {@code token} as a bearer token. Rewriting any
of that onto the S3
+ * spellings would leave the dataset unreachable, which is what this class
exists to prevent.
+ *
+ * <p>{@link #fromDorisProperties} is empty for want of a translation, not for
want of input: Doris
+ * does model these - {@code OSSProperties} and {@code COSProperties} carry an
endpoint and
+ * credentials like any other. Writing that translation means committing to a
vocabulary per
+ * provider with no way to exercise it here, which is how the rewriting bug
above got in, so it
+ * waits for a backend that can be tested against.
+ */
+final class LancePassThroughStorageProvider implements LanceStorageProvider {
+
+ static final LancePassThroughStorageProvider INSTANCE = new
LancePassThroughStorageProvider();
+
+ private LancePassThroughStorageProvider() {
+ }
+
+ @Override
+ public Map<String, String> fromDorisProperties(List<StorageProperties>
storageProperties) {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, String> normalizeVended(Map<String, String>
vendedOptions) {
+ return vendedOptions == null ? new HashMap<>() : new
HashMap<>(vendedOptions);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceS3StorageProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceS3StorageProvider.java
new file mode 100644
index 00000000000..eb723a0201f
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceS3StorageProvider.java
@@ -0,0 +1,193 @@
+// 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.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
+import org.apache.doris.datasource.property.storage.S3Properties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * S3-compatible storage, which Lance reaches through object_store's AWS
provider.
+ *
+ * <p>The emitted spelling is the one object_store reports as canonical,
because that is what
+ * {@code StorageOptions::with_env_s3} looks for before pulling the same
option out of the process
+ * environment:
+ *
+ * <pre>
+ * // lance-io/src/object_store/providers/aws.rs
+ * if let Ok(config_key) =
AmazonS3ConfigKey::from_str(&key.to_ascii_lowercase())
+ * && !self.0.contains_key(config_key.as_ref()) //
"aws_access_key_id"
+ * </pre>
+ *
+ * <p>Any other accepted alias leaves that check unsatisfied, so a stray
{@code AWS_ACCESS_KEY_ID}
+ * in the FE or BE environment is inserted next to the configured value, and
object_store's
+ * {@code as_s3_options()} then folds both onto one config key and keeps
whichever its HashMap
+ * yields last - independently per option and per process. Lance's OpenDAL S3
backend accepts these
+ * same spellings as serde aliases of its own field names.
+ */
+final class LanceS3StorageProvider implements LanceStorageProvider {
+
+ static final LanceS3StorageProvider INSTANCE = new
LanceS3StorageProvider();
+
+ private static final String ACCESS_KEY_ID = "aws_access_key_id";
+ private static final String SECRET_ACCESS_KEY = "aws_secret_access_key";
+ private static final String SESSION_TOKEN = "aws_session_token";
+ private static final String REGION = "aws_region";
+ private static final String ENDPOINT = "aws_endpoint";
+ private static final String VIRTUAL_HOSTED_STYLE =
"aws_virtual_hosted_style_request";
+ /**
+ * Not prefixed: object_store carries this as a shared client option, and
reports it as
+ * canonical under this name. The spelling buys nothing against the
environment here, though -
+ * {@code StorageOptions::new} overwrites this key outright from {@code
AWS_ALLOW_HTTP} before
+ * {@code with_env_s3} runs.
+ */
+ private static final String ALLOW_HTTP = "allow_http";
+
+ /**
+ * Every spelling object_store accepts for the options above, mapped onto
the one emitted.
+ *
+ * <p>Confined to those options on purpose. They are the only ones this
class contributes, so
+ * they are the only ones a vended option can collide with; anything else
a namespace sends is
+ * between the namespace and Lance.
+ *
+ * <p>{@code token} is included because this provider only ever speaks for
an S3 dataset, where
+ * it is unambiguously the session token. It means a bearer token to
object_store's Azure
+ * parser, which is why it can only be resolved once the provider is known.
+ *
+ * <p>{@code aws_endpoint_url_s3} is deliberately absent. object_store
parses it into a config
+ * key of its own and prefers it over the generic endpoint, so a vended
one already wins
+ * without being rewritten, and folding it in would replace a defined
precedence with map order.
+ */
+ private static final Map<String, String> CANONICAL_BY_ALIAS =
ImmutableMap.<String, String>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)
+ .put("aws_virtual_hosted_style_request", VIRTUAL_HOSTED_STYLE)
+ // OpenDAL's own field name, of which the two above are serde
aliases. All three would
+ // be the same field supplied more than once, which fails the
operator build outright.
+ .put("enable_virtual_host_style", VIRTUAL_HOSTED_STYLE)
+ .put("allow_http", ALLOW_HTTP)
+ .put("aws_allow_http", ALLOW_HTTP)
+ .build();
+
+ private LanceS3StorageProvider() {
+ }
+
+ @Override
+ public Map<String, String> fromDorisProperties(List<StorageProperties>
storageProperties) {
+ Map<String, String> result = new HashMap<>();
+ AbstractS3CompatibleProperties properties =
selectS3Compatible(storageProperties);
+ if (properties == null) {
+ return result;
+ }
+ putIfNotEmpty(result, ACCESS_KEY_ID, properties.getAccessKey());
+ putIfNotEmpty(result, SECRET_ACCESS_KEY, properties.getSecretKey());
+ putIfNotEmpty(result, SESSION_TOKEN, properties.getSessionToken());
+ putIfNotEmpty(result, ENDPOINT, properties.getEndpoint());
+ putIfNotEmpty(result, REGION, properties.getRegion());
+
+ String usePathStyle = properties.getUsePathStyle();
+ if (usePathStyle != null && !usePathStyle.isEmpty()) {
+ result.put(VIRTUAL_HOSTED_STYLE,
String.valueOf(!Boolean.parseBoolean(usePathStyle)));
+ }
+
+ // Lance refuses a plain-HTTP endpoint unless this is set, and Doris
configures one for
+ // MinIO. It describes the endpoint just mapped, so it is derived from
the same properties.
+ String endpoint = properties.getEndpoint();
+ if (endpoint != null && endpoint.startsWith("http://";)) {
+ result.put(ALLOW_HTTP, "true");
+ }
+ return result;
+ }
+
+ /**
+ * Picks the one S3-compatible configuration to read, preferring a
concrete provider over the
+ * generic {@link S3Properties}: naming OSS or COS explicitly is a choice,
while S3Properties is
+ * also what a heuristic match lands on. The list is not a user-ordered
one - it follows
+ * {@code StorageProperties.PROVIDERS} and may lead with a default HDFS
entry - so it has to be
+ * filtered by type rather than indexed.
+ *
+ * <p>Same rule as {@code AbstractIcebergProperties.toFileIOProperties},
deliberately copied
+ * rather than shared: hoisting it would mean changing Iceberg in this
patch.
+ */
+ private static AbstractS3CompatibleProperties selectS3Compatible(
+ List<StorageProperties> storageProperties) {
+ if (storageProperties == null) {
+ return null;
+ }
+ AbstractS3CompatibleProperties fallback = null;
+ AbstractS3CompatibleProperties concrete = null;
+ for (StorageProperties candidate : storageProperties) {
+ if (!(candidate instanceof AbstractS3CompatibleProperties)) {
+ continue;
+ }
+ if (fallback == null) {
+ fallback = (AbstractS3CompatibleProperties) candidate;
+ }
+ if (concrete == null && !(candidate instanceof S3Properties)) {
+ concrete = (AbstractS3CompatibleProperties) candidate;
+ }
+ }
+ return concrete != null ? concrete : fallback;
+ }
+
+ @Override
+ public Map<String, String> normalizeVended(Map<String, String>
vendedOptions) {
+ Map<String, String> result = new HashMap<>();
+ if (vendedOptions == null) {
+ return result;
+ }
+ vendedOptions.forEach((key, value) -> {
+ String canonical =
CANONICAL_BY_ALIAS.getOrDefault(key.toLowerCase(Locale.ROOT), key);
+ String previous = result.put(canonical, value);
+ // Two spellings of one option, disagreeing. Picking one would be
the coin toss this
+ // whole class exists to remove, so say so instead.
+ if (previous != null && !previous.equals(value)) {
+ throw new IllegalArgumentException(
+ "Lance namespace vended conflicting values for storage
option '"
+ + canonical + "'");
+ }
+ });
+ return result;
+ }
+
+ private static void putIfNotEmpty(Map<String, String> target, String key,
String value) {
+ if (value != null && !value.isEmpty()) {
+ target.put(key, value);
+ }
+ }
+}
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 971f461baf8..65d9ba2a8a7 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
@@ -17,61 +17,96 @@
package org.apache.doris.datasource.lance;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-/** Converts normalized Doris storage properties to Lance object-store
options. */
+/**
+ * Builds the Lance object-store options for one dataset.
+ *
+ * <p>Both 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 neither can reach a
dataset by a
+ * configuration the other never saw.
+ *
+ * <p>The option vocabulary belongs to {@link LanceStorageProvider}, chosen
from the dataset's URL
+ * the same way Lance chooses one. This class only decides what happens when
the catalog and the
+ * namespace both describe a dataset: the namespace wins, because it just
described the table.
+ */
public final class LanceStorageOptions {
- private static final Map<String, String> 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");
- }
private LanceStorageOptions() {
}
- public static Map<String, String> forJavaSdk(Map<String, String>
backendProperties) {
- Map<String, String> 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",
- String.valueOf(!Boolean.parseBoolean(usePathStyle)));
- }
+ /**
+ * Doris's own storage configuration, in the vocabulary of the provider
Lance routes
+ * {@code uri} to.
+ *
+ * <p>Used wherever no namespace is involved: the storage a namespace
client reads itself, and
+ * the {@code s3()} table-valued function.
+ */
+ public static Map<String, String> forUri(String uri,
List<StorageProperties> storageProperties) {
+ Map<String, String> result = new HashMap<>(
+
LanceStorageProvider.forDataset(uri).fromDorisProperties(storageProperties));
+ result.forEach((key, value) -> rejectUntransportable(key, value,
+ "Doris storage configuration"));
return result;
}
- /** Merge Lance storage options returned by a namespace into properties
understood by Doris BE. */
- public static Map<String, String> forBackend(Map<String, String>
staticBackendProperties,
- Map<String, String> lanceStorageOptions) {
- Map<String, String> result = new HashMap<>(staticBackendProperties);
- if (lanceStorageOptions == null || lanceStorageOptions.isEmpty()) {
+ /**
+ * The same, plus whatever a namespace vended for one table, which wins on
any option both
+ * sides name - it just described the table.
+ *
+ * <p>Putting both halves in one provider's vocabulary is what makes that
possible: otherwise
+ * two spellings of one option reach Lance as separate entries, and
object_store keeps
+ * whichever its HashMap yields last - independently in the FE and in the
BE.
+ *
+ * <p>{@code vendedOptions} may be null or empty; a namespace that
describes a table without
+ * vending storage options is ordinary, and this then degenerates to
{@link #forUri}.
+ */
+ public static Map<String, String> forVendedTable(String datasetUri,
+ List<StorageProperties> storageProperties, Map<String, String>
vendedOptions) {
+ Map<String, String> result = forUri(datasetUri, storageProperties);
+ if (vendedOptions == null || vendedOptions.isEmpty()) {
return result;
}
- S3_KEYS.forEach((dorisKey, lanceKey) -> putIfNotEmpty(result, dorisKey,
- lanceStorageOptions.get(lanceKey)));
+ vendedOptions.forEach(LanceStorageOptions::validateVendedOption);
+ // Safe to validate the vended half before normalizing:
normalizeVended only ever renames a
+ // key to one of this class's own constants or passes it through
unchanged, so it cannot
+ // introduce a NUL that the check above would have missed.
+
result.putAll(LanceStorageProvider.forDataset(datasetUri).normalizeVended(vendedOptions));
+ return 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)));
+ /**
+ * Rejects an option a namespace had no business sending.
+ *
+ * <p>Null is not expressible at all, and a NUL cannot survive the
boundary - see
+ * {@link #rejectUntransportable}.
+ */
+ private static void validateVendedOption(String key, String value) {
+ if (key == null || value == null) {
+ throw new IllegalArgumentException(
+ "Lance namespace vended a storage option with a null key
or value");
}
- return result;
+ rejectUntransportable(key, value, "Lance namespace");
}
- private static void putIfNotEmpty(Map<String, String> target, String key,
String value) {
- if (value != null && !value.isEmpty()) {
- target.put(key, value);
+ /**
+ * Rejects what cannot cross into Lance unchanged, whichever side it came
from.
+ *
+ * <p>These options are handed to lance-c and to the Lance Java SDK as C
strings, so a NUL
+ * truncates one there while the FE goes on using the whole thing, and the
two halves open the
+ * dataset with different configuration. That has to fail loudly: dropping
the option instead
+ * only moves the divergence, since a component that drops it and one that
does not disagree in
+ * exactly the same way. The BE repeats this check as its own last line of
defence.
+ */
+ private static void rejectUntransportable(String key, String value, String
source) {
+ if (key.indexOf('\0') >= 0 || value.indexOf('\0') >= 0) {
+ throw new IllegalArgumentException(source + " supplied the storage
option '"
+ + key.replace('\0', '?') + "' with a NUL in its key or
value, which cannot "
+ + "reach Lance intact");
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageProvider.java
new file mode 100644
index 00000000000..9b980351af1
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceStorageProvider.java
@@ -0,0 +1,84 @@
+// 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.apache.doris.datasource.property.storage.StorageProperties;
+
+import com.google.common.collect.ImmutableSet;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * One storage provider's option vocabulary, as Lance reads it.
+ *
+ * <p>Lance routes a dataset to a provider by URL scheme, and each provider
accepts its own set of
+ * option names. Nothing here may be applied without knowing which provider a
dataset uses: the
+ * spellings overlap but do not agree, so rewriting an option for the wrong
provider destroys it.
+ * {@code endpoint}, for instance, is what object_store's Azure parser and
Lance's OSS provider
+ * read, while its S3 parser also accepts {@code aws_endpoint} - renaming onto
the S3 spelling
+ * silently breaks the other two.
+ *
+ * <p>An implementation therefore only ever speaks for datasets {@link
#forDataset} routed to it.
+ */
+public interface LanceStorageProvider {
+
+ /** Schemes lance-io registers for its AWS provider
(rust/lance-io/.../providers.rs). */
+ Set<String> S3_SCHEMES = ImmutableSet.of("s3", "s3+ddb");
+
+ /**
+ * Converts the catalog's own storage configuration into this provider's
Lance options.
+ *
+ * <p>Takes Doris's typed properties rather than the flattened backend
map, which is only ever
+ * a re-encoding of them: {@code
AbstractS3CompatibleProperties.doBuildS3Configuration} builds
+ * that map out of the same getters read here, mixed with BE-only knobs
Lance has no use for,
+ * and flattens every configured storage into one namespace where two
S3-compatible ones would
+ * silently overwrite each other.
+ *
+ * <p>Empty when the list holds nothing this provider can read, which is
every provider but S3
+ * today - those datasets are reachable only through what a namespace
vends.
+ */
+ Map<String, String> fromDorisProperties(List<StorageProperties>
storageProperties);
+
+ /**
+ * Rewrites the options a namespace vended onto the spelling {@link
#fromDorisProperties}
+ * emits, so that the two cannot reach Lance as competing entries for one
config key.
+ *
+ * <p>Only the options Doris itself emits are rewritten. Everything else
is passed through
+ * untouched: the Lance Namespace specification describes {@code
storage_options} as
+ * configuration "passed directly to Lance", so a client cannot assume a
vocabulary beyond the
+ * one it contributes to itself.
+ */
+ Map<String, String> normalizeVended(Map<String, String> vendedOptions);
+
+ /** The provider Lance will route this dataset to. */
+ static LanceStorageProvider forDataset(String datasetUri) {
+ return S3_SCHEMES.contains(schemeOf(datasetUri))
+ ? LanceS3StorageProvider.INSTANCE :
LancePassThroughStorageProvider.INSTANCE;
+ }
+
+ static String schemeOf(String datasetUri) {
+ if (datasetUri == null) {
+ return "";
+ }
+ int separator = datasetUri.indexOf("://");
+ return separator < 0 ? "" : datasetUri.substring(0,
separator).toLowerCase(Locale.ROOT);
+ }
+}
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 22d0156085c..2e0f413c76b 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
@@ -34,33 +34,33 @@ public class LanceTableMetadata {
private final List<LanceFragmentInfo> fragments;
private final Map<String, Integer> lanceFieldIds;
private final List<LanceIndexSegmentInfo> indexSegments;
- private final Map<String, String> backendStorageOptions;
+ private final Map<String, String> lanceStorageOptions;
public static LanceTableMetadata withoutIndexSegments(String datasetUri,
long version,
Schema schema, List<LanceFragmentInfo> fragments,
- Map<String, String> backendStorageOptions) {
+ Map<String, String> lanceStorageOptions) {
return new LanceTableMetadata(datasetUri, version, schema, fragments,
- Collections.emptyMap(), Collections.emptyList(),
backendStorageOptions);
+ Collections.emptyMap(), Collections.emptyList(),
lanceStorageOptions);
}
public static LanceTableMetadata withIndexSegments(String datasetUri, long
version,
Schema schema, List<LanceFragmentInfo> fragments,
Map<String, Integer> lanceFieldIds, List<LanceIndexSegmentInfo>
indexSegments,
- Map<String, String> backendStorageOptions) {
+ Map<String, String> lanceStorageOptions) {
return new LanceTableMetadata(datasetUri, version, schema, fragments,
- lanceFieldIds, indexSegments, backendStorageOptions);
+ lanceFieldIds, indexSegments, lanceStorageOptions);
}
private LanceTableMetadata(String datasetUri, long version, Schema schema,
List<LanceFragmentInfo> fragments, Map<String, Integer>
lanceFieldIds,
- List<LanceIndexSegmentInfo> indexSegments, Map<String, String>
backendStorageOptions) {
+ List<LanceIndexSegmentInfo> indexSegments, Map<String, String>
lanceStorageOptions) {
this.datasetUri = datasetUri;
this.version = version;
this.schema = schema;
this.fragments = Collections.unmodifiableList(new
ArrayList<>(fragments));
this.lanceFieldIds = Collections.unmodifiableMap(new
HashMap<>(lanceFieldIds));
this.indexSegments = Collections.unmodifiableList(new
ArrayList<>(indexSegments));
- this.backendStorageOptions = Collections.unmodifiableMap(new
HashMap<>(backendStorageOptions));
+ this.lanceStorageOptions = Collections.unmodifiableMap(new
HashMap<>(lanceStorageOptions));
}
public String getDatasetUri() {
@@ -88,8 +88,9 @@ public class LanceTableMetadata {
return fieldId == null ? OptionalInt.empty() : OptionalInt.of(fieldId);
}
- public Map<String, String> getBackendStorageOptions() {
- return backendStorageOptions;
+ /** Lance object-store options, understood as-is by both the FE SDK and
lance-c. */
+ public Map<String, String> 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 69fd70c466b..ccae0570ec9 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
@@ -164,6 +164,11 @@ public class LanceScanNode extends FileQueryScanNode {
if (lanceSubstraitFilter.length > 0) {
params.setLanceSubstraitFilter(ByteBuffer.wrap(lanceSubstraitFilter));
}
+ // Set at ScanNode level so credentials are not serialized once per
fragment split.
+ Map<String, String> lanceStorageOptions =
plannedMetadata.getLanceStorageOptions();
+ if (!lanceStorageOptions.isEmpty()) {
+ params.setLanceStorageOptions(lanceStorageOptions);
+ }
}
@Override
@@ -409,7 +414,9 @@ public class LanceScanNode extends FileQueryScanNode {
@Override
protected Map<String, String> 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/property/metastore/AbstractLanceProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java
index c82b74fa3b0..9b6806c34e8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java
@@ -78,6 +78,15 @@ public abstract class AbstractLanceProperties extends
MetastoreProperties {
public abstract LanceNamespace createNamespace(
BufferAllocator allocator, Map<String, String> javaStorageOptions);
+ /**
+ * The URL whose storage the namespace client reads itself, which decides
how its options have
+ * to be spelled. Empty when it reads none: a REST namespace is reached
over HTTP and ignores
+ * the options entirely.
+ */
+ public String getNamespaceStorageUri() {
+ return "";
+ }
+
protected abstract void validateCatalogProperties();
public String getNamespaceParent() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceFileSystemMetastoreProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceFileSystemMetastoreProperties.java
index c6518f07f7e..19aa4afef1f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceFileSystemMetastoreProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceFileSystemMetastoreProperties.java
@@ -64,6 +64,12 @@ public class LanceFileSystemMetastoreProperties extends
AbstractLanceProperties
return warehouse;
}
+ /** A directory namespace opens the warehouse itself, so its options
follow that URL. */
+ @Override
+ public String getNamespaceStorageUri() {
+ return warehouse;
+ }
+
@Override
protected void validateCatalogProperties() {
warehouse = origProps.get(WAREHOUSE);
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 2aa7efcb7e2..ba1a47396d6 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
@@ -32,6 +32,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.LanceFragmentInfo;
+import org.apache.doris.datasource.lance.LanceStorageOptions;
import org.apache.doris.datasource.lance.source.LanceSplit;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
@@ -125,6 +126,21 @@ public class TVFScanNode extends FileQueryScanNode {
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<String, String> lanceStorageOptions =
LanceStorageOptions.forUri(
+ tableValuedFunction.getFilePath(),
+
Collections.singletonList(tableValuedFunction.getStorageProperties()));
+ if (!lanceStorageOptions.isEmpty()) {
+ params.setLanceStorageOptions(lanceStorageOptions);
+ }
+ }
+ }
+
@Override
public List<String> 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 5a61c3c3b16..0e80b40eada 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
@@ -46,6 +46,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.LanceFragmentInfo;
+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;
@@ -171,6 +172,14 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
return backendConnectProperties;
}
+ /**
+ * The typed storage configuration this function was analyzed with. Prefer
this over
+ * {@link #getBrokerDesc()}, which builds a fresh {@link
StorageProperties} on every call.
+ */
+ public StorageProperties getStorageProperties() {
+ return storageProperties;
+ }
+
public List<String> getPathPartitionKeys() {
return pathPartitionKeys;
}
@@ -527,6 +536,14 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
Map<String, String> 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.
+ Map<String, String> lanceStorageOptions =
LanceStorageOptions.forUri(
+ filePath, Collections.singletonList(storageProperties));
+ if (!lanceStorageOptions.isEmpty()) {
+
fileScanRangeParams.setLanceStorageOptions(lanceStorageOptions);
+ }
+ }
fileScanRangeParams.setFileAttributes(getFileAttributes());
ConnectContext ctx = ConnectContext.get();
fileScanRangeParams.setLoadId(ctx.queryId());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
index 61be8f0af22..2858e90f2b6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
@@ -27,6 +27,7 @@ import org.apache.doris.datasource.lance.LanceTableMetadata;
import org.apache.doris.datasource.property.storage.StorageProperties;
import org.apache.doris.thrift.TFileType;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -64,8 +65,8 @@ public class S3TableValuedFunction extends
ExternalFileTableValuedFunction {
}
if (isLanceFormat()) {
try {
- LanceTableMetadata metadata =
- LanceMetadataLoader.loadLatestForTvf(filePath,
backendConnectProperties);
+ LanceTableMetadata metadata =
LanceMetadataLoader.loadLatestForTvf(
+ filePath,
Collections.singletonList(storageProperties));
setLanceTableMetadata(metadata);
} catch (Exception e) {
throw new AnalysisException(
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 dc21fafa314..c541af25049 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.Assert;
import org.junit.Test;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
public class LanceThriftContractTest {
@@ -77,4 +80,44 @@ public class LanceThriftContractTest {
// 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<String, String> 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 0e1a787bdda..c10de9944d3 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
@@ -44,24 +44,6 @@ import java.util.concurrent.atomic.AtomicReference;
public class LanceFilesystemCatalogTest {
- @Test
- public void testMinioStorageOptionMapping() {
- Map<String, String> 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<String, String> 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 73842bf0ba6..20014bc04d5 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 class LanceSnapshotTest {
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("aws_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 @@ public class LanceSnapshotTest {
return
LanceTableMetadata.withoutIndexSegments("s3://bucket/table.lance", version,
new Schema(Collections.singletonList(field)),
Collections.singletonList(new LanceFragmentInfo(version, 1,
1)),
- Collections.singletonMap("s3.endpoint", "http://minio:9000";));
+ Collections.singletonMap("aws_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 00000000000..9e3ca4c5f4a
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceStorageOptionsTest.java
@@ -0,0 +1,314 @@
+// 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.apache.doris.datasource.property.storage.StorageProperties;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+public class LanceStorageOptionsTest {
+
+ private static final String S3_URI = "s3://warehouse/table.lance";
+
+ /**
+ * Parsed by Doris exactly as a real catalog would, so these fixtures also
have to satisfy its
+ * rules: endpoint and region are required, and the access key and secret
key must be set
+ * together or not at all.
+ */
+ private static List<StorageProperties> createAll(Map<String, String>
properties) {
+ try {
+ return StorageProperties.createAll(properties);
+ } catch (Exception e) {
+ throw new IllegalStateException("failed to parse test storage
properties", e);
+ }
+ }
+
+ private static Map<String, String> minioProperties() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("s3.endpoint", "http://minio:9000";);
+ properties.put("s3.region", "us-east-1");
+ properties.put("s3.access_key", "ak");
+ properties.put("s3.secret_key", "sk");
+ properties.put("use_path_style", "true");
+ return properties;
+ }
+
+ private static List<StorageProperties> minioCatalog() {
+ return createAll(minioProperties());
+ }
+
+ @Test
+ public void testCatalogPropertiesMapToTheCanonicalS3Spelling() {
+ Map<String, String> properties = minioProperties();
+ properties.put("s3.session_token", "token");
+
+ Map<String, String> options =
+ LanceStorageOptions.forUri(S3_URI, createAll(properties));
+ Assertions.assertEquals("ak", options.get("aws_access_key_id"));
+ Assertions.assertEquals("sk", options.get("aws_secret_access_key"));
+ Assertions.assertEquals("token", options.get("aws_session_token"));
+ Assertions.assertEquals("http://minio:9000";,
options.get("aws_endpoint"));
+ Assertions.assertEquals("us-east-1", options.get("aws_region"));
+ Assertions.assertEquals("false",
options.get("aws_virtual_hosted_style_request"));
+ // Lance refuses a plain-HTTP endpoint without this.
+ Assertions.assertEquals("true", options.get("allow_http"));
+
+ Assertions.assertEquals(
+ new TreeSet<>(Arrays.asList("aws_access_key_id",
"aws_secret_access_key",
+ "aws_session_token", "aws_endpoint", "aws_region",
+ "aws_virtual_hosted_style_request", "allow_http")),
+ new TreeSet<>(options.keySet()));
+ }
+
+ /** Doris allows anonymous access; the credential keys are then absent
rather than empty. */
+ @Test
+ public void testAnonymousAccessEmitsNoCredentials() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("s3.endpoint", "https://s3.amazonaws.com";);
+ properties.put("s3.region", "us-east-1");
+
+ Map<String, String> options =
+ LanceStorageOptions.forUri(S3_URI, createAll(properties));
+ Assertions.assertNull(options.get("aws_access_key_id"));
+ Assertions.assertNull(options.get("aws_secret_access_key"));
+ Assertions.assertNull(options.get("aws_session_token"));
+ // allow_http only makes sense for a plain-HTTP endpoint.
+ Assertions.assertNull(options.get("allow_http"));
+ Assertions.assertEquals("https://s3.amazonaws.com";,
options.get("aws_endpoint"));
+ }
+
+ /**
+ * The case this exists for: a catalog with static credentials whose
namespace also vends them,
+ * spelled the way real servers spell them. Both must land on one key with
the namespace's
+ * value, or object_store folds them onto one config key and keeps
whichever its HashMap yields
+ * last - independently in the FE and in the BE.
+ */
+ @Test
+ public void testVendedCredentialsSupersedeTheCatalogsOnS3() {
+ Map<String, String> vended = new HashMap<>();
+ vended.put("access_key_id", "vended-ak");
+ vended.put("secret_access_key", "vended-sk");
+ vended.put("session_token", "vended-token");
+ vended.put("region", "eu-west-1");
+ vended.put("virtual_hosted_style_request", "true");
+
+ Map<String, String> merged =
+ LanceStorageOptions.forVendedTable(S3_URI, minioCatalog(),
vended);
+
+ Assertions.assertEquals("vended-ak", merged.get("aws_access_key_id"));
+ Assertions.assertEquals("vended-sk",
merged.get("aws_secret_access_key"));
+ Assertions.assertEquals("vended-token",
merged.get("aws_session_token"));
+ Assertions.assertEquals("eu-west-1", merged.get("aws_region"));
+ Assertions.assertEquals("true",
merged.get("aws_virtual_hosted_style_request"));
+ // The catalog's endpoint is untouched, and no unprefixed twin
survives anywhere.
+ Assertions.assertEquals("http://minio:9000";,
merged.get("aws_endpoint"));
+ for (String alias : vended.keySet()) {
+ Assertions.assertNull(merged.get(alias), alias + " must not
survive beside its twin");
+ }
+ }
+
+ /** Every accepted spelling has to collapse, or the race just moves to the
ones missed. */
+ @Test
+ public void testEveryS3AliasCollapsesOntoOneEntry() {
+ for (String alias : new String[] {"endpoint", "endpoint_url",
"aws_endpoint",
+ "aws_endpoint_url", "ENDPOINT", "AWS_Endpoint_Url"}) {
+ Map<String, String> vended = new HashMap<>();
+ vended.put(alias, "http://127.0.0.1:9000";);
+
+ Map<String, String> merged =
+ LanceStorageOptions.forVendedTable(S3_URI, minioCatalog(),
vended);
+ long endpoints = merged.keySet().stream().filter(k ->
k.contains("endpoint")).count();
+ Assertions.assertEquals(1, endpoints, alias + " left a competing
entry");
+ Assertions.assertEquals("http://127.0.0.1:9000";,
merged.get("aws_endpoint"),
+ alias + " did not win");
+ }
+ }
+
+ /**
+ * {@code token} means an S3 session token to object_store's S3 parser and
a bearer token to
+ * its Azure one. Knowing the provider is what makes it resolvable at all.
+ */
+ @Test
+ public void testAmbiguousTokenResolvesOnceTheProviderIsKnown() {
+ Map<String, String> properties = minioProperties();
+ properties.put("s3.session_token", "static-token");
+ List<StorageProperties> catalog = createAll(properties);
+
+ Map<String, String> vended = new HashMap<>();
+ vended.put("token", "vended-token");
+
+ Map<String, String> onS3 = LanceStorageOptions.forVendedTable(S3_URI,
catalog, vended);
+ Assertions.assertEquals("vended-token", onS3.get("aws_session_token"));
+ Assertions.assertNull(onS3.get("token"));
+
+ Map<String, String> onAzure = LanceStorageOptions.forVendedTable(
+ "az://container/table.lance", catalog, vended);
+ Assertions.assertEquals("vended-token", onAzure.get("token"));
+ Assertions.assertNull(onAzure.get("aws_session_token"));
+ }
+
+ /**
+ * The regression that motivated all of this: object_store's Azure parser
reads
+ * {@code endpoint} but not {@code aws_endpoint}, and Lance's OSS provider
requires
+ * {@code endpoint} and reads {@code access_key_id}. Rewriting those onto
the S3 spellings
+ * leaves the dataset unreachable, so a non-S3 dataset must come through
untouched.
+ */
+ @Test
+ public void testNonS3DatasetsAreNeverRewritten() {
+ Map<String, String> vended = new HashMap<>();
+ vended.put("endpoint", "http://azurite:10000";);
+ vended.put("access_key_id", "vended-ak");
+ vended.put("secret_access_key", "vended-sk");
+
+ for (String uri : new String[] {"az://container/table.lance",
"abfss://fs@acct/table",
+ "oss://bucket/table.lance", "gs://bucket/table.lance",
"cos://bucket/table",
+ "file:///tmp/table.lance"}) {
+ Map<String, String> merged =
+ LanceStorageOptions.forVendedTable(uri, minioCatalog(),
vended);
+ Assertions.assertEquals(vended, merged,
+ uri + " must reach Lance exactly as the namespace wrote
it");
+ }
+ }
+
+ /**
+ * A filesystem catalog routes on its warehouse URL, so a local warehouse
must not be handed
+ * S3 options - and a REST namespace, which is reached over HTTP and reads
no storage of its
+ * own, reports no URL at all and gets nothing.
+ */
+ @Test
+ public void testCatalogWithoutAnS3UrlGetsNoS3Options() {
+ for (String uri : new String[] {"file:///warehouse/lance", "", null}) {
+ Assertions.assertTrue(
+ LanceStorageOptions.forUri(uri, minioCatalog()).isEmpty(),
+ "expected no options for " + uri);
+ }
+ }
+
+ /** Doris has no Lance vocabulary for a non-S3 provider yet, so it
contributes none. */
+ @Test
+ public void testCatalogPropertiesAreNotAppliedToANonS3Dataset() {
+ Map<String, String> merged = LanceStorageOptions.forUri(
+ "oss://bucket/table.lance", minioCatalog());
+ Assertions.assertTrue(merged.isEmpty(), "S3 credentials must not leak
onto another provider");
+ }
+
+ /**
+ * The list is not user-ordered - it follows {@code
StorageProperties.PROVIDERS} and leads with
+ * a default HDFS entry - so the S3-compatible one has to be found by
type, not by index.
+ */
+ @Test
+ public void testSelectionSkipsNonS3CompatibleEntries() {
+ List<StorageProperties> catalog = minioCatalog();
+ Assertions.assertTrue(catalog.size() > 1,
+ "expected Doris to add its default non-S3 entry ahead of the
S3 one");
+ Assertions.assertEquals("ak",
+ LanceStorageOptions.forUri(S3_URI,
catalog).get("aws_access_key_id"));
+ }
+
+ @Test
+ public void testUnknownVendedOptionsArePassedThroughOnS3Too() {
+ Map<String, String> vended = new HashMap<>();
+ // Lance reads this on its namespace-backed refresh path; Doris
assigns it no meaning.
+ vended.put("expires_at_millis", "1760000000000");
+ vended.put("azure_storage_sas_token", "sas");
+ // Empty is expressible as a C string, so it is carried rather than
second-guessed.
+ vended.put("deliberately_empty", "");
+
+ Map<String, String> merged =
+ LanceStorageOptions.forVendedTable(S3_URI,
Collections.emptyList(), vended);
+ Assertions.assertEquals(vended, merged);
+ }
+
+ @Test
+ public void testAbsentVendedOptionsLeaveCatalogOptionsIntact() {
+ Map<String, String> catalogOptions =
+ LanceStorageOptions.forUri(S3_URI, minioCatalog());
+ Assertions.assertEquals(catalogOptions,
+ LanceStorageOptions.forUri(S3_URI, minioCatalog()));
+ Assertions.assertEquals(catalogOptions,
+ LanceStorageOptions.forVendedTable(S3_URI, minioCatalog(), new
HashMap<>()));
+ }
+
+ /** A namespace contradicting itself is not something to resolve by coin
toss. */
+ @Test
+ public void testConflictingSpellingsOfOneOptionAreRejected() {
+ Map<String, String> vended = new HashMap<>();
+ vended.put("access_key_id", "one");
+ vended.put("aws_access_key_id", "another");
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> LanceStorageOptions.forVendedTable(S3_URI,
minioCatalog(), vended));
+
+ // Agreeing on the value is not a conflict.
+ Map<String, String> agreeing = new HashMap<>();
+ agreeing.put("access_key_id", "same");
+ agreeing.put("aws_access_key_id", "same");
+ Assertions.assertEquals("same", LanceStorageOptions
+ .forVendedTable(S3_URI, minioCatalog(),
agreeing).get("aws_access_key_id"));
+ }
+
+ /**
+ * The transport boundary is not specific to what a namespace vends:
Doris's own configuration
+ * reaches Lance as C strings too, so a NUL there has to fail on the same
terms.
+ */
+ @Test
+ public void testCatalogConfigurationWithEmbeddedNulIsRejected() {
+ Map<String, String> properties = minioProperties();
+ properties.put("s3.access_key", "ak\0ignored");
+
+ IllegalArgumentException thrown =
Assertions.assertThrows(IllegalArgumentException.class,
+ () -> LanceStorageOptions.forUri(S3_URI,
createAll(properties)));
+ // The message has to say which side to fix - the catalog, not the
namespace.
+ Assertions.assertTrue(thrown.getMessage().contains("Doris storage
configuration"),
+ "unexpected message: " + thrown.getMessage());
+ }
+
+ /**
+ * lance-c reads these as C strings, so a NUL truncates the option there
while the FE keeps
+ * reading the whole key. Dropping it would only move the divergence, so
it has to fail.
+ */
+ @Test
+ public void testOptionsThatCannotReachTheBackendAreRejected() {
+ Map<String, String> withNulKey = new HashMap<>();
+ withNulKey.put("bucket\0ignored", "other-bucket");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
LanceStorageOptions
+ .forVendedTable(S3_URI, Collections.emptyList(), withNulKey));
+
+ Map<String, String> withNulValue = new HashMap<>();
+ withNulValue.put("aws_region", "us-east-1\0ignored");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
LanceStorageOptions
+ .forVendedTable(S3_URI, Collections.emptyList(),
withNulValue));
+
+ Map<String, String> withNullValue = new HashMap<>();
+ withNullValue.put("aws_region", null);
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
LanceStorageOptions
+ .forVendedTable(S3_URI, Collections.emptyList(),
withNullValue));
+
+ Map<String, String> withNullKey = new HashMap<>();
+ withNullKey.put(null, "value");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
LanceStorageOptions
+ .forVendedTable(S3_URI, Collections.emptyList(), withNullKey));
+ }
+}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 93273ea75a3..f0c26fa1175 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -639,6 +639,15 @@ struct TFileScanRangeParams {
// query. Lance vector search uses one range per fragment and Doris merges
the split-local
// candidates.
38: 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.
+ // 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.
+ 39: optional map<string, string> 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 62a570c8d16..cb8003936c2 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 82cd85858f8..75f2c100793 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 {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
