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 8375559ce88 [refactor](lance) improve lance catalog implementation in
FE (#66581)
8375559ce88 is described below
commit 8375559ce889f20f208a2a68df711bf622aa390f
Author: zhangstar333 <[email protected]>
AuthorDate: Sat Aug 8 01:42:02 2026 +0800
[refactor](lance) improve lance catalog implementation in FE (#66581)
### What problem does this PR solve?
Problem Summary:
refactor some code including:
- Adds dedicated properties for filesystem and REST catalogs.
- Uses the Lance `tableExists` API for table existence checks.
- Refactors metadata loading and snapshot version resolution.
- Improves method naming and documentation for readability.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [x] No need to test or manual test. Explain why:
- [x] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../iceberg/scripts/lance_rest_server.py | 27 ++
.../datasource/lance/LanceExternalCatalog.java | 340 ++++++---------------
.../datasource/lance/LanceMetadataLoader.java | 66 ++--
.../doris/datasource/lance/LanceNamespaceName.java | 81 +++--
.../doris/datasource/lance/LanceReadOptions.java | 43 +++
.../datasource/lance/LanceSnapshotResolver.java | 20 ++
.../doris/datasource/lance/LanceVectorQuery.java | 75 +++--
.../metastore/AbstractLanceProperties.java | 104 +++++++
.../LanceFileSystemMetastoreProperties.java | 104 +++++++
.../property/metastore/LancePropertiesFactory.java | 45 +++
.../metastore/LanceRestMetastoreProperties.java | 229 ++++++++++++++
.../property/metastore/MetastoreProperties.java | 2 +-
.../doris/tablefunction/S3TableValuedFunction.java | 2 +-
.../VectorSearchTableValuedFunction.java | 14 +-
.../lance/LanceFilesystemCatalogTest.java | 52 ++--
.../datasource/lance/LanceRestCatalogTest.java | 51 ++++
.../datasource/lance/LanceVectorQueryTest.java | 25 +-
.../property/metastore/LancePropertiesTest.java | 63 ++++
18 files changed, 965 insertions(+), 378 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 7ef594a5c42..8aef278c024 100644
--- a/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py
+++ b/docker/thirdparties/docker-compose/iceberg/scripts/lance_rest_server.py
@@ -72,6 +72,16 @@ def _decode_identifier(identifier: str) -> tuple[str, ...]:
return tuple(part for part in identifier.split(DELIMITER) if part)
+def _namespace_exists(namespace: tuple[str, ...]) -> bool:
+ if not namespace:
+ return True
+ return any(
+ len(identifier) > len(namespace)
+ and identifier[: len(namespace)] == namespace
+ for identifier in TABLES
+ )
+
+
class LanceRestHandler(BaseHTTPRequestHandler):
server_version = "DorisLanceRestFixture/1.0"
@@ -147,6 +157,18 @@ class LanceRestHandler(BaseHTTPRequestHandler):
)
return
+ table_exists_match = re.fullmatch(r"/v1/table/(.+)/exists", path)
+ if table_exists_match:
+ identifier = _decode_identifier(table_exists_match.group(1))
+ if identifier in TABLES:
+ self._write_empty(200)
+ return
+ if not _namespace_exists(identifier[:-1]):
+ self._write_json(404, {"error": "namespace not found", "code":
1})
+ return
+ self._write_json(404, {"error": "table not found", "code": 4})
+ return
+
self._write_json(404, {"error": "not found", "code": 4})
def _authorized(self) -> bool:
@@ -168,6 +190,11 @@ class LanceRestHandler(BaseHTTPRequestHandler):
self.end_headers()
self.wfile.write(response)
+ def _write_empty(self, status: int) -> None:
+ self.send_response(status)
+ self.send_header("Content-Length", "0")
+ self.end_headers()
+
def log_message(self, message: str, *args: object) -> None:
print(f"{self.address_string()} - {message % args}", flush=True)
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 d943b185e7f..4339a863993 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
@@ -24,22 +24,25 @@ import org.apache.doris.datasource.CatalogProperty;
import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.InitCatalogLog;
import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.datasource.property.metastore.AbstractLanceProperties;
+import
org.apache.doris.datasource.property.metastore.LanceFileSystemMetastoreProperties;
+import
org.apache.doris.datasource.property.metastore.LanceRestMetastoreProperties;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableNotFoundException;
import org.lance.namespace.model.DescribeTableRequest;
import org.lance.namespace.model.DescribeTableResponse;
import org.lance.namespace.model.ListNamespacesRequest;
import org.lance.namespace.model.ListNamespacesResponse;
import org.lance.namespace.model.ListTablesRequest;
import org.lance.namespace.model.ListTablesResponse;
+import org.lance.namespace.model.TableExistsRequest;
-import java.net.URI;
-import java.nio.file.Path;
-import java.nio.file.Paths;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
@@ -47,35 +50,27 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
-import java.util.regex.Pattern;
/** Read-only Lance Directory or REST Namespace catalog. */
public class LanceExternalCatalog extends ExternalCatalog {
- public static final String LANCE_CATALOG_TYPE = "lance.catalog.type";
- public static final String LANCE_FILESYSTEM = "filesystem";
- public static final String LANCE_REST = "rest";
- public static final String WAREHOUSE = "warehouse";
- public static final String NAMESPACE_PARENT = "lance.namespace.parent";
- public static final String NAMESPACE_DELIMITER =
"lance.namespace.delimiter";
- public static final String ROOT_DATABASE = "lance.namespace.root_database";
- public static final String REST_URI = "lance.rest.uri";
- public static final String REST_SECURITY_TYPE = "lance.rest.security.type";
- public static final String REST_BEARER_TOKEN = "lance.rest.bearer-token";
- public static final String REST_API_KEY = "lance.rest.api-key";
- public static final String REST_HEADER_PREFIX = "lance.rest.header.";
-
- private static final String DEFAULT_DELIMITER = "$";
- private static final String DEFAULT_ROOT_DATABASE = "default";
+ public static final String LANCE_CATALOG_TYPE =
AbstractLanceProperties.LANCE_CATALOG_TYPE;
+ public static final String LANCE_FILESYSTEM =
AbstractLanceProperties.LANCE_FILESYSTEM;
+ public static final String LANCE_REST = AbstractLanceProperties.LANCE_REST;
+ public static final String WAREHOUSE =
LanceFileSystemMetastoreProperties.WAREHOUSE;
+ public static final String NAMESPACE_PARENT =
AbstractLanceProperties.NAMESPACE_PARENT;
+ public static final String NAMESPACE_DELIMITER =
AbstractLanceProperties.NAMESPACE_DELIMITER;
+ public static final String ROOT_DATABASE =
AbstractLanceProperties.ROOT_DATABASE;
+ public static final String REST_URI =
LanceRestMetastoreProperties.REST_URI;
+ public static final String REST_SECURITY_TYPE =
LanceRestMetastoreProperties.REST_SECURITY_TYPE;
+ public static final String REST_BEARER_TOKEN =
LanceRestMetastoreProperties.REST_BEARER_TOKEN;
+ public static final String REST_API_KEY =
LanceRestMetastoreProperties.REST_API_KEY;
+ public static final String REST_HEADER_PREFIX =
LanceRestMetastoreProperties.REST_HEADER_PREFIX;
+
private static final String DATABASE_NAMESPACE_DELIMITER = ".";
- private static final String REST_SECURITY_NONE = "none";
- private static final String REST_SECURITY_BEARER = "bearer";
- private static final String REST_SECURITY_API_KEY = "api_key";
- private static final Pattern HTTP_HEADER_NAME =
Pattern.compile("^[!#$%&'*+.^_`|~0-9A-Za-z-]+$");
private static final int PAGE_SIZE = 1000;
private static final long ALLOCATOR_LIMIT = 256L * 1024 * 1024;
@@ -83,7 +78,6 @@ public class LanceExternalCatalog extends ExternalCatalog {
private transient BufferAllocator allocator;
private transient List<String> parentNamespace = Collections.emptyList();
private transient String catalogType;
- private transient String namespaceDelimiter;
private transient String rootDatabase;
private transient Map<String, String> javaStorageOptions =
Collections.emptyMap();
private transient Map<String, String> backendStorageOptions =
Collections.emptyMap();
@@ -99,20 +93,19 @@ public class LanceExternalCatalog extends ExternalCatalog {
protected void initLocalObjectsImpl() {
try {
namespaceLock = new Object();
- catalogType = normalizedCatalogType();
- namespaceDelimiter =
catalogProperty.getOrDefault(NAMESPACE_DELIMITER, DEFAULT_DELIMITER);
- rootDatabase = catalogProperty.getOrDefault(ROOT_DATABASE,
DEFAULT_ROOT_DATABASE);
- parentNamespace = LanceNamespaceName.parseParent(
- catalogProperty.getOrDefault(NAMESPACE_PARENT, ""),
namespaceDelimiter);
+ AbstractLanceProperties properties = getLanceProperties();
+ catalogType = properties.getLanceCatalogType();
+ rootDatabase = properties.getRootDatabase();
+ parentNamespace = LanceNamespaceName.parseParentNamespace(
+ properties.getNamespaceParent(),
properties.getNamespaceDelimiter());
backendStorageOptions =
catalogProperty.getBackendStorageProperties();
javaStorageOptions =
LanceStorageOptions.forJavaSdk(backendStorageOptions);
allocator = new RootAllocator(ALLOCATOR_LIMIT);
- namespace = connectNamespace(catalogType, allocator,
javaStorageOptions);
+ namespace = properties.createNamespace(allocator,
javaStorageOptions);
} catch (Exception e) {
closeLanceObjects();
- throw new RuntimeException("Failed to initialize Lance " +
normalizedCatalogType() + " catalog '"
- + getName()
+ throw new RuntimeException("Failed to initialize Lance catalog '"
+ getName()
+ "': " + sanitizedRootCauseMessage(e), safeCause(e));
}
}
@@ -126,14 +119,14 @@ public class LanceExternalCatalog extends ExternalCatalog
{
return;
}
+ AbstractLanceProperties properties = getLanceProperties();
Map<String, String> storageOptions = LanceStorageOptions.forJavaSdk(
catalogProperty.getBackendStorageProperties());
- String delimiter = catalogProperty.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER);
- List<String> parent = LanceNamespaceName.parseParent(
- catalogProperty.getOrDefault(NAMESPACE_PARENT, ""), delimiter);
- String type = normalizedCatalogType();
+ List<String> parent = LanceNamespaceName.parseParentNamespace(
+ properties.getNamespaceParent(),
properties.getNamespaceDelimiter());
+ String type = properties.getLanceCatalogType();
try (BufferAllocator testAllocator = new
RootAllocator(ALLOCATOR_LIMIT)) {
- LanceNamespace testNamespace = connectNamespace(type,
testAllocator, storageOptions);
+ LanceNamespace testNamespace =
properties.createNamespace(testAllocator, storageOptions);
try {
testNamespace.listTables(new
ListTablesRequest().id(parent).limit(1));
testNamespace.listNamespaces(new
ListNamespacesRequest().id(parent).limit(1));
@@ -146,226 +139,32 @@ public class LanceExternalCatalog extends
ExternalCatalog {
}
}
- private LanceNamespace connectNamespace(String type, BufferAllocator
namespaceAllocator,
- Map<String, String> storageOptions) {
- if (LANCE_FILESYSTEM.equals(type)) {
- return connectDirectoryNamespace(namespaceAllocator,
storageOptions);
- }
- if (LANCE_REST.equals(type)) {
- return connectRestNamespace(namespaceAllocator);
- }
- throw new IllegalArgumentException("Unsupported Lance catalog type '"
+ type + "'");
- }
-
- private LanceNamespace connectDirectoryNamespace(BufferAllocator
namespaceAllocator,
- Map<String, String> storageOptions) {
- Map<String, String> namespaceProperties = new HashMap<>();
- namespaceProperties.put("root",
catalogProperty.getOrDefault(WAREHOUSE, ""));
- storageOptions.forEach((key, value) ->
namespaceProperties.put("storage." + key, value));
- return LanceNamespace.connect("dir", namespaceProperties,
namespaceAllocator);
- }
-
- private LanceNamespace connectRestNamespace(BufferAllocator
namespaceAllocator) {
- Map<String, String> namespaceProperties = new HashMap<>();
- namespaceProperties.put("uri", normalizedRestUri());
- namespaceProperties.put("delimiter",
- catalogProperty.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER));
-
- Map<String, String> properties = catalogProperty.getProperties();
- properties.forEach((key, value) -> {
- if (key.startsWith(REST_HEADER_PREFIX)) {
- namespaceProperties.put("header." +
key.substring(REST_HEADER_PREFIX.length()), value);
- }
- });
- String securityType = properties.getOrDefault(REST_SECURITY_TYPE,
REST_SECURITY_NONE)
- .trim().toLowerCase(Locale.ROOT);
- if (REST_SECURITY_BEARER.equals(securityType)) {
- namespaceProperties.put("header.Authorization", "Bearer " +
properties.get(REST_BEARER_TOKEN));
- } else if (REST_SECURITY_API_KEY.equals(securityType)) {
- namespaceProperties.put("header.x-api-key",
properties.get(REST_API_KEY));
- }
- return LanceNamespace.connect("rest", namespaceProperties,
namespaceAllocator);
- }
-
@Override
public void checkProperties() throws DdlException {
super.checkProperties();
- Map<String, String> properties = catalogProperty.getProperties();
- String type = normalizedCatalogType();
- if (!LANCE_FILESYSTEM.equals(type) && !LANCE_REST.equals(type)) {
- throw new DdlException("Property '" + LANCE_CATALOG_TYPE
- + "' must be 'filesystem' or 'rest', but was '" + type +
"'");
- }
- validateCommonProperties(properties);
- if (LANCE_FILESYSTEM.equals(type)) {
- validateFilesystemProperties(properties);
- } else {
- validateRestProperties(properties);
- }
- }
-
- private void validateCommonProperties(Map<String, String> properties)
throws DdlException {
- String delimiter = properties.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER);
- if (delimiter.isEmpty() || delimiter.indexOf('\\') >= 0) {
- throw new DdlException("Property '" + NAMESPACE_DELIMITER
- + "' cannot be empty or contain the escape character
'\\'");
- }
- String rootDb = properties.getOrDefault(ROOT_DATABASE,
DEFAULT_ROOT_DATABASE);
- if (StringUtils.isBlank(rootDb)) {
- throw new DdlException("Property '" + ROOT_DATABASE
- + "' must be non-empty");
- }
-
LanceNamespaceName.parseParent(properties.getOrDefault(NAMESPACE_PARENT, ""),
delimiter);
- }
-
- private void validateFilesystemProperties(Map<String, String> properties)
throws DdlException {
- String warehouse = properties.get(WAREHOUSE);
- if (StringUtils.isBlank(warehouse)) {
- throw new DdlException("Missing required property 'warehouse' for
Lance filesystem catalog");
- }
- validateWarehouse(warehouse);
- for (String key : properties.keySet()) {
- if (key.startsWith("lance.rest.")) {
- throw new DdlException("Property '" + key + "' is not valid
for Lance filesystem catalog");
- }
- }
- }
-
- private void validateRestProperties(Map<String, String> properties) throws
DdlException {
- if (properties.containsKey(WAREHOUSE)) {
- throw new DdlException("Property 'warehouse' is not valid for
Lance REST catalog");
- }
- String restUri = properties.get(REST_URI);
- if (StringUtils.isBlank(restUri)) {
- throw new DdlException("Missing required property '" + REST_URI +
"' for Lance REST catalog");
- }
- validateRestUri(restUri);
-
- String securityType = properties.getOrDefault(REST_SECURITY_TYPE,
REST_SECURITY_NONE)
- .trim().toLowerCase(Locale.ROOT);
- boolean bearerTokenConfigured =
properties.containsKey(REST_BEARER_TOKEN);
- boolean apiKeyConfigured = properties.containsKey(REST_API_KEY);
- boolean hasBearerToken =
StringUtils.isNotBlank(properties.get(REST_BEARER_TOKEN));
- boolean hasApiKey =
StringUtils.isNotBlank(properties.get(REST_API_KEY));
- switch (securityType) {
- case REST_SECURITY_NONE:
- if (bearerTokenConfigured || apiKeyConfigured) {
- throw new DdlException("Lance REST security type 'none'
cannot configure '"
- + REST_BEARER_TOKEN + "' or '" + REST_API_KEY +
"'");
- }
- break;
- case REST_SECURITY_BEARER:
- if (!hasBearerToken || apiKeyConfigured) {
- throw new DdlException("Lance REST security type 'bearer'
requires only '"
- + REST_BEARER_TOKEN + "'");
- }
- validateCredentialHeaderValue(REST_BEARER_TOKEN,
properties.get(REST_BEARER_TOKEN));
- break;
- case REST_SECURITY_API_KEY:
- if (!hasApiKey || bearerTokenConfigured) {
- throw new DdlException("Lance REST security type 'api_key'
requires only '"
- + REST_API_KEY + "'");
- }
- validateCredentialHeaderValue(REST_API_KEY,
properties.get(REST_API_KEY));
- break;
- default:
- throw new DdlException("Property '" + REST_SECURITY_TYPE
- + "' must be 'none', 'bearer', or 'api_key'");
- }
-
- Set<String> supportedKeys = new HashSet<>();
- Collections.addAll(supportedKeys, REST_URI, REST_SECURITY_TYPE,
REST_BEARER_TOKEN, REST_API_KEY);
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- String key = entry.getKey();
- if (!key.startsWith("lance.rest.") || supportedKeys.contains(key))
{
- continue;
- }
- if (!key.startsWith(REST_HEADER_PREFIX)) {
- throw new DdlException("Unsupported Lance REST property '" +
key + "'");
- }
- validateRestHeader(key.substring(REST_HEADER_PREFIX.length()),
entry.getValue());
- }
- }
-
- private void validateRestHeader(String headerName, String headerValue)
throws DdlException {
- if (StringUtils.isBlank(headerName) ||
!HTTP_HEADER_NAME.matcher(headerName).matches()) {
- throw new DdlException("Invalid HTTP header name in property '" +
REST_HEADER_PREFIX
- + headerName + "'");
- }
- if ("authorization".equalsIgnoreCase(headerName) ||
"x-api-key".equalsIgnoreCase(headerName)) {
- throw new DdlException("Authentication header '" + headerName + "'
must be configured through '"
- + REST_SECURITY_TYPE + "'");
- }
- if (headerValue == null || headerValue.indexOf('\r') >= 0 ||
headerValue.indexOf('\n') >= 0) {
- throw new DdlException("Invalid HTTP header value in property '" +
REST_HEADER_PREFIX
- + headerName + "'");
- }
- }
-
- private void validateCredentialHeaderValue(String propertyName, String
value) throws DdlException {
- if (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0) {
- throw new DdlException("Invalid HTTP credential value in property
'" + propertyName + "'");
- }
- }
-
- private void validateRestUri(String value) throws DdlException {
- URI uri;
try {
- uri = URI.create(value.trim());
+ AbstractLanceProperties properties = getLanceProperties();
+ LanceNamespaceName.parseParentNamespace(
+ properties.getNamespaceParent(),
properties.getNamespaceDelimiter());
} catch (IllegalArgumentException e) {
- throw new DdlException("Invalid Lance REST URI in property '" +
REST_URI + "'", e);
- }
- String scheme = uri.getScheme();
- if (scheme == null || (!("http".equalsIgnoreCase(scheme)) &&
!("https".equalsIgnoreCase(scheme)))) {
- throw new DdlException("Property '" + REST_URI + "' must use http
or https");
- }
- if (StringUtils.isBlank(uri.getRawAuthority()) || uri.getRawUserInfo()
!= null
- || uri.getRawQuery() != null || uri.getRawFragment() != null) {
- throw new DdlException("Property '" + REST_URI
- + "' must contain an authority and cannot contain
user-info, query, or fragment");
- }
- }
-
- private String normalizedCatalogType() {
- return catalogProperty.getOrDefault(LANCE_CATALOG_TYPE,
LANCE_FILESYSTEM)
- .trim().toLowerCase(Locale.ROOT);
- }
-
- private String normalizedRestUri() {
- String uri = catalogProperty.getOrDefault(REST_URI, "").trim();
- while (uri.endsWith("/")) {
- uri = uri.substring(0, uri.length() - 1);
+ throw new DdlException(e.getMessage(), e);
}
- return uri;
}
- private void validateWarehouse(String warehouse) throws DdlException {
- URI uri;
- try {
- uri = URI.create(warehouse);
- } catch (IllegalArgumentException e) {
- throw new DdlException("Invalid Lance warehouse URI: " +
warehouse, e);
- }
- if (uri.getScheme() == null) {
- Path path = Paths.get(warehouse);
- if (!path.isAbsolute()) {
- throw new DdlException("Local Lance warehouse must be an
absolute path: " + warehouse);
- }
- return;
- }
- String scheme = uri.getScheme().toLowerCase();
- if (!"file".equals(scheme) && !"s3".equals(scheme)) {
- throw new DdlException("Unsupported Lance filesystem warehouse
scheme '" + scheme
- + "'; first phase supports local/file and s3");
- }
+ private AbstractLanceProperties getLanceProperties() {
+ return (AbstractLanceProperties)
catalogProperty.getMetastoreProperties();
}
@Override
protected List<String> listDatabaseNames() {
makeSureInitialized();
+
+ // The configured root database represents the empty relative Lance
namespace.
LinkedHashSet<String> databases = new LinkedHashSet<>();
databases.add(rootDatabase);
+ // Breadth-first traversal starts at the catalog's configured parent
namespace.
+ // Queue entries remain relative so they can be exposed as Doris
database names.
Queue<List<String>> queue = new ArrayDeque<>();
queue.add(Collections.emptyList());
Set<List<String>> visited = new HashSet<>();
@@ -374,17 +173,30 @@ public class LanceExternalCatalog extends ExternalCatalog
{
if (!visited.add(relativeParent)) {
continue;
}
- for (String child :
listChildNamespaces(fullNamespace(relativeParent))) {
+
+ // The Lance API expects a full namespace, including the
configured parent.
+ List<String> fullParentNamespace =
buildFullNamespace(relativeParent);
+ for (String child : listChildNamespaces(fullParentNamespace)) {
List<String> relativeChild = new ArrayList<>(relativeParent);
relativeChild.add(child);
- databases.add(LanceNamespaceName.encode(
+
+ // Doris exposes each hierarchical relative namespace as one
flat database name.
+ databases.add(LanceNamespaceName.namespaceToDorisDatabaseName(
relativeChild, DATABASE_NAMESPACE_DELIMITER,
rootDatabase));
+
+ // Visit this child later to discover namespaces nested below
it.
queue.add(relativeChild);
}
}
return new ArrayList<>(databases);
}
+ /**
+ * Lists all direct child namespace names under the given full Lance
namespace.
+ *
+ * <p>Each request asks for at most {@link #PAGE_SIZE} children. If Lance
returns a
+ * page token, this method keeps requesting subsequent pages until all
children are collected.
+ */
private List<String> listChildNamespaces(List<String> namespaceId) {
List<String> result = new ArrayList<>();
String pageToken = null;
@@ -410,8 +222,9 @@ public class LanceExternalCatalog extends ExternalCatalog {
protected List<String> listTableNamesFromRemote(SessionContext ctx, String
dbName) {
makeSureInitialized();
try {
- List<String> namespaceId = fullNamespace(
- LanceNamespaceName.decode(dbName,
DATABASE_NAMESPACE_DELIMITER, rootDatabase));
+ List<String> relativeNamespace =
LanceNamespaceName.dorisDatabaseNameToNamespace(
+ dbName, DATABASE_NAMESPACE_DELIMITER, rootDatabase);
+ List<String> namespaceId = buildFullNamespace(relativeNamespace);
List<String> result = new ArrayList<>();
String pageToken = null;
Set<String> consumedTokens = new HashSet<>();
@@ -437,11 +250,21 @@ public class LanceExternalCatalog extends ExternalCatalog
{
@Override
public boolean tableExist(SessionContext ctx, String dbName, String
tblName) {
+ makeSureInitialized();
try {
- describeTable(dbName, tblName);
+ List<String> relativeNamespace =
LanceNamespaceName.dorisDatabaseNameToNamespace(
+ dbName, DATABASE_NAMESPACE_DELIMITER, rootDatabase);
+ List<String> tableId = buildFullNamespace(relativeNamespace);
+ tableId.add(tblName);
+ TableExistsRequest request = new TableExistsRequest().id(tableId);
+ synchronized (namespaceLock) {
+ namespace.tableExists(request);
+ }
return true;
- } catch (RuntimeException e) {
+ } catch (TableNotFoundException | NamespaceNotFoundException e) {
return false;
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
}
}
@@ -480,13 +303,13 @@ public class LanceExternalCatalog extends ExternalCatalog
{
throw new IllegalArgumentException(
"Cannot parse Lance FOR TIME AS OF value '" +
snapshot.getValue() + "'");
}
- version = LanceMetadataLoader.resolveVersionAtOrBefore(
+ version = LanceSnapshotResolver.getVersionAtOrBefore(
datasetUri, storageOptions, timestamp, allocator);
}
- return LanceMetadataLoader.load(datasetUri, storageOptions,
+ return LanceMetadataLoader.loadVersion(datasetUri,
storageOptions,
tableBackendStorageOptions, version, allocator);
}
- return LanceMetadataLoader.load(datasetUri, storageOptions,
+ return LanceMetadataLoader.loadLatest(datasetUri, storageOptions,
tableBackendStorageOptions, allocator);
} catch (Exception e) {
throw new RuntimeException("Failed to load Lance table metadata
for " + dbName + "." + tableName
@@ -496,8 +319,9 @@ public class LanceExternalCatalog extends ExternalCatalog {
private DescribeTableResponse describeTable(String dbName, String
tableName) {
try {
- List<String> tableId = fullNamespace(
- LanceNamespaceName.decode(dbName,
DATABASE_NAMESPACE_DELIMITER, rootDatabase));
+ List<String> relativeNamespace =
LanceNamespaceName.dorisDatabaseNameToNamespace(
+ dbName, DATABASE_NAMESPACE_DELIMITER, rootDatabase);
+ List<String> tableId = buildFullNamespace(relativeNamespace);
tableId.add(tableName);
DescribeTableRequest request = new
DescribeTableRequest().id(tableId).withTableUri(true)
.vendCredentials(LANCE_REST.equals(catalogType));
@@ -509,7 +333,15 @@ public class LanceExternalCatalog extends ExternalCatalog {
}
}
- private List<String> fullNamespace(List<String> relativeNamespace) {
+ /**
+ * Prepends the configured parent namespace to a namespace relative to
this catalog.
+ *
+ * <p>For example, if {@code parentNamespace} is {@code [company,
analytics]} and
+ * {@code relativeNamespace} is {@code [sales, daily]}, this method returns
+ * {@code [company, analytics, sales, daily]}. The returned list is a new
mutable list;
+ * neither input list is modified.
+ */
+ private List<String> buildFullNamespace(List<String> relativeNamespace) {
List<String> result = new ArrayList<>(parentNamespace.size() +
relativeNamespace.size());
result.addAll(parentNamespace);
result.addAll(relativeNamespace);
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 76a59e5f318..a0502fee873 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
@@ -21,7 +21,6 @@ import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.lance.Dataset;
import org.lance.Fragment;
-import org.lance.ReadOptions;
import java.util.ArrayList;
import java.util.List;
@@ -31,54 +30,58 @@ import java.util.OptionalLong;
/** Loads one fixed Lance dataset snapshot through the Lance Java SDK. */
public final class LanceMetadataLoader {
private static final long ALLOCATOR_LIMIT = 256L * 1024 * 1024;
- private static final long METADATA_CACHE_SIZE = 64L * 1024 * 1024;
private LanceMetadataLoader() {
}
/**
- * Load metadata for a directly-addressed dataset, using Doris backend
storage properties.
- * This overload owns a short-lived allocator and is suitable for an S3
TVF.
+ * Loads the latest metadata for a directly addressed dataset and owns a
short-lived allocator.
+ *
+ * <p>Called by {@link
org.apache.doris.tablefunction.S3TableValuedFunction} when reading a
+ * Lance dataset through an S3 TVF.
*/
- public static LanceTableMetadata load(String datasetUri, Map<String,
String> backendStorageOptions)
+ public static LanceTableMetadata loadLatestForTvf(
+ String datasetUri, Map<String, String> backendStorageOptions)
throws Exception {
try (BufferAllocator allocator = new RootAllocator(ALLOCATOR_LIMIT)) {
- return load(datasetUri,
LanceStorageOptions.forJavaSdk(backendStorageOptions),
+ return loadLatest(datasetUri,
LanceStorageOptions.forJavaSdk(backendStorageOptions),
backendStorageOptions, allocator);
}
}
/**
- * Load metadata with an allocator and Java storage options already owned
by a catalog.
- * Schema, version and fragments are all read from the same opened dataset
snapshot.
+ * Loads the latest metadata with the allocator and storage options owned
by a catalog.
+ *
+ * <p>Called by
+ * {@link LanceExternalCatalog#loadTableMetadata(String, String,
java.util.Optional)} when no
+ * time-travel version is requested. Schema, version, and fragments are
read from the same
+ * opened dataset snapshot.
*/
- public static LanceTableMetadata load(String datasetUri, Map<String,
String> javaStorageOptions,
+ public static LanceTableMetadata loadLatest(String datasetUri, Map<String,
String> javaStorageOptions,
Map<String, String> backendStorageOptions, BufferAllocator
allocator) throws Exception {
- return load(datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.empty(), allocator);
+ return loadInternal(
+ datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.empty(), allocator);
}
- /** Load metadata from an explicitly selected Lance version. */
- public static LanceTableMetadata load(String datasetUri, Map<String,
String> javaStorageOptions,
+ /**
+ * Loads metadata from an explicitly selected Lance version.
+ *
+ * <p>Called by
+ * {@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 {
- return load(datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.of(version), allocator);
+ return loadInternal(
+ datasetUri, javaStorageOptions, backendStorageOptions,
OptionalLong.of(version), allocator);
}
- /** Resolve the latest Lance version whose commit time does not exceed the
requested timestamp. */
- public static long resolveVersionAtOrBefore(String datasetUri, Map<String,
String> javaStorageOptions,
- long timestampMillis, BufferAllocator allocator) throws Exception {
- ReadOptions readOptions = readOptions(javaStorageOptions,
OptionalLong.empty());
- try (Dataset dataset =
Dataset.open().allocator(allocator).uri(datasetUri)
- .readOptions(readOptions).build()) {
- return
LanceSnapshotResolver.versionAtOrBefore(dataset.listVersions(),
timestampMillis);
- }
- }
-
- private static LanceTableMetadata load(String datasetUri, Map<String,
String> javaStorageOptions,
+ /** 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,
BufferAllocator allocator) throws Exception {
- ReadOptions readOptions = readOptions(javaStorageOptions, version);
try (Dataset dataset =
Dataset.open().allocator(allocator).uri(datasetUri)
- .readOptions(readOptions).build()) {
+ .readOptions(LanceReadOptions.build(javaStorageOptions,
version)).build()) {
long resolvedVersion = dataset.version();
List<LanceTableMetadata.LanceFragmentInfo> fragments = new
ArrayList<>();
for (Fragment fragment : dataset.getFragments()) {
@@ -89,15 +92,4 @@ public final class LanceMetadataLoader {
backendStorageOptions);
}
}
-
- private static ReadOptions readOptions(Map<String, String>
javaStorageOptions, OptionalLong version) {
- ReadOptions.Builder builder = new ReadOptions.Builder()
- .setStorageOptions(javaStorageOptions)
- .setIndexCacheSizeBytes(0)
- .setMetadataCacheSizeBytes(METADATA_CACHE_SIZE);
- if (version.isPresent()) {
- builder.setVersion(version.getAsLong());
- }
- return builder.build();
- }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceName.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceName.java
index bea8f09ad94..caa014afb1f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceName.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceName.java
@@ -23,14 +23,32 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-/** Reversible mapping between a hierarchical Lance namespace and one Doris
database name. */
+/**
+ * Converts between a hierarchical Lance namespace and the flat database name
exposed by Doris.
+ *
+ * <p>The input namespace is relative to the optional parent namespace
configured on the catalog.
+ * An empty relative namespace maps to {@code rootDatabase}. Other namespace
components are joined
+ * with {@code delimiter}; delimiter occurrences and backslashes inside a
component are escaped so
+ * the conversion remains reversible. A one-level namespace equal to {@code
rootDatabase} receives
+ * an additional leading escape to distinguish it from the empty root
namespace.
+ *
+ * <p>This class only converts names. It does not access the remote Lance
namespace service.
+ */
final class LanceNamespaceName {
private static final char ESCAPE = '\\';
private LanceNamespaceName() {
}
- static String encode(List<String> relativeNamespace, String delimiter,
String rootDatabase) {
+ /**
+ * Converts relative Lance namespace components to one Doris database name.
+ *
+ * <p>For example, with {@code "."} as the delimiter,
+ * {@code [company.analytics, raw]} becomes {@code company\.analytics.raw}.
+ * An empty namespace becomes {@code rootDatabase}.
+ */
+ static String namespaceToDorisDatabaseName(
+ List<String> relativeNamespace, String delimiter, String
rootDatabase) {
if (relativeNamespace.isEmpty()) {
return rootDatabase;
}
@@ -39,32 +57,58 @@ final class LanceNamespaceName {
if (i > 0) {
result.append(delimiter);
}
- result.append(escape(relativeNamespace.get(i), delimiter));
+ result.append(escapeNamespaceComponent(relativeNamespace.get(i),
delimiter));
}
// The local root database represents the empty Lance namespace. Escape
// the rare, colliding one-level namespace so the mapping stays
reversible.
return rootDatabase.contentEquals(result) ? ESCAPE + result.toString()
: result.toString();
}
- static List<String> decode(String database, String delimiter, String
rootDatabase) throws DdlException {
- if (rootDatabase.equals(database)) {
+ /** Escapes backslashes and delimiters occurring inside one namespace
component. */
+ private static String escapeNamespaceComponent(String component, String
delimiter) {
+ return component.replace("\\", "\\\\").replace(delimiter, "\\" +
delimiter);
+ }
+
+ /**
+ * Restores the relative Lance namespace represented by one Doris database
name.
+ *
+ * <p>For example, with {@code "."} as the delimiter,
+ * {@code company\.analytics.raw} becomes {@code [company.analytics, raw]}.
+ * A database name equal to {@code rootDatabase} becomes an empty
namespace.
+ */
+ static List<String> dorisDatabaseNameToNamespace(
+ String databaseName, String delimiter, String rootDatabase) throws
DdlException {
+ if (rootDatabase.equals(databaseName)) {
return Collections.emptyList();
}
- if (database.length() > 1 && database.charAt(0) == ESCAPE
- && rootDatabase.equals(database.substring(1))) {
- return splitEscaped(database.substring(1), delimiter);
+ if (databaseName.length() > 1 && databaseName.charAt(0) == ESCAPE
+ && rootDatabase.equals(databaseName.substring(1))) {
+ return parseEscapedNamespace(databaseName.substring(1), delimiter);
}
- return splitEscaped(database, delimiter);
+ return parseEscapedNamespace(databaseName, delimiter);
}
- static List<String> parseParent(String value, String delimiter) throws
DdlException {
+ /**
+ * Parses the escaped parent namespace configured in the catalog
properties.
+ *
+ * <p>For example, with {@code "."} as the delimiter,
+ * {@code company.analytics} becomes {@code [company, analytics]}.
+ */
+ static List<String> parseParentNamespace(String value, String delimiter)
throws DdlException {
if (value == null || value.isEmpty()) {
return Collections.emptyList();
}
- return splitEscaped(value, delimiter);
+ return parseEscapedNamespace(value, delimiter);
}
- private static List<String> splitEscaped(String value, String delimiter)
throws DdlException {
+ /**
+ * Splits an escaped namespace and rejects invalid escapes or empty
components.
+ *
+ * <p>For example, with {@code "."} as the delimiter, {@code
company\.analytics.raw}
+ * is parsed as {@code [company.analytics, raw]}.
+ */
+ private static List<String> parseEscapedNamespace(String value, String
delimiter)
+ throws DdlException {
List<String> result = new ArrayList<>();
StringBuilder current = new StringBuilder();
for (int i = 0; i < value.length();) {
@@ -80,22 +124,19 @@ final class LanceNamespaceName {
throw new DdlException("Invalid escape in Lance namespace
'" + value + "'");
}
} else if (value.startsWith(delimiter, i)) {
- addComponent(result, current, value);
+ addNamespaceComponent(result, current, value);
i += delimiter.length();
} else {
current.append(value.charAt(i++));
}
}
- addComponent(result, current, value);
+ addNamespaceComponent(result, current, value);
return result;
}
- private static String escape(String component, String delimiter) {
- return component.replace("\\", "\\\\").replace(delimiter, "\\" +
delimiter);
- }
-
- private static void addComponent(List<String> result, StringBuilder
current, String encoded)
- throws DdlException {
+ /** Adds one parsed component while enforcing that Lance namespace
components are non-empty. */
+ private static void addNamespaceComponent(
+ List<String> result, StringBuilder current, String encoded) throws
DdlException {
if (current.length() == 0) {
throw new DdlException("Empty Lance namespace component in '" +
encoded + "'");
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceReadOptions.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceReadOptions.java
new file mode 100644
index 00000000000..432426b2c95
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceReadOptions.java
@@ -0,0 +1,43 @@
+// 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.lance.ReadOptions;
+
+import java.util.Map;
+import java.util.OptionalLong;
+
+/** Builds consistently configured SDK read options for Lance dataset metadata
access. */
+final class LanceReadOptions {
+ private static final long METADATA_CACHE_SIZE = 64L * 1024 * 1024;
+
+ private LanceReadOptions() {
+ }
+
+ /** Called by metadata loading and timestamp-based snapshot resolution. */
+ static ReadOptions build(Map<String, String> javaStorageOptions,
OptionalLong version) {
+ ReadOptions.Builder builder = new ReadOptions.Builder()
+ .setStorageOptions(javaStorageOptions)
+ .setIndexCacheSizeBytes(0)
+ .setMetadataCacheSizeBytes(METADATA_CACHE_SIZE);
+ if (version.isPresent()) {
+ builder.setVersion(version.getAsLong());
+ }
+ return builder.build();
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSnapshotResolver.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSnapshotResolver.java
index bf194ad0292..7398367c98a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSnapshotResolver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSnapshotResolver.java
@@ -17,11 +17,15 @@
package org.apache.doris.datasource.lance;
+import org.apache.arrow.memory.BufferAllocator;
+import org.lance.Dataset;
import org.lance.Version;
import java.time.Instant;
import java.util.Comparator;
import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
/** Resolves Doris time-travel selectors to immutable Lance version IDs. */
final class LanceSnapshotResolver {
@@ -43,6 +47,22 @@ final class LanceSnapshotResolver {
return version;
}
+ /**
+ * Gets the latest Lance version whose commit time does not exceed the
requested timestamp.
+ *
+ * <p>Called by
+ * {@link LanceExternalCatalog#loadTableMetadata(String, String,
java.util.Optional)} to resolve
+ * a {@code FOR TIME AS OF} clause before loading the selected metadata
snapshot.
+ */
+ static long getVersionAtOrBefore(String datasetUri, Map<String, String>
javaStorageOptions,
+ long timestampMillis, BufferAllocator allocator) throws Exception {
+ try (Dataset dataset =
Dataset.open().allocator(allocator).uri(datasetUri)
+ .readOptions(LanceReadOptions.build(javaStorageOptions,
OptionalLong.empty())).build()) {
+ return versionAtOrBefore(dataset.listVersions(), timestampMillis);
+ }
+ }
+
+ /** Selects a version from the version list fetched by {@link
#getVersionAtOrBefore}. */
static long versionAtOrBefore(List<Version> versions, long
timestampMillis) {
Instant requestedTime = Instant.ofEpochMilli(timestampMillis);
return versions.stream()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceVectorQuery.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceVectorQuery.java
index 27bf7c67767..5b5ebd8f88c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceVectorQuery.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceVectorQuery.java
@@ -44,10 +44,10 @@ public final class LanceVectorQuery {
}
/**
- * Resolve a column using Doris' case-insensitive identifier behavior
while preserving the
+ * Find a vector column using Doris' case-insensitive identifier behavior
while preserving the
* physical Lance field name sent to the backend.
*/
- public static Field resolveVectorField(Schema schema, String column)
throws AnalysisException {
+ public static Field findVectorColumnField(Schema schema, String column)
throws AnalysisException {
Field match = null;
for (Field field : schema.getFields()) {
if (field.getName().equalsIgnoreCase(column)) {
@@ -64,7 +64,19 @@ public final class LanceVectorQuery {
return match;
}
- public static TSearchVector encode(Field field, String json) throws
AnalysisException {
+ public static TSearchVector parseAndEncodeQueryVector(Field field, String
json)
+ throws AnalysisException {
+ VectorEncodingSpec encodingSpec = analyzeVectorField(field);
+ JsonArray values = parseQueryVector(json, field,
encodingSpec.dimension);
+ byte[] encodedValues = encodeQueryVectorValues(field, values,
encodingSpec);
+
+ return new TSearchVector()
+ .setElementType(encodingSpec.elementType)
+ .setDimension(encodingSpec.dimension)
+ .setValues(encodedValues);
+ }
+
+ private static VectorEncodingSpec analyzeVectorField(Field field) throws
AnalysisException {
if (hasExtension(field) || field.getDictionary() != null
|| field.getType().getTypeID() !=
ArrowType.ArrowTypeID.FixedSizeList
|| field.getChildren().size() != 1) {
@@ -81,30 +93,36 @@ public final class LanceVectorQuery {
if (hasExtension(elementField) || elementField.getDictionary() !=
null) {
throw unsupportedVectorType(field);
}
- ElementEncoding encoding = elementEncoding(field,
elementField.getType());
+ return determineVectorEncoding(field, elementField.getType(),
dimension);
+ }
- JsonArray values;
+ private static JsonArray parseQueryVector(String json, Field field, int
dimension)
+ throws AnalysisException {
try {
JsonElement root = JsonParser.parseString(json);
if (!root.isJsonArray()) {
throw new AnalysisException("'query_vector' must be a JSON
array");
}
- values = root.getAsJsonArray();
+ JsonArray values = root.getAsJsonArray();
+ if (values.size() != dimension) {
+ throw new AnalysisException("Query vector dimension " +
values.size()
+ + " does not match Lance column '" + field.getName()
+ + "' dimension " + dimension);
+ }
+ return values;
} catch (AnalysisException e) {
throw e;
} catch (RuntimeException e) {
throw new AnalysisException("Invalid 'query_vector' JSON: " +
e.getMessage(), e);
}
- if (values.size() != dimension) {
- throw new AnalysisException("Query vector dimension " +
values.size()
- + " does not match Lance column '" + field.getName()
- + "' dimension " + dimension);
- }
+ }
- long encodedSize = (long) dimension * encoding.byteWidth;
+ private static byte[] encodeQueryVectorValues(Field field, JsonArray
values,
+ VectorEncodingSpec encodingSpec) throws AnalysisException {
+ long encodedSize = (long) encodingSpec.dimension *
encodingSpec.byteWidth;
if (encodedSize > Integer.MAX_VALUE) {
throw new AnalysisException("Lance vector column '" +
field.getName()
- + "' is too large to encode: " + dimension + " elements");
+ + "' is too large to encode: " + encodingSpec.dimension +
" elements");
}
ByteBuffer buffer = ByteBuffer.allocate((int)
encodedSize).order(ByteOrder.LITTLE_ENDIAN);
for (int i = 0; i < values.size(); ++i) {
@@ -112,34 +130,35 @@ public final class LanceVectorQuery {
if (!value.isJsonPrimitive() ||
!value.getAsJsonPrimitive().isNumber()) {
throw new AnalysisException("Query vector element " + i + "
must be a number");
}
- encodeElement(value.getAsJsonPrimitive(), i, encoding.elementType,
buffer);
+ writeQueryVectorElement(
+ value.getAsJsonPrimitive(), i, encodingSpec.elementType,
buffer);
}
- return new TSearchVector()
- .setElementType(encoding.elementType)
- .setDimension(dimension)
- .setValues(buffer.array());
+ return buffer.array();
}
- private static ElementEncoding elementEncoding(Field vectorField,
ArrowType elementType)
- throws AnalysisException {
+ private static VectorEncodingSpec determineVectorEncoding(
+ Field vectorField, ArrowType elementType, int dimension) throws
AnalysisException {
switch (elementType.getTypeID()) {
case FloatingPoint:
FloatingPointPrecision precision =
((ArrowType.FloatingPoint) elementType).getPrecision();
switch (precision) {
case HALF:
- return new ElementEncoding(TVectorElementType.FLOAT16,
Short.BYTES);
+ return new VectorEncodingSpec(
+ dimension, TVectorElementType.FLOAT16,
Short.BYTES);
case SINGLE:
- return new ElementEncoding(TVectorElementType.FLOAT32,
Float.BYTES);
+ return new VectorEncodingSpec(
+ dimension, TVectorElementType.FLOAT32,
Float.BYTES);
case DOUBLE:
- return new ElementEncoding(TVectorElementType.FLOAT64,
Double.BYTES);
+ return new VectorEncodingSpec(
+ dimension, TVectorElementType.FLOAT64,
Double.BYTES);
default:
throw unsupportedVectorType(vectorField);
}
case Int:
ArrowType.Int integer = (ArrowType.Int) elementType;
if (integer.getBitWidth() == Byte.SIZE) {
- return new ElementEncoding(integer.getIsSigned()
+ return new VectorEncodingSpec(dimension,
integer.getIsSigned()
? TVectorElementType.INT8 :
TVectorElementType.UINT8, Byte.BYTES);
}
throw unsupportedVectorType(vectorField);
@@ -148,7 +167,7 @@ public final class LanceVectorQuery {
}
}
- private static void encodeElement(JsonPrimitive value, int index,
+ private static void writeQueryVectorElement(JsonPrimitive value, int index,
TVectorElementType elementType, ByteBuffer buffer) throws
AnalysisException {
try {
switch (elementType) {
@@ -226,11 +245,13 @@ public final class LanceVectorQuery {
+ " is not representable as " + type.name().toLowerCase());
}
- private static class ElementEncoding {
+ private static class VectorEncodingSpec {
+ private final int dimension;
private final TVectorElementType elementType;
private final int byteWidth;
- private ElementEncoding(TVectorElementType elementType, int byteWidth)
{
+ private VectorEncodingSpec(int dimension, TVectorElementType
elementType, int byteWidth) {
+ this.dimension = dimension;
this.elementType = elementType;
this.byteWidth = byteWidth;
}
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
new file mode 100644
index 00000000000..c82b74fa3b0
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java
@@ -0,0 +1,104 @@
+// 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.property.metastore;
+
+import org.apache.doris.foundation.property.ConnectorProperty;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.commons.lang3.StringUtils;
+import org.lance.namespace.LanceNamespace;
+
+import java.util.Map;
+
+/** Common properties shared by Lance filesystem and REST namespace catalogs.
*/
+public abstract class AbstractLanceProperties extends MetastoreProperties {
+ public static final String LANCE_CATALOG_TYPE = "lance.catalog.type";
+ public static final String LANCE_FILESYSTEM = "filesystem";
+ public static final String LANCE_REST = "rest";
+ public static final String NAMESPACE_PARENT = "lance.namespace.parent";
+ public static final String NAMESPACE_DELIMITER =
"lance.namespace.delimiter";
+ public static final String ROOT_DATABASE = "lance.namespace.root_database";
+
+ public static final String DEFAULT_DELIMITER = "$";
+ public static final String DEFAULT_ROOT_DATABASE = "default";
+
+ @ConnectorProperty(
+ names = {NAMESPACE_PARENT},
+ required = false,
+ description = "The optional Lance namespace under which Doris
databases are exposed.")
+ private String namespaceParent = "";
+
+ @ConnectorProperty(
+ names = {NAMESPACE_DELIMITER},
+ required = false,
+ description = "The delimiter used in an escaped parent namespace.
Default: $."
+ )
+ private String namespaceDelimiter = DEFAULT_DELIMITER;
+
+ @ConnectorProperty(
+ names = {ROOT_DATABASE},
+ required = false,
+ description = "The Doris database name representing the catalog's
root namespace. Default: default."
+ )
+ private String rootDatabase = DEFAULT_ROOT_DATABASE;
+
+ protected AbstractLanceProperties(Map<String, String> props) {
+ super(Type.LANCE, props);
+ }
+
+ @Override
+ public final void initNormalizeAndCheckProps() {
+ super.initNormalizeAndCheckProps();
+ // Preserve namespace whitespace and explicit empty values because
both are meaningful
+ // to the escaping rules and must be validated rather than silently
replaced by defaults.
+ namespaceParent = origProps.getOrDefault(NAMESPACE_PARENT, "");
+ namespaceDelimiter = origProps.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER);
+ rootDatabase = origProps.getOrDefault(ROOT_DATABASE,
DEFAULT_ROOT_DATABASE);
+ validateCommonProperties();
+ validateCatalogProperties();
+ }
+
+ public abstract String getLanceCatalogType();
+
+ public abstract LanceNamespace createNamespace(
+ BufferAllocator allocator, Map<String, String> javaStorageOptions);
+
+ protected abstract void validateCatalogProperties();
+
+ public String getNamespaceParent() {
+ return namespaceParent;
+ }
+
+ public String getNamespaceDelimiter() {
+ return namespaceDelimiter;
+ }
+
+ public String getRootDatabase() {
+ return rootDatabase;
+ }
+
+ private void validateCommonProperties() {
+ if (namespaceDelimiter.isEmpty() || namespaceDelimiter.indexOf('\\')
>= 0) {
+ throw new IllegalArgumentException("Property '" +
NAMESPACE_DELIMITER
+ + "' cannot be empty or contain the escape character
'\\'");
+ }
+ if (StringUtils.isBlank(rootDatabase)) {
+ throw new IllegalArgumentException("Property '" + ROOT_DATABASE +
"' must be non-empty");
+ }
+ }
+}
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
new file mode 100644
index 00000000000..c6518f07f7e
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceFileSystemMetastoreProperties.java
@@ -0,0 +1,104 @@
+// 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.property.metastore;
+
+import org.apache.doris.foundation.property.ConnectorProperty;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.commons.lang3.StringUtils;
+import org.lance.namespace.LanceNamespace;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/** Properties for a Lance directory namespace backed by a filesystem
warehouse. */
+public class LanceFileSystemMetastoreProperties extends
AbstractLanceProperties {
+ public static final String WAREHOUSE = "warehouse";
+
+ @ConnectorProperty(
+ names = {WAREHOUSE},
+ required = false,
+ description = "The local, file, or S3 warehouse containing Lance
datasets."
+ )
+ private String warehouse;
+
+ public LanceFileSystemMetastoreProperties(Map<String, String> props) {
+ super(props);
+ }
+
+ @Override
+ public String getLanceCatalogType() {
+ return LANCE_FILESYSTEM;
+ }
+
+ @Override
+ public LanceNamespace createNamespace(
+ BufferAllocator allocator, Map<String, String> javaStorageOptions)
{
+ Map<String, String> namespaceProperties = new HashMap<>();
+ namespaceProperties.put("root", warehouse);
+ javaStorageOptions.forEach(
+ (key, value) -> namespaceProperties.put("storage." + key,
value));
+ return LanceNamespace.connect("dir", namespaceProperties, allocator);
+ }
+
+ public String getWarehouse() {
+ return warehouse;
+ }
+
+ @Override
+ protected void validateCatalogProperties() {
+ warehouse = origProps.get(WAREHOUSE);
+ if (StringUtils.isBlank(warehouse)) {
+ throw new IllegalArgumentException(
+ "Missing required property 'warehouse' for Lance
filesystem catalog");
+ }
+ validateWarehouse(warehouse);
+ for (String key : origProps.keySet()) {
+ if (key.startsWith("lance.rest.")) {
+ throw new IllegalArgumentException(
+ "Property '" + key + "' is not valid for Lance
filesystem catalog");
+ }
+ }
+ }
+
+ private static void validateWarehouse(String warehouse) {
+ final URI uri;
+ try {
+ uri = URI.create(warehouse);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Invalid Lance warehouse URI: "
+ warehouse, e);
+ }
+ if (uri.getScheme() == null) {
+ Path path = Paths.get(warehouse);
+ if (!path.isAbsolute()) {
+ throw new IllegalArgumentException(
+ "Local Lance warehouse must be an absolute path: " +
warehouse);
+ }
+ return;
+ }
+ String scheme = uri.getScheme().toLowerCase(Locale.ROOT);
+ if (!"file".equals(scheme) && !"s3".equals(scheme)) {
+ throw new IllegalArgumentException("Unsupported Lance filesystem
warehouse scheme '" + scheme
+ + "'; first phase supports local/file and s3");
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LancePropertiesFactory.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LancePropertiesFactory.java
new file mode 100644
index 00000000000..3a3669b83da
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LancePropertiesFactory.java
@@ -0,0 +1,45 @@
+// 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.property.metastore;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/** Selects Lance filesystem or REST properties from {@code
lance.catalog.type}. */
+public class LancePropertiesFactory extends AbstractMetastorePropertiesFactory
{
+ public LancePropertiesFactory() {
+ register(AbstractLanceProperties.LANCE_FILESYSTEM,
LanceFileSystemMetastoreProperties::new);
+ register(AbstractLanceProperties.LANCE_REST,
LanceRestMetastoreProperties::new);
+ }
+
+ @Override
+ public MetastoreProperties create(Map<String, String> props) {
+ String type =
props.getOrDefault(AbstractLanceProperties.LANCE_CATALOG_TYPE,
+
AbstractLanceProperties.LANCE_FILESYSTEM).trim().toLowerCase(Locale.ROOT);
+ if (!AbstractLanceProperties.LANCE_FILESYSTEM.equals(type)
+ && !AbstractLanceProperties.LANCE_REST.equals(type)) {
+ throw new IllegalArgumentException("Property '" +
AbstractLanceProperties.LANCE_CATALOG_TYPE
+ + "' must be 'filesystem' or 'rest', but was '" + type +
"'");
+ }
+ Map<String, String> normalizedProperties = new HashMap<>(props);
+ normalizedProperties.put(AbstractLanceProperties.LANCE_CATALOG_TYPE,
type);
+ return createInternal(normalizedProperties,
AbstractLanceProperties.LANCE_CATALOG_TYPE,
+ AbstractLanceProperties.LANCE_FILESYSTEM);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceRestMetastoreProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceRestMetastoreProperties.java
new file mode 100644
index 00000000000..e989b0e6f21
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/LanceRestMetastoreProperties.java
@@ -0,0 +1,229 @@
+// 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.property.metastore;
+
+import org.apache.doris.foundation.property.ConnectorProperty;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.commons.lang3.StringUtils;
+import org.lance.namespace.LanceNamespace;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/** Properties for a Lance REST namespace catalog. */
+public class LanceRestMetastoreProperties extends AbstractLanceProperties {
+ public static final String REST_URI = "lance.rest.uri";
+ public static final String REST_SECURITY_TYPE = "lance.rest.security.type";
+ public static final String REST_BEARER_TOKEN = "lance.rest.bearer-token";
+ public static final String REST_API_KEY = "lance.rest.api-key";
+ public static final String REST_HEADER_PREFIX = "lance.rest.header.";
+
+ private static final String REST_SECURITY_NONE = "none";
+ private static final String REST_SECURITY_BEARER = "bearer";
+ private static final String REST_SECURITY_API_KEY = "api_key";
+ private static final Pattern HTTP_HEADER_NAME =
+ Pattern.compile("^[!#$%&'*+.^_`|~0-9A-Za-z-]+$");
+
+ @ConnectorProperty(
+ names = {REST_URI},
+ required = false,
+ description = "The HTTP or HTTPS endpoint of the Lance REST
namespace service."
+ )
+ private String restUri;
+
+ @ConnectorProperty(
+ names = {REST_SECURITY_TYPE},
+ required = false,
+ description = "REST authentication type: none, bearer, or api_key.
Default: none."
+ )
+ private String securityType = REST_SECURITY_NONE;
+
+ @ConnectorProperty(
+ names = {REST_BEARER_TOKEN},
+ required = false,
+ sensitive = true,
+ description = "Bearer token used when lance.rest.security.type is
bearer."
+ )
+ private String bearerToken;
+
+ @ConnectorProperty(
+ names = {REST_API_KEY},
+ required = false,
+ sensitive = true,
+ description = "API key used when lance.rest.security.type is
api_key."
+ )
+ private String apiKey;
+
+ public LanceRestMetastoreProperties(Map<String, String> props) {
+ super(props);
+ }
+
+ @Override
+ public String getLanceCatalogType() {
+ return LANCE_REST;
+ }
+
+ @Override
+ public LanceNamespace createNamespace(
+ BufferAllocator allocator, Map<String, String> javaStorageOptions)
{
+ Map<String, String> namespaceProperties = new HashMap<>();
+ namespaceProperties.put("uri", normalizedRestUri());
+ namespaceProperties.put("delimiter", getNamespaceDelimiter());
+
+ origProps.forEach((key, value) -> {
+ if (key.startsWith(REST_HEADER_PREFIX)) {
+ namespaceProperties.put(
+ "header." +
key.substring(REST_HEADER_PREFIX.length()), value);
+ }
+ });
+ if (REST_SECURITY_BEARER.equals(securityType)) {
+ namespaceProperties.put("header.Authorization", "Bearer " +
bearerToken);
+ } else if (REST_SECURITY_API_KEY.equals(securityType)) {
+ namespaceProperties.put("header.x-api-key", apiKey);
+ }
+ return LanceNamespace.connect("rest", namespaceProperties, allocator);
+ }
+
+ public String getRestUri() {
+ return normalizedRestUri();
+ }
+
+ public String getSecurityType() {
+ return securityType;
+ }
+
+ @Override
+ protected void validateCatalogProperties() {
+ restUri = origProps.get(REST_URI);
+ securityType = origProps.getOrDefault(REST_SECURITY_TYPE,
REST_SECURITY_NONE);
+ bearerToken = origProps.get(REST_BEARER_TOKEN);
+ apiKey = origProps.get(REST_API_KEY);
+ if
(origProps.containsKey(LanceFileSystemMetastoreProperties.WAREHOUSE)) {
+ throw new IllegalArgumentException(
+ "Property 'warehouse' is not valid for Lance REST
catalog");
+ }
+ if (StringUtils.isBlank(restUri)) {
+ throw new IllegalArgumentException(
+ "Missing required property '" + REST_URI + "' for Lance
REST catalog");
+ }
+ validateRestUri(restUri);
+
+ securityType = securityType.trim().toLowerCase(Locale.ROOT);
+ boolean bearerTokenConfigured =
origProps.containsKey(REST_BEARER_TOKEN);
+ boolean apiKeyConfigured = origProps.containsKey(REST_API_KEY);
+ boolean hasBearerToken = StringUtils.isNotBlank(bearerToken);
+ boolean hasApiKey = StringUtils.isNotBlank(apiKey);
+ switch (securityType) {
+ case REST_SECURITY_NONE:
+ if (bearerTokenConfigured || apiKeyConfigured) {
+ throw new IllegalArgumentException("Lance REST security
type 'none' cannot configure '"
+ + REST_BEARER_TOKEN + "' or '" + REST_API_KEY +
"'");
+ }
+ break;
+ case REST_SECURITY_BEARER:
+ if (!hasBearerToken || apiKeyConfigured) {
+ throw new IllegalArgumentException("Lance REST security
type 'bearer' requires only '"
+ + REST_BEARER_TOKEN + "'");
+ }
+ validateCredentialHeaderValue(REST_BEARER_TOKEN, bearerToken);
+ break;
+ case REST_SECURITY_API_KEY:
+ if (!hasApiKey || bearerTokenConfigured) {
+ throw new IllegalArgumentException("Lance REST security
type 'api_key' requires only '"
+ + REST_API_KEY + "'");
+ }
+ validateCredentialHeaderValue(REST_API_KEY, apiKey);
+ break;
+ default:
+ throw new IllegalArgumentException("Property '" +
REST_SECURITY_TYPE
+ + "' must be 'none', 'bearer', or 'api_key'");
+ }
+
+ Set<String> supportedKeys = new HashSet<>(Arrays.asList(
+ REST_URI, REST_SECURITY_TYPE, REST_BEARER_TOKEN,
REST_API_KEY));
+ for (Map.Entry<String, String> entry : origProps.entrySet()) {
+ String key = entry.getKey();
+ if (!key.startsWith("lance.rest.") || supportedKeys.contains(key))
{
+ continue;
+ }
+ if (!key.startsWith(REST_HEADER_PREFIX)) {
+ throw new IllegalArgumentException("Unsupported Lance REST
property '" + key + "'");
+ }
+ validateRestHeader(key.substring(REST_HEADER_PREFIX.length()),
entry.getValue());
+ }
+ }
+
+ private static void validateRestHeader(String headerName, String
headerValue) {
+ if (StringUtils.isBlank(headerName) ||
!HTTP_HEADER_NAME.matcher(headerName).matches()) {
+ throw new IllegalArgumentException("Invalid HTTP header name in
property '"
+ + REST_HEADER_PREFIX + headerName + "'");
+ }
+ if ("authorization".equalsIgnoreCase(headerName)
+ || "x-api-key".equalsIgnoreCase(headerName)) {
+ throw new IllegalArgumentException("Authentication header '" +
headerName
+ + "' must be configured through '" + REST_SECURITY_TYPE +
"'");
+ }
+ if (headerValue == null || headerValue.indexOf('\r') >= 0 ||
headerValue.indexOf('\n') >= 0) {
+ throw new IllegalArgumentException("Invalid HTTP header value in
property '"
+ + REST_HEADER_PREFIX + headerName + "'");
+ }
+ }
+
+ private static void validateCredentialHeaderValue(String propertyName,
String value) {
+ if (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0) {
+ throw new IllegalArgumentException(
+ "Invalid HTTP credential value in property '" +
propertyName + "'");
+ }
+ }
+
+ private static void validateRestUri(String value) {
+ final URI uri;
+ try {
+ uri = URI.create(value.trim());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Invalid Lance REST URI in property '" + REST_URI + "'",
e);
+ }
+ String scheme = uri.getScheme();
+ if (scheme == null
+ || (!("http".equalsIgnoreCase(scheme)) &&
!("https".equalsIgnoreCase(scheme)))) {
+ throw new IllegalArgumentException(
+ "Property '" + REST_URI + "' must use http or https");
+ }
+ if (StringUtils.isBlank(uri.getRawAuthority()) || uri.getRawUserInfo()
!= null
+ || uri.getRawQuery() != null || uri.getRawFragment() != null) {
+ throw new IllegalArgumentException("Property '" + REST_URI
+ + "' must contain an authority and cannot contain
user-info, query, or fragment");
+ }
+ }
+
+ private String normalizedRestUri() {
+ String uri = restUri.trim();
+ while (uri.endsWith("/")) {
+ uri = uri.substring(0, uri.length() - 1);
+ }
+ return uri;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/MetastoreProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/MetastoreProperties.java
index 0c9b030a20f..9d4b02fa8f9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/MetastoreProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/MetastoreProperties.java
@@ -88,7 +88,7 @@ public class MetastoreProperties extends ConnectionProperties
{
register(Type.ICEBERG, new IcebergPropertiesFactory());
register(Type.PAIMON, new PaimonPropertiesFactory());
register(Type.TRINO_CONNECTOR, new TrinoConnectorPropertiesFactory());
- register(Type.LANCE, props -> new MetastoreProperties(Type.LANCE,
props));
+ register(Type.LANCE, new LancePropertiesFactory());
}
public static void register(Type type, MetastorePropertiesFactory factory)
{
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 d461e82f1d3..61be8f0af22 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
@@ -65,7 +65,7 @@ public class S3TableValuedFunction extends
ExternalFileTableValuedFunction {
if (isLanceFormat()) {
try {
LanceTableMetadata metadata =
- LanceMetadataLoader.load(filePath,
backendConnectProperties);
+ LanceMetadataLoader.loadLatestForTvf(filePath,
backendConnectProperties);
setLanceTableMetadata(metadata);
} catch (Exception e) {
throw new AnalysisException(
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/VectorSearchTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/VectorSearchTableValuedFunction.java
index 1829d91e329..50ffc053d4c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/VectorSearchTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/VectorSearchTableValuedFunction.java
@@ -96,8 +96,7 @@ public class VectorSearchTableValuedFunction extends
TableValuedFunctionIf {
throws AnalysisException {
Map<String, String> params = normalizeProperties(properties);
sourceTableName = parseTableName(required(params, TABLE));
- checkSelectPrivilege(sourceTableName);
- sourceTable = resolveLanceTable(sourceTableName);
+ sourceTable = findLanceExternalTable(sourceTableName);
try {
metadata = sourceTable.loadMetadata();
} catch (RuntimeException e) {
@@ -108,9 +107,9 @@ public class VectorSearchTableValuedFunction extends
TableValuedFunctionIf {
throw new AnalysisException("Lance vector search requires a fixed
positive dataset version");
}
- Field vectorField = LanceVectorQuery.resolveVectorField(
+ Field vectorField = LanceVectorQuery.findVectorColumnField(
metadata.getSchema(), required(params, COLUMN));
- TSearchVector queryVector = LanceVectorQuery.encode(
+ TSearchVector queryVector = LanceVectorQuery.parseAndEncodeQueryVector(
vectorField, required(params, QUERY_VECTOR));
long topK = parseLong(params.getOrDefault(TOP_K, "10"), TOP_K, 1,
Long.MAX_VALUE);
long offset = parseLong(params.getOrDefault(OFFSET, "0"), OFFSET, 0,
Long.MAX_VALUE);
@@ -236,7 +235,8 @@ public class VectorSearchTableValuedFunction extends
TableValuedFunctionIf {
return new TableName(names.get(0), names.get(1), names.get(2));
}
- private static void checkSelectPrivilege(TableName tableName) throws
AnalysisException {
+ private static LanceExternalTable findLanceExternalTable(TableName
tableName)
+ throws AnalysisException {
ConnectContext context = ConnectContext.get();
if (!Env.getCurrentEnv().getAccessManager()
.checkTblPriv(context, tableName, PrivPredicate.SELECT)) {
@@ -244,10 +244,6 @@ public class VectorSearchTableValuedFunction extends
TableValuedFunctionIf {
context.getQualifiedUser(), context.getRemoteIP(),
tableName.getDb() + ": " + tableName.getTbl());
}
- }
-
- private static LanceExternalTable resolveLanceTable(TableName tableName)
- throws AnalysisException {
CatalogIf<?> catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(tableName.getCtl());
if (!(catalog instanceof LanceExternalCatalog)) {
throw new AnalysisException("Catalog '" + tableName.getCtl()
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 d169a1d303a..6b4a9f9c2bc 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
@@ -46,34 +46,48 @@ public class LanceFilesystemCatalogTest {
@Test
public void testNamespaceNameRoundTrip() throws Exception {
- Assert.assertEquals(Collections.emptyList(), LanceNamespaceName.decode(
- LanceNamespaceName.encode(Collections.emptyList(), ".",
"default"), ".", "default"));
+ Assert.assertEquals(Collections.emptyList(),
LanceNamespaceName.dorisDatabaseNameToNamespace(
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ Collections.emptyList(), ".", "default"),
+ ".", "default"));
Assert.assertEquals("doris",
- LanceNamespaceName.encode(Collections.singletonList("doris"),
".", "default"));
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ Collections.singletonList("doris"), ".", "default"));
Assert.assertEquals("company.analytics",
- LanceNamespaceName.encode(java.util.Arrays.asList("company",
"analytics"), ".", "default"));
- Assert.assertEquals(java.util.Arrays.asList("company", "analytics"),
LanceNamespaceName.decode(
- LanceNamespaceName.encode(java.util.Arrays.asList("company",
"analytics"), ".", "default"),
- ".", "default"));
- Assert.assertEquals(java.util.Arrays.asList("a.b", "c"),
LanceNamespaceName.decode(
- LanceNamespaceName.encode(java.util.Arrays.asList("a.b", "c"),
".", "default"),
- ".", "default"));
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ java.util.Arrays.asList("company", "analytics"), ".",
"default"));
+ Assert.assertEquals(java.util.Arrays.asList("company", "analytics"),
+ LanceNamespaceName.dorisDatabaseNameToNamespace(
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ java.util.Arrays.asList("company",
"analytics"), ".", "default"),
+ ".", "default"));
+ Assert.assertEquals(java.util.Arrays.asList("a.b", "c"),
+ LanceNamespaceName.dorisDatabaseNameToNamespace(
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ java.util.Arrays.asList("a.b", "c"), ".",
"default"),
+ ".", "default"));
java.util.List<String> delimiterAtEnd = java.util.Arrays.asList("a.",
"b");
java.util.List<String> delimiterAtStart = java.util.Arrays.asList("a",
".b");
- String encodedAtEnd = LanceNamespaceName.encode(delimiterAtEnd, ".",
"default");
- String encodedAtStart = LanceNamespaceName.encode(delimiterAtStart,
".", "default");
+ String encodedAtEnd =
+
LanceNamespaceName.namespaceToDorisDatabaseName(delimiterAtEnd, ".", "default");
+ String encodedAtStart =
+
LanceNamespaceName.namespaceToDorisDatabaseName(delimiterAtStart, ".",
"default");
Assert.assertNotEquals(encodedAtEnd, encodedAtStart);
- Assert.assertEquals(delimiterAtEnd,
LanceNamespaceName.decode(encodedAtEnd, ".", "default"));
- Assert.assertEquals(delimiterAtStart,
LanceNamespaceName.decode(encodedAtStart, ".", "default"));
- Assert.assertEquals(java.util.Arrays.asList("a\\b", "c"),
LanceNamespaceName.decode(
- LanceNamespaceName.encode(java.util.Arrays.asList("a\\b",
"c"), ".", "default"),
- ".", "default"));
+ Assert.assertEquals(delimiterAtEnd,
+ LanceNamespaceName.dorisDatabaseNameToNamespace(encodedAtEnd,
".", "default"));
+ Assert.assertEquals(delimiterAtStart,
+
LanceNamespaceName.dorisDatabaseNameToNamespace(encodedAtStart, ".",
"default"));
+ Assert.assertEquals(java.util.Arrays.asList("a\\b", "c"),
+ LanceNamespaceName.dorisDatabaseNameToNamespace(
+ LanceNamespaceName.namespaceToDorisDatabaseName(
+ java.util.Arrays.asList("a\\b", "c"), ".",
"default"),
+ ".", "default"));
- String rootCollision = LanceNamespaceName.encode(
+ String rootCollision = LanceNamespaceName.namespaceToDorisDatabaseName(
Collections.singletonList("default"), ".", "default");
Assert.assertEquals("\\default", rootCollision);
Assert.assertEquals(Collections.singletonList("default"),
- LanceNamespaceName.decode(rootCollision, ".", "default"));
+ LanceNamespaceName.dorisDatabaseNameToNamespace(rootCollision,
".", "default"));
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceRestCatalogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceRestCatalogTest.java
index 510153b5b97..46ece3f2ebd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceRestCatalogTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceRestCatalogTest.java
@@ -41,6 +41,9 @@ import java.util.concurrent.CopyOnWriteArrayList;
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class LanceRestCatalogTest {
private static final String BEARER_TOKEN = "test-bearer-token";
+ private static final String EXISTING_TABLE = "existing_table";
+ private static final String MISSING_TABLE = "missing_table";
+ private static final String MISSING_NAMESPACE = "missing_namespace";
private final List<RequestRecord> requests = new CopyOnWriteArrayList<>();
private HttpServer server;
@@ -156,6 +159,25 @@ public class LanceRestCatalogTest {
assertInvalid(filesystemWithRest, "not valid for Lance filesystem
catalog");
}
+ @Test
+ public void testTableExist() {
+ LanceExternalCatalog catalog = new LanceExternalCatalog(
+ 201, "lance_rest_table_exists", null, restProperties(), "");
+ try {
+ Assertions.assertTrue(catalog.tableExist(null, "default",
EXISTING_TABLE));
+ Assertions.assertFalse(catalog.tableExist(null, "default",
MISSING_TABLE));
+ Assertions.assertFalse(catalog.tableExist(null, MISSING_NAMESPACE,
MISSING_TABLE));
+ } finally {
+ catalog.onClose();
+ }
+ }
+
+ @Test
+ public void testTableExistDoesNotHideServiceErrors() {
+ assertTableExistThrows(202, "lance_rest_table_exists_unauthorized",
"fail/");
+ assertTableExistThrows(203, "lance_rest_table_exists_unavailable",
"service-error/");
+ }
+
private Map<String, String> restProperties() {
Map<String, String> properties = restPropertiesWithoutUri();
properties.put(LanceExternalCatalog.REST_URI, restUri);
@@ -177,6 +199,20 @@ public class LanceRestCatalogTest {
Assertions.assertTrue(exception.getMessage().contains(expectedMessage),
exception.getMessage());
}
+ private void assertTableExistThrows(long catalogId, String catalogName,
String endpointPrefix) {
+ Map<String, String> properties = restProperties();
+ properties.put(LanceExternalCatalog.REST_URI, restUri +
endpointPrefix);
+ LanceExternalCatalog catalog = new LanceExternalCatalog(
+ catalogId, catalogName, null, properties, "");
+ try {
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> catalog.tableExist(null, "default", EXISTING_TABLE));
+ } finally {
+ catalog.onClose();
+ }
+ }
+
private void handleRequest(HttpExchange exchange) throws IOException {
ByteStreams.toByteArray(exchange.getRequestBody());
String path = exchange.getRequestURI().getPath();
@@ -189,6 +225,21 @@ public class LanceRestCatalogTest {
if (path.startsWith("/fail/")) {
status = 401;
response = "{\"error\":\"invalid token " + BEARER_TOKEN +
"\",\"code\":16}";
+ } else if (path.startsWith("/service-error/")) {
+ status = 503;
+ response = "{\"error\":\"service unavailable\",\"code\":17}";
+ } else if ("POST".equals(exchange.getRequestMethod())
+ && path.matches("/v1/table/[^/]+/exists")) {
+ String tableId = path.substring("/v1/table/".length(),
path.length() - "/exists".length());
+ if (EXISTING_TABLE.equals(tableId)) {
+ exchange.sendResponseHeaders(200, -1);
+ exchange.close();
+ return;
+ }
+ status = 404;
+ response = tableId.startsWith(MISSING_NAMESPACE + "$")
+ ? "{\"error\":\"namespace not found\",\"code\":1}"
+ : "{\"error\":\"table not found\",\"code\":4}";
} else if (path.endsWith("/table/list")) {
response = "{\"tables\":[]}";
} else if (path.endsWith("/list")) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceVectorQueryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceVectorQueryTest.java
index 0a9f1821b0d..1ccd5fa7db6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceVectorQueryTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceVectorQueryTest.java
@@ -41,7 +41,8 @@ public class LanceVectorQueryTest {
public void testEncodesFloat32AsLittleEndianTypedVector() throws Exception
{
Field field = vectorField("embedding",
new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE), 3);
- TSearchVector vector = LanceVectorQuery.encode(field, "[0.25, -1.5,
3]");
+ TSearchVector vector =
+ LanceVectorQuery.parseAndEncodeQueryVector(field, "[0.25,
-1.5, 3]");
Assertions.assertEquals(TVectorElementType.FLOAT32,
vector.getElementType());
Assertions.assertEquals(3, vector.getDimension());
@@ -54,24 +55,24 @@ public class LanceVectorQueryTest {
@Test
public void testEncodesAllLanceCElementTypes() throws Exception {
- TSearchVector float16 = LanceVectorQuery.encode(vectorField("f16",
+ TSearchVector float16 =
LanceVectorQuery.parseAndEncodeQueryVector(vectorField("f16",
new ArrowType.FloatingPoint(FloatingPointPrecision.HALF), 1),
"[1.5]");
Assertions.assertEquals(TVectorElementType.FLOAT16,
float16.getElementType());
Assertions.assertEquals(1.5F, Float16.toFloat(
float16.bufferForValues().order(ByteOrder.LITTLE_ENDIAN).getShort()));
- TSearchVector float64 = LanceVectorQuery.encode(vectorField("f64",
+ TSearchVector float64 =
LanceVectorQuery.parseAndEncodeQueryVector(vectorField("f64",
new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE),
1), "[1.25]");
Assertions.assertEquals(TVectorElementType.FLOAT64,
float64.getElementType());
Assertions.assertEquals(1.25,
float64.bufferForValues().order(ByteOrder.LITTLE_ENDIAN).getDouble());
- TSearchVector uint8 = LanceVectorQuery.encode(
+ TSearchVector uint8 = LanceVectorQuery.parseAndEncodeQueryVector(
vectorField("u8", new ArrowType.Int(8, false), 2), "[0, 255]");
Assertions.assertEquals(TVectorElementType.UINT8,
uint8.getElementType());
Assertions.assertArrayEquals(new byte[] {0, (byte) 255},
uint8.getValues());
- TSearchVector int8 = LanceVectorQuery.encode(
+ TSearchVector int8 = LanceVectorQuery.parseAndEncodeQueryVector(
vectorField("i8", new ArrowType.Int(8, true), 2), "[-128,
127]");
Assertions.assertEquals(TVectorElementType.INT8,
int8.getElementType());
Assertions.assertArrayEquals(new byte[] {(byte) -128, 127},
int8.getValues());
@@ -82,12 +83,14 @@ public class LanceVectorQueryTest {
Field float32 = vectorField("embedding",
new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE), 3);
AnalysisException dimension = Assertions.assertThrows(
- AnalysisException.class, () ->
LanceVectorQuery.encode(float32, "[1, 2]"));
+ AnalysisException.class,
+ () -> LanceVectorQuery.parseAndEncodeQueryVector(float32, "[1,
2]"));
Assertions.assertTrue(dimension.getMessage().contains("dimension"));
Field uint8 = vectorField("embedding", new ArrowType.Int(8, false), 1);
AnalysisException range = Assertions.assertThrows(
- AnalysisException.class, () -> LanceVectorQuery.encode(uint8,
"[256]"));
+ AnalysisException.class,
+ () -> LanceVectorQuery.parseAndEncodeQueryVector(uint8,
"[256]"));
Assertions.assertTrue(range.getMessage().contains("uint8"));
}
@@ -97,7 +100,8 @@ public class LanceVectorQueryTest {
Collections.singletonList(Field.nullable("item",
new
ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE))));
Assertions.assertThrows(
- AnalysisException.class, () -> LanceVectorQuery.encode(list,
"[1]"));
+ AnalysisException.class,
+ () -> LanceVectorQuery.parseAndEncodeQueryVector(list, "[1]"));
Field bfloat16Item = new Field("item",
new FieldType(true, new ArrowType.FixedSizeBinary(2), null,
@@ -107,7 +111,8 @@ public class LanceVectorQueryTest {
FieldType.nullable(new ArrowType.FixedSizeList(1)),
Collections.singletonList(bfloat16Item));
Assertions.assertThrows(
- AnalysisException.class, () ->
LanceVectorQuery.encode(bfloat16Vector, "[1]"));
+ AnalysisException.class,
+ () ->
LanceVectorQuery.parseAndEncodeQueryVector(bfloat16Vector, "[1]"));
}
@Test
@@ -120,7 +125,7 @@ public class LanceVectorQueryTest {
AnalysisException ambiguous = Assertions.assertThrows(
AnalysisException.class,
- () -> LanceVectorQuery.resolveVectorField(schema,
"embedding"));
+ () -> LanceVectorQuery.findVectorColumnField(schema,
"embedding"));
Assertions.assertTrue(ambiguous.getMessage().contains("ambiguous"));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java
new file mode 100644
index 00000000000..1702d583f4f
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java
@@ -0,0 +1,63 @@
+// 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.property.metastore;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class LancePropertiesTest {
+ @Test
+ public void testDefaultFilesystemProperties() throws Exception {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("type", "lance");
+ properties.put(LanceFileSystemMetastoreProperties.WAREHOUSE,
"/tmp/lance");
+
+ AbstractLanceProperties lanceProperties =
+ (AbstractLanceProperties)
MetastoreProperties.create(properties);
+
+ Assertions.assertInstanceOf(LanceFileSystemMetastoreProperties.class,
lanceProperties);
+ Assertions.assertEquals(AbstractLanceProperties.LANCE_FILESYSTEM,
+ lanceProperties.getLanceCatalogType());
+ Assertions.assertEquals(AbstractLanceProperties.DEFAULT_DELIMITER,
+ lanceProperties.getNamespaceDelimiter());
+ Assertions.assertEquals(AbstractLanceProperties.DEFAULT_ROOT_DATABASE,
+ lanceProperties.getRootDatabase());
+ }
+
+ @Test
+ public void testRestProperties() throws Exception {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("type", "lance");
+ properties.put(AbstractLanceProperties.LANCE_CATALOG_TYPE,
+ AbstractLanceProperties.LANCE_REST);
+ properties.put(LanceRestMetastoreProperties.REST_URI,
"http://localhost:8080/";);
+ properties.put(LanceRestMetastoreProperties.REST_SECURITY_TYPE,
"bearer");
+ properties.put(LanceRestMetastoreProperties.REST_BEARER_TOKEN,
"token");
+
+ AbstractLanceProperties lanceProperties =
+ (AbstractLanceProperties)
MetastoreProperties.create(properties);
+
+ Assertions.assertInstanceOf(LanceRestMetastoreProperties.class,
lanceProperties);
+ LanceRestMetastoreProperties restProperties =
(LanceRestMetastoreProperties) lanceProperties;
+ Assertions.assertEquals("http://localhost:8080";,
restProperties.getRestUri());
+ Assertions.assertEquals("bearer", restProperties.getSecurityType());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
