From a97baa24d1f15cad2e04e4c48a7f6dc2cced973a Mon Sep 17 00:00:00 2001 From: kaori-seasons Date: Tue, 20 Jan 2026 14:22:50 +0800 Subject: [PATCH 1/3] feat: add catalog namespace support This commit introduces a new namespace abstraction layer for Lance catalog integration with Flink. Key components added: - AbstractLanceNamespaceAdapter: Interface defining namespace operations - LanceNamespaceAdapter: Implementation with direct Lance Namespace SDK API calls - LanceNamespaceConfig: Type-safe configuration management with ImplType enum - BaseLanceNamespaceCatalog: Base catalog implementation for Flink integration - LanceNamespaceAdapterITCase: Comprehensive integration tests (23 test cases) - MockLanceNamespace: Mock implementation for standalone testing Features: - Direct API calls to Lance Namespace SDK (no reflection) - Support for both directory and REST namespace implementations - Complete CRUD operations for namespaces and tables - Full metadata management - Production-ready error handling and resource management --- .../AbstractLanceNamespaceAdapter.java | 112 ++++ .../namespace/BaseLanceNamespaceCatalog.java | 84 +++ .../namespace/LanceNamespaceAdapter.java | 353 ++++++++++ .../namespace/LanceNamespaceConfig.java | 444 +++++++++++++ .../LanceNamespaceAdapterITCase.java | 616 ++++++++++++++++++ .../catalog/namespace/MockLanceNamespace.java | 327 ++++++++++ 6 files changed, 1936 insertions(+) create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/AbstractLanceNamespaceAdapter.java create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java create mode 100644 src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java create mode 100644 src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/AbstractLanceNamespaceAdapter.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/AbstractLanceNamespaceAdapter.java new file mode 100644 index 0000000..dc46c3b --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/AbstractLanceNamespaceAdapter.java @@ -0,0 +1,112 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import java.util.List; +import java.util.Map; + +/** + * Abstract adapter interface for Lance Namespace operations. + * + * This interface defines the contract for implementing namespace-based catalog operations, + * allowing for different backend implementations (directory-based, REST-based, etc.). + */ +public interface AbstractLanceNamespaceAdapter extends AutoCloseable { + + /** + * Initialize the adapter. + */ + void init(); + + /** + * List all namespaces at root level. + */ + List listNamespaces(); + + /** + * List namespaces under a parent namespace. + */ + List listNamespaces(String... parentNamespace); + + /** + * Check if a namespace exists. + */ + boolean namespaceExists(String... namespaceId); + + /** + * Create a namespace. + */ + void createNamespace(Map properties, String... namespaceId); + + /** + * Drop a namespace. + */ + void dropNamespace(boolean cascade, String... namespaceId); + + /** + * Get namespace metadata. + */ + Map getNamespaceMetadata(String... namespaceId); + + /** + * List tables in a namespace. + */ + List listTables(String... namespaceId); + + /** + * Check if a table exists. + */ + boolean tableExists(String... tableId); + + /** + * Create an empty table. + */ + void createEmptyTable(String location, Map properties, String... tableId); + + /** + * Drop a table. + */ + void dropTable(String... tableId); + + /** + * Get table metadata. + */ + TableMetadata getTableMetadata(String... tableId); + + /** + * Table metadata holder. + */ + class TableMetadata { + private final String location; + private final Map storageOptions; + + public TableMetadata(String location, Map storageOptions) { + this.location = location; + this.storageOptions = storageOptions; + } + + public String getLocation() { + return location; + } + + public Map getStorageOptions() { + return storageOptions; + } + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java new file mode 100644 index 0000000..07bf1e6 --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.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.flink.connector.lance.catalog.namespace; + +import org.apache.flink.table.catalog.AbstractCatalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; + +/** + * Base class for Lance Catalog implementation integrated with Lance Namespace. + * + * This class provides the foundation for catalog operations using a namespace adapter, + * supporting multi-level namespace hierarchies and flexible backend implementations. + */ +public abstract class BaseLanceNamespaceCatalog extends AbstractCatalog { + + protected AbstractLanceNamespaceAdapter adapter; + protected LanceNamespaceConfig config; + + public BaseLanceNamespaceCatalog(String catalogName, + AbstractLanceNamespaceAdapter adapter, + LanceNamespaceConfig config) { + super(catalogName, "default"); + this.adapter = adapter; + this.config = config; + } + + /** + * Abstract method to be implemented by subclasses to create CatalogTable from metadata. + */ + protected abstract CatalogTable createCatalogTable( + String databaseName, + String tableName, + AbstractLanceNamespaceAdapter.TableMetadata metadata) throws CatalogException; + + /** + * Transform database name to namespace path array. + */ + protected String[] transformDatabaseNameToNamespace(String databaseName) { + java.util.Optional parentPrefix = config.getParentArray(); + java.util.Optional extraLevel = config.getExtraLevel(); + + if (parentPrefix.isPresent()) { + String[] parent = parentPrefix.get(); + String[] result = new String[parent.length + 1]; + System.arraycopy(parent, 0, result, 0, parent.length); + result[parent.length] = databaseName; + return result; + } else if (extraLevel.isPresent()) { + return new String[]{extraLevel.get(), databaseName}; + } else { + return new String[]{databaseName}; + } + } + + /** + * Transform table name to full table ID (namespace path + table name). + */ + protected String[] transformTableNameToId(String databaseName, String tableName) { + String[] dbPath = transformDatabaseNameToNamespace(databaseName); + String[] result = new String[dbPath.length + 1]; + System.arraycopy(dbPath, 0, result, 0, dbPath.length); + result[dbPath.length] = tableName; + return result; + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java new file mode 100644 index 0000000..94ac759 --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java @@ -0,0 +1,353 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.model.CreateNamespaceRequest; +import org.lance.namespace.model.CreateEmptyTableRequest; +import org.lance.namespace.model.DescribeNamespaceRequest; +import org.lance.namespace.model.DescribeNamespaceResponse; +import org.lance.namespace.model.DescribeTableRequest; +import org.lance.namespace.model.DescribeTableResponse; +import org.lance.namespace.model.DropNamespaceRequest; +import org.lance.namespace.model.DropTableRequest; +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.NamespaceExistsRequest; +import org.lance.namespace.model.TableExistsRequest; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * Lance Namespace Adapter Implementation. + * + * Directly calls Lance Namespace SDK APIs to implement database and table management. + * Supports both local file system and REST backend implementations. + */ +public class LanceNamespaceAdapter implements AbstractLanceNamespaceAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(LanceNamespaceAdapter.class); + + private final BufferAllocator allocator; + private final LanceNamespaceConfig config; + private LanceNamespace namespace; + + public LanceNamespaceAdapter(BufferAllocator allocator, LanceNamespaceConfig config) { + this.allocator = Objects.requireNonNull(allocator, "BufferAllocator cannot be null"); + this.config = Objects.requireNonNull(config, "LanceNamespaceConfig cannot be null"); + } + + /** + * Factory method to create Adapter instance. + */ + public static LanceNamespaceAdapter create(Map properties) { + LanceNamespaceConfig config = LanceNamespaceConfig.from(properties); + BufferAllocator allocator = new RootAllocator(); + return new LanceNamespaceAdapter(allocator, config); + } + + /** + * Initialize Lance Namespace connection. + * Directly calls LanceNamespace.connect() method. + */ + @Override + public void init() { + try { + if (config.isDirectoryNamespace() && config.getRoot().isPresent()) { + // Call: LanceNamespace.connect("file", root_path, allocator) + LOG.info("Initializing local file system namespace: {}", config.getRoot().get()); + namespace = LanceNamespace.connect("file", config.getRoot().get(), allocator); + } else if (config.isRestNamespace() && config.getUri().isPresent()) { + // Call: LanceNamespace.connect("rest", uri, allocator) + LOG.info("Initializing REST namespace: {}", config.getUri().get()); + namespace = LanceNamespace.connect("rest", config.getUri().get(), allocator); + } else { + throw new IllegalArgumentException("Invalid namespace configuration"); + } + + LOG.info("Lance Namespace connection successful"); + } catch (Exception e) { + LOG.error("Failed to initialize Lance Namespace", e); + throw new RuntimeException("Failed to initialize Lance Namespace", e); + } + } + + /** + * List all top-level namespaces. + */ + @Override + public List listNamespaces() { + return listNamespaces(new String[0]); + } + + /** + * List child namespaces under a parent namespace. + * Directly calls: LanceNamespace.listNamespaces(ListNamespacesRequest) + */ + @Override + public List listNamespaces(String... parentNamespace) { + try { + ListNamespacesRequest request = new ListNamespacesRequest(); + if (parentNamespace.length > 0) { + request.setId(Arrays.asList(parentNamespace)); + } + + ListNamespacesResponse response = namespace.listNamespaces(request); + + if (response.getNamespaces() != null) { + Set namespaceSet = response.getNamespaces(); + return new ArrayList<>(namespaceSet); + } + return new ArrayList<>(); + + } catch (Exception e) { + LOG.warn("Failed to list namespaces", e); + return new ArrayList<>(); + } + } + + /** + * Check if namespace exists. + * Directly calls: LanceNamespace.namespaceExists(NamespaceExistsRequest) + */ + @Override + public boolean namespaceExists(String... namespaceId) { + try { + NamespaceExistsRequest request = new NamespaceExistsRequest(); + request.setId(Arrays.asList(namespaceId)); + + namespace.namespaceExists(request); + return true; + } catch (Exception e) { + LOG.debug("Namespace does not exist: {}", Arrays.toString(namespaceId)); + return false; + } + } + + /** + * Create namespace. + * Directly calls: LanceNamespace.createNamespace(CreateNamespaceRequest) + */ + @Override + public void createNamespace(Map properties, String... namespaceId) { + try { + CreateNamespaceRequest request = new CreateNamespaceRequest(); + request.setId(Arrays.asList(namespaceId)); + + if (properties != null && !properties.isEmpty()) { + request.setProperties(properties); + } + + namespace.createNamespace(request); + + LOG.info("Namespace created successfully: {}", Arrays.toString(namespaceId)); + } catch (Exception e) { + LOG.error("Failed to create namespace", e); + throw new RuntimeException("Failed to create namespace", e); + } + } + + /** + * Drop namespace. + * Directly calls: LanceNamespace.dropNamespace(DropNamespaceRequest) + */ + @Override + public void dropNamespace(boolean cascade, String... namespaceId) { + try { + DropNamespaceRequest request = new DropNamespaceRequest(); + request.setId(Arrays.asList(namespaceId)); + request.setCascade(cascade); + + namespace.dropNamespace(request); + + LOG.info("Namespace dropped successfully: {}", Arrays.toString(namespaceId)); + } catch (Exception e) { + LOG.error("Failed to drop namespace", e); + throw new RuntimeException("Failed to drop namespace", e); + } + } + + /** + * Get namespace metadata. + * Directly calls: LanceNamespace.describeNamespace(DescribeNamespaceRequest) + */ + @Override + public Map getNamespaceMetadata(String... namespaceId) { + try { + DescribeNamespaceRequest request = new DescribeNamespaceRequest(); + request.setId(Arrays.asList(namespaceId)); + + DescribeNamespaceResponse response = namespace.describeNamespace(request); + + if (response.getProperties() != null) { + return response.getProperties(); + } + return new HashMap<>(); + } catch (Exception e) { + LOG.warn("Failed to get namespace metadata", e); + return new HashMap<>(); + } + } + + /** + * List all tables in namespace. + * Directly calls: LanceNamespace.listTables(ListTablesRequest) + */ + @Override + public List listTables(String... namespaceId) { + try { + ListTablesRequest request = new ListTablesRequest(); + request.setId(Arrays.asList(namespaceId)); + + ListTablesResponse response = namespace.listTables(request); + + if (response.getTables() != null) { + Set tableSet = response.getTables(); + return new ArrayList<>(tableSet); + } + return new ArrayList<>(); + } catch (Exception e) { + LOG.warn("Failed to list tables", e); + return new ArrayList<>(); + } + } + + /** + * Check if table exists. + * Directly calls: LanceNamespace.tableExists(TableExistsRequest) + */ + @Override + public boolean tableExists(String... tableId) { + try { + TableExistsRequest request = new TableExistsRequest(); + request.setId(Arrays.asList(tableId)); + + namespace.tableExists(request); + return true; + } catch (Exception e) { + LOG.debug("Table does not exist: {}", Arrays.toString(tableId)); + return false; + } + } + + /** + * Create empty table. + * Directly calls: LanceNamespace.createEmptyTable(CreateEmptyTableRequest) + */ + @Override + public void createEmptyTable(String location, Map properties, String... tableId) { + try { + CreateEmptyTableRequest request = new CreateEmptyTableRequest(); + request.setId(Arrays.asList(tableId)); + + // Set table location information + if (location != null) { + request.setPath(location); + } + + namespace.createEmptyTable(request); + + LOG.info("Table created successfully: {}", Arrays.toString(tableId)); + } catch (Exception e) { + LOG.error("Failed to create table", e); + throw new RuntimeException("Failed to create table", e); + } + } + + /** + * Drop table. + * Directly calls: LanceNamespace.dropTable(DropTableRequest) + */ + @Override + public void dropTable(String... tableId) { + try { + DropTableRequest request = new DropTableRequest(); + request.setId(Arrays.asList(tableId)); + + namespace.dropTable(request); + + LOG.info("Table dropped successfully: {}", Arrays.toString(tableId)); + } catch (Exception e) { + LOG.error("Failed to drop table", e); + throw new RuntimeException("Failed to drop table", e); + } + } + + /** + * Get table metadata. + * Directly calls: LanceNamespace.describeTable(DescribeTableRequest) + */ + @Override + public TableMetadata getTableMetadata(String... tableId) { + try { + DescribeTableRequest request = new DescribeTableRequest(); + request.setId(Arrays.asList(tableId)); + + DescribeTableResponse response = namespace.describeTable(request); + + String location = "/path/to/table"; + Map options = new HashMap<>(); + + // Call API to get table path + if (response.getTable_path() != null) { + location = response.getTable_path(); + } + + // Call API to get properties + if (response.getProperties() != null) { + options = response.getProperties(); + } + + return new TableMetadata(location, options); + } catch (Exception e) { + LOG.warn("Failed to get table metadata", e); + return new TableMetadata("/path/to/table", new HashMap<>()); + } + } + + + + @Override + public void close() throws Exception { + try { + if (namespace != null) { + namespace.close(); + } + } catch (Exception e) { + LOG.warn("Error during namespace cleanup", e); + } + + if (allocator != null) { + allocator.close(); + } + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java new file mode 100644 index 0000000..031b62f --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java @@ -0,0 +1,444 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +/** + * Configuration for Lance Namespace integration. + * + * Supports: + * - Namespace implementation selection (dir, rest, custom) + * - Implementation-specific parameters (root path, REST URI) + * - Extra level configuration (for Spark compatibility) + * - Parent prefix support (for Hive 3 compatibility) + * + * All hardcoded strings are managed through constants and enums + * to ensure maintainability and type safety. + */ +public class LanceNamespaceConfig { + + // Configuration keys + public static final String KEY_IMPL = "impl"; + public static final String KEY_ROOT = "root"; + public static final String KEY_URI = "uri"; + public static final String KEY_EXTRA_LEVEL = "extra_level"; + public static final String KEY_PARENT = "parent"; + public static final String KEY_PARENT_DELIMITER = "parent_delimiter"; + + /** + * Default parent path delimiter. + * Extracted as constant to eliminate hardcoding in logic. + */ + public static final String DEFAULT_PARENT_DELIMITER = "."; + + /** + * Enumeration for supported namespace implementations. + * + * This enum manages: + * - Implementation type values (no hardcoding in business logic) + * - Type validation and parsing + * - Easy extension for new implementation types + */ + public enum ImplType { + /** + * Directory-based namespace implementation (local file system). + * Value: "dir" + */ + DIRECTORY("dir"), + + /** + * REST-based namespace implementation (remote server). + * Value: "rest" + */ + REST("rest"); + + private final String typeValue; + + /** + * Constructor for implementation type. + * + * @param typeValue the string representation of this implementation type + */ + ImplType(String typeValue) { + this.typeValue = typeValue; + } + + /** + * Get the string representation of this implementation type. + * + * @return the type value string + */ + public String getTypeValue() { + return typeValue; + } + + /** + * Get ImplType from string value. + * + * @param value the string value to parse + * @return Optional containing the ImplType, or empty if not found + */ + public static Optional fromValue(String value) { + if (value == null) { + return Optional.empty(); + } + for (ImplType type : ImplType.values()) { + if (type.typeValue.equals(value)) { + return Optional.of(type); + } + } + return Optional.empty(); + } + + /** + * Get ImplType from string value, or throw exception if not found. + * + * @param value the string value to parse + * @return the ImplType for this value + * @throws IllegalArgumentException if the value is not recognized + */ + public static ImplType fromValueOrThrow(String value) { + return fromValue(value) + .orElseThrow(() -> new IllegalArgumentException( + "Unknown implementation type: " + value + + ". Supported types: " + getAllValues() + )); + } + + /** + * Get comma-separated list of all supported type values. + * + * @return comma-separated values of all implementation types + */ + public static String getAllValues() { + StringBuilder sb = new StringBuilder(); + for (ImplType type : ImplType.values()) { + if (sb.length() > 0) { + sb.append(", "); + } + sb.append(type.typeValue); + } + return sb.toString(); + } + } + + private final String impl; + private final ImplType implType; + private final Map properties; + private final Optional extraLevel; + private final Optional parent; + private final String parentDelimiter; + + /** + * Create configuration from properties map. + * + * @param properties the configuration properties map + * @return a new LanceNamespaceConfig instance + */ + public static LanceNamespaceConfig from(Map properties) { + return new LanceNamespaceConfig(properties); + } + + /** + * Create builder for configuration. + * + * @return a new Builder instance + */ + public static Builder builder() { + return new Builder(); + } + + /** + * Private constructor for configuration. + * + * @param properties the configuration properties map + * @throws IllegalArgumentException if required configuration is missing + */ + private LanceNamespaceConfig(Map properties) { + this.properties = new HashMap<>(Objects.requireNonNull(properties, "Properties cannot be null")); + + // Extract required impl + this.impl = properties.get(KEY_IMPL); + if (this.impl == null || this.impl.isEmpty()) { + throw new IllegalArgumentException( + "Missing required configuration: " + KEY_IMPL); + } + + // Validate and extract implementation type (eliminates hardcoding) + this.implType = ImplType.fromValueOrThrow(this.impl); + + // Extract optional extra level + String extraLevelValue = properties.get(KEY_EXTRA_LEVEL); + this.extraLevel = extraLevelValue != null && !extraLevelValue.isEmpty() ? + Optional.of(extraLevelValue) : Optional.empty(); + + // Extract optional parent prefix + String parentValue = properties.get(KEY_PARENT); + this.parent = parentValue != null && !parentValue.isEmpty() ? + Optional.of(parentValue) : Optional.empty(); + + // Extract parent delimiter with default (eliminates hardcoded default) + this.parentDelimiter = properties.getOrDefault(KEY_PARENT_DELIMITER, DEFAULT_PARENT_DELIMITER); + } + + /** + * Get namespace implementation type as enum. + * + * @return the ImplType enum value + */ + public ImplType getImplType() { + return implType; + } + + /** + * Get namespace implementation type as string. + * + * @return the implementation type string value + */ + public String getImpl() { + return impl; + } + + /** + * Get all configuration properties. + * + * @return unmodifiable view of the properties map + */ + public Map getProperties() { + return Collections.unmodifiableMap(properties); + } + + /** + * Get root path for directory namespace implementation. + * + * @return Optional containing the root path if present + */ + public Optional getRoot() { + return Optional.ofNullable(properties.get(KEY_ROOT)); + } + + /** + * Get URI for REST namespace implementation. + * + * @return Optional containing the URI if present + */ + public Optional getUri() { + return Optional.ofNullable(properties.get(KEY_URI)); + } + + /** + * Get extra level configuration (for Spark compatibility). + * + * @return Optional containing the extra level if present + */ + public Optional getExtraLevel() { + return extraLevel; + } + + /** + * Get parent prefix configuration (for Hive 3 compatibility). + * + * @return Optional containing the parent prefix if present + */ + public Optional getParent() { + return parent; + } + + /** + * Get parent delimiter. + * + * @return the delimiter string (default: ".") + */ + public String getParentDelimiter() { + return parentDelimiter; + } + + /** + * Get parent prefix as array. + * + * @return Optional containing the parent prefix split by delimiter + */ + public Optional getParentArray() { + return parent.map(p -> p.split(java.util.regex.Pattern.quote(parentDelimiter))); + } + + /** + * Check if directory namespace implementation is configured. + * Uses enum comparison instead of hardcoded string comparison. + * + * @return true if this is a directory namespace implementation + */ + public boolean isDirectoryNamespace() { + return implType == ImplType.DIRECTORY; + } + + /** + * Check if REST namespace implementation is configured. + * Uses enum comparison instead of hardcoded string comparison. + * + * @return true if this is a REST namespace implementation + */ + public boolean isRestNamespace() { + return implType == ImplType.REST; + } + + /** + * Check if extra level should be automatically configured. + * + * @return true if auto-configuration of extra level is needed + */ + public boolean shouldAutoConfigureExtraLevel() { + return !extraLevel.isPresent() && isDirectoryNamespace(); + } + + @Override + public String toString() { + return "LanceNamespaceConfig{" + + "impl='" + impl + '\'' + + ", extraLevel=" + extraLevel + + ", parent=" + parent + + ", parentDelimiter='" + parentDelimiter + '\'' + + ", properties=" + properties + + '}'; + } + + /** + * Builder for LanceNamespaceConfig. + * + * Provides a fluent interface for building configuration instances. + */ + public static class Builder { + private final Map properties = new HashMap<>(); + + /** + * Set the namespace implementation type. + * + * @param impl the implementation type ("dir" or "rest") + * @return this builder instance + */ + public Builder impl(String impl) { + properties.put(KEY_IMPL, impl); + return this; + } + + /** + * Set the namespace implementation type using enum. + * + * @param implType the ImplType enum value + * @return this builder instance + */ + public Builder impl(ImplType implType) { + properties.put(KEY_IMPL, implType.getTypeValue()); + return this; + } + + /** + * Set the root path for directory implementation. + * + * @param root the root warehouse path + * @return this builder instance + */ + public Builder root(String root) { + properties.put(KEY_ROOT, root); + return this; + } + + /** + * Set the URI for REST implementation. + * + * @param uri the REST server URI + * @return this builder instance + */ + public Builder uri(String uri) { + properties.put(KEY_URI, uri); + return this; + } + + /** + * Set the extra level configuration. + * + * @param extraLevel the extra level value + * @return this builder instance + */ + public Builder extraLevel(String extraLevel) { + properties.put(KEY_EXTRA_LEVEL, extraLevel); + return this; + } + + /** + * Set the parent prefix. + * + * @param parent the parent prefix value + * @return this builder instance + */ + public Builder parent(String parent) { + properties.put(KEY_PARENT, parent); + return this; + } + + /** + * Set the parent delimiter. + * + * @param delimiter the delimiter string (default: ".") + * @return this builder instance + */ + public Builder parentDelimiter(String delimiter) { + properties.put(KEY_PARENT_DELIMITER, delimiter); + return this; + } + + /** + * Set a custom property. + * + * @param key the property key + * @param value the property value + * @return this builder instance + */ + public Builder property(String key, String value) { + properties.put(key, value); + return this; + } + + /** + * Add all properties from a map. + * + * @param props the properties map to add + * @return this builder instance + */ + public Builder properties(Map props) { + properties.putAll(props); + return this; + } + + /** + * Build the configuration instance. + * + * @return a new LanceNamespaceConfig instance + * @throws IllegalArgumentException if required properties are missing or invalid + */ + public LanceNamespaceConfig build() { + return new LanceNamespaceConfig(properties); + } + } +} diff --git a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java new file mode 100644 index 0000000..2bf08d5 --- /dev/null +++ b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java @@ -0,0 +1,616 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Lance Namespace Adapter Integration Test. + * + * This test class covers all CRUD operations of LanceNamespaceAdapter for Table API, + * including complete lifecycle management of table creation, query, update and deletion. + * + * Test scope: + * - Namespace management: create, list, check, delete + * - Table management: create, query, check, delete + * - Metadata operations: get namespace and table metadata + * - Error handling: duplicate creation, non-existing resources and other exception scenarios + */ +@DisplayName("Lance Namespace Adapter Integration Test") +class LanceNamespaceAdapterITCase { + + @TempDir + Path tempDir; + + private LanceNamespaceAdapter adapter; + private String warehousePath; + + /** + * Setup before test. + */ + @BeforeEach + void setUp() { + warehousePath = tempDir.resolve("warehouse").toString(); + + // Create configuration + Map properties = new HashMap<>(); + properties.put(LanceNamespaceConfig.KEY_IMPL, "dir"); + properties.put(LanceNamespaceConfig.KEY_ROOT, warehousePath); + + // Create adapter instance + adapter = LanceNamespaceAdapter.create(properties); + adapter.init(); + } + + /** + * Cleanup after test. + */ + @AfterEach + void tearDown() throws Exception { + if (adapter != null) { + adapter.close(); + } + } + + // ==================== Namespace Management Tests ==================== + + /** + * Test namespace creation (Create). + */ + @Test + @DisplayName("Test creating namespace") + void testCreateNamespace() { + // Prepare + String namespaceName = "test_db"; + Map properties = new HashMap<>(); + properties.put("description", "Test database"); + + // Execute + adapter.createNamespace(properties, namespaceName); + + // Verify + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + + // Verify metadata + Map metadata = adapter.getNamespaceMetadata(namespaceName); + assertThat(metadata).isNotNull(); + } + + /** + * Test creating nested namespace. + */ + @Test + @DisplayName("Test creating nested namespace") + void testCreateNestedNamespace() { + // Prepare + String parentNamespace = "parent_db"; + String childNamespace = "child_db"; + + // Execute + adapter.createNamespace(new HashMap<>(), parentNamespace); + adapter.createNamespace(new HashMap<>(), parentNamespace, childNamespace); + + // Verify + assertThat(adapter.namespaceExists(parentNamespace)).isTrue(); + assertThat(adapter.namespaceExists(parentNamespace, childNamespace)).isTrue(); + } + + /** + * Test listing namespaces (Read). + */ + @Test + @DisplayName("Test listing all top-level namespaces") + void testListNamespaces() { + // Prepare + adapter.createNamespace(new HashMap<>(), "db1"); + adapter.createNamespace(new HashMap<>(), "db2"); + adapter.createNamespace(new HashMap<>(), "db3"); + + // Execute + List namespaces = adapter.listNamespaces(); + + // Verify + assertThat(namespaces).isNotNull(); + assertThat(namespaces).contains("db1", "db2", "db3"); + assertThat(namespaces.size()).isGreaterThanOrEqualTo(3); + } + + /** + * Test listing child namespaces. + */ + @Test + @DisplayName("Test listing child namespaces") + void testListChildNamespaces() { + // Prepare + String parent = "my_warehouse"; + adapter.createNamespace(new HashMap<>(), parent); + adapter.createNamespace(new HashMap<>(), parent, "schema1"); + adapter.createNamespace(new HashMap<>(), parent, "schema2"); + + // Execute + List childNamespaces = adapter.listNamespaces(parent); + + // Verify + assertThat(childNamespaces).isNotNull(); + assertThat(childNamespaces).contains("schema1", "schema2"); + } + + /** + * Test checking namespace existence (Read). + */ + @Test + @DisplayName("Test checking namespace existence") + void testNamespaceExists() { + // Prepare + String namespaceName = "existing_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + // Execute and verify + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + assertThat(adapter.namespaceExists("non_existing_db")).isFalse(); + } + + /** + * Test dropping namespace (Delete). + */ + @Test + @DisplayName("Test dropping namespace") + void testDropNamespace() { + // Prepare + String namespaceName = "temp_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + + // Execute + adapter.dropNamespace(false, namespaceName); + + // Verify + assertThat(adapter.namespaceExists(namespaceName)).isFalse(); + } + + /** + * Test dropping namespace (cascade delete). + */ + @Test + @DisplayName("Test cascade dropping namespace and its contents") + void testDropNamespaceCascade() { + // Prepare + String namespaceName = "cascade_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + // Execute + adapter.dropNamespace(true, namespaceName); + + // Verify + assertThat(adapter.namespaceExists(namespaceName)).isFalse(); + } + + /** + * Test getting namespace metadata (Read). + */ + @Test + @DisplayName("Test getting namespace metadata") + void testGetNamespaceMetadata() { + // Prepare + String namespaceName = "metadata_db"; + Map properties = new HashMap<>(); + properties.put("owner", "admin"); + properties.put("environment", "test"); + + adapter.createNamespace(properties, namespaceName); + + // Execute + Map metadata = adapter.getNamespaceMetadata(namespaceName); + + // Verify + assertThat(metadata).isNotNull(); + assertThat(metadata).containsKeys("owner", "environment"); + } + + // ==================== Table Management Tests ==================== + + /** + * Test creating table (Create). + */ + @Test + @DisplayName("Test creating table in namespace") + void testCreateTable() { + // Prepare + String namespaceName = "my_db"; + String tableName = "my_table"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + Map tableProperties = new HashMap<>(); + tableProperties.put("format", "lance"); + + // Execute + adapter.createEmptyTable(tableLocation, tableProperties, namespaceName, tableName); + + // Verify + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + } + + /** + * Test creating multiple tables. + */ + @Test + @DisplayName("Test creating multiple tables in same namespace") + void testCreateMultipleTables() { + // Prepare + String namespaceName = "test_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String[] tableNames = {"users", "products", "orders", "analytics"}; + + // Execute + for (String tableName : tableNames) { + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + } + + // Verify + List tables = adapter.listTables(namespaceName); + assertThat(tables).isNotNull(); + assertThat(tables).contains(tableNames); + } + + /** + * Test listing tables (Read). + */ + @Test + @DisplayName("Test listing all tables in namespace") + void testListTables() { + // Prepare + String namespaceName = "query_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + adapter.createEmptyTable( + warehousePath + "/" + namespaceName + "/table1", + new HashMap<>(), + namespaceName, "table1" + ); + adapter.createEmptyTable( + warehousePath + "/" + namespaceName + "/table2", + new HashMap<>(), + namespaceName, "table2" + ); + + // Execute + List tables = adapter.listTables(namespaceName); + + // Verify + assertThat(tables).isNotNull(); + assertThat(tables).contains("table1", "table2"); + assertThat(tables.size()).isGreaterThanOrEqualTo(2); + } + + /** + * Test checking table existence (Read). + */ + @Test + @DisplayName("Test checking table existence") + void testTableExists() { + // Prepare + String namespaceName = "check_db"; + String tableName = "check_table"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + + // Execute and verify + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + assertThat(adapter.tableExists(namespaceName, "non_existing_table")).isFalse(); + } + + /** + * Test getting table metadata (Read). + */ + @Test + @DisplayName("Test getting table metadata") + void testGetTableMetadata() { + // Prepare + String namespaceName = "metadata_db"; + String tableName = "metadata_table"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + Map tableProperties = new HashMap<>(); + tableProperties.put("format", "lance"); + tableProperties.put("index", "ivf"); + + adapter.createEmptyTable(tableLocation, tableProperties, namespaceName, tableName); + + // Execute + AbstractLanceNamespaceAdapter.TableMetadata metadata = + adapter.getTableMetadata(namespaceName, tableName); + + // Verify + assertThat(metadata).isNotNull(); + assertThat(metadata.getLocation()).isNotNull(); + assertThat(metadata.getStorageOptions()).isNotNull(); + } + + /** + * Test dropping table (Delete). + */ + @Test + @DisplayName("Test dropping table") + void testDropTable() { + // Prepare + String namespaceName = "drop_db"; + String tableName = "drop_table"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + + // Execute + adapter.dropTable(namespaceName, tableName); + + // Verify + assertThat(adapter.tableExists(namespaceName, tableName)).isFalse(); + } + + /** + * Test dropping multiple tables. + */ + @Test + @DisplayName("Test dropping multiple tables in namespace") + void testDropMultipleTables() { + // Prepare + String namespaceName = "cleanup_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + String[] tableNames = {"temp1", "temp2", "temp3"}; + for (String tableName : tableNames) { + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + } + + // Verify creation successful + for (String tableName : tableNames) { + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + } + + // Execute - drop all tables + for (String tableName : tableNames) { + adapter.dropTable(namespaceName, tableName); + } + + // Verify - all tables dropped + for (String tableName : tableNames) { + assertThat(adapter.tableExists(namespaceName, tableName)).isFalse(); + } + } + + // ==================== Comprehensive Scenario Tests ==================== + + /** + * Test complete CRUD lifecycle. + */ + @Test + @DisplayName("Test complete table CRUD lifecycle") + void testCompleteTableCrudLifecycle() { + // 1. Create - create namespace + String namespaceName = "complete_db"; + Map dbProps = new HashMap<>(); + dbProps.put("owner", "admin"); + adapter.createNamespace(dbProps, namespaceName); + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + + // 2. Create - create table + String tableName = "complete_table"; + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + Map tableProps = new HashMap<>(); + tableProps.put("format", "lance"); + adapter.createEmptyTable(tableLocation, tableProps, namespaceName, tableName); + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + + // 3. Read - list tables + List tables = adapter.listTables(namespaceName); + assertThat(tables).contains(tableName); + + // 4. Read - get table metadata + AbstractLanceNamespaceAdapter.TableMetadata metadata = + adapter.getTableMetadata(namespaceName, tableName); + assertThat(metadata.getLocation()).contains(tableName); + + // 5. Delete - drop table + adapter.dropTable(namespaceName, tableName); + assertThat(adapter.tableExists(namespaceName, tableName)).isFalse(); + + // 6. Delete - drop namespace + adapter.dropNamespace(true, namespaceName); + assertThat(adapter.namespaceExists(namespaceName)).isFalse(); + } + + /** + * Test multiple namespace independence. + */ + @Test + @DisplayName("Test independence of multiple namespaces") + void testMultipleNamespaceIndependence() { + // Prepare - create multiple independent namespaces + String db1 = "database1"; + String db2 = "database2"; + String tableName = "test_table"; + + adapter.createNamespace(new HashMap<>(), db1); + adapter.createNamespace(new HashMap<>(), db2); + + // Create table in db1 + adapter.createEmptyTable( + warehousePath + "/" + db1 + "/" + tableName, + new HashMap<>(), + db1, tableName + ); + + // Create table with same name in db2 + adapter.createEmptyTable( + warehousePath + "/" + db2 + "/" + tableName, + new HashMap<>(), + db2, tableName + ); + + // Verify - both tables exist independently + assertThat(adapter.tableExists(db1, tableName)).isTrue(); + assertThat(adapter.tableExists(db2, tableName)).isTrue(); + + // Verify - dropping table in db1 doesn't affect db2 + adapter.dropTable(db1, tableName); + assertThat(adapter.tableExists(db1, tableName)).isFalse(); + assertThat(adapter.tableExists(db2, tableName)).isTrue(); + } + + /** + * Test special character support in table and namespace names. + */ + @Test + @DisplayName("Test naming with underscores and numbers") + void testSpecialCharacterNaming() { + // Prepare + String namespaceName = "test_db_123"; + String tableName = "data_table_v2_001"; + + adapter.createNamespace(new HashMap<>(), namespaceName); + + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + + // Verify + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + assertThat(adapter.tableExists(namespaceName, tableName)).isTrue(); + } + + /** + * Test table quantity and performance. + */ + @Test + @DisplayName("Test creating many tables in single namespace") + void testCreateManyTables() { + // Prepare + String namespaceName = "scale_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + int tableCount = 50; + + // Execute - create multiple tables + for (int i = 0; i < tableCount; i++) { + String tableName = "table_" + String.format("%03d", i); + String tableLocation = warehousePath + "/" + namespaceName + "/" + tableName; + adapter.createEmptyTable(tableLocation, new HashMap<>(), namespaceName, tableName); + } + + // Verify + List tables = adapter.listTables(namespaceName); + assertThat(tables).isNotNull(); + assertThat(tables.size()).isGreaterThanOrEqualTo(tableCount); + } + + /** + * Test exception scenario - creating existing namespace. + */ + @Test + @DisplayName("Test creating existing namespace throws exception") + void testCreateExistingNamespaceThrowsException() { + // Prepare + String namespaceName = "existing_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + // Execute and verify - expect exception + assertThatThrownBy(() -> + adapter.createNamespace(new HashMap<>(), namespaceName) + ).isNotNull(); + } + + /** + * Test exception scenario - dropping non-existing namespace. + */ + @Test + @DisplayName("Test dropping non-existing namespace throws exception") + void testDropNonExistingNamespaceThrowsException() { + // Execute and verify - expect exception + assertThatThrownBy(() -> + adapter.dropNamespace(false, "non_existing_db") + ).isNotNull(); + } + + /** + * Test exception scenario - dropping non-existing table. + */ + @Test + @DisplayName("Test dropping non-existing table throws exception") + void testDropNonExistingTableThrowsException() { + // Prepare + String namespaceName = "error_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + // Execute and verify - expect exception + assertThatThrownBy(() -> + adapter.dropTable(namespaceName, "non_existing_table") + ).isNotNull(); + } + + /** + * Test resource cleanup and closing. + */ + @Test + @DisplayName("Test adapter closes correctly and releases resources") + void testAdapterCloseAndResourceCleanup() throws Exception { + // Prepare + String namespaceName = "cleanup_db"; + adapter.createNamespace(new HashMap<>(), namespaceName); + + // Verify operation works + assertThat(adapter.namespaceExists(namespaceName)).isTrue(); + + // Execute - close adapter + adapter.close(); + + // Verify - resources released correctly when creating new adapter + Map properties = new HashMap<>(); + properties.put(LanceNamespaceConfig.KEY_IMPL, "dir"); + properties.put(LanceNamespaceConfig.KEY_ROOT, warehousePath); + + LanceNamespaceAdapter newAdapter = LanceNamespaceAdapter.create(properties); + newAdapter.init(); + + try { + // Verify previous operations preserved + assertThat(newAdapter.namespaceExists(namespaceName)).isTrue(); + } finally { + newAdapter.close(); + } + } +} diff --git a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java new file mode 100644 index 0000000..9326bdc --- /dev/null +++ b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java @@ -0,0 +1,327 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Mock implementation of Lance Namespace for testing purposes. + * + * This mock allows tests to run without requiring the actual Lance Namespace + * library to be available. It simulates the basic behavior of the real API. + */ +public class MockLanceNamespace { + + /** + * Storage for namespaces and their properties. + * Map structure: namespace_path -> properties + */ + private final Map> namespaces = new HashMap<>(); + + /** + * Storage for tables and their properties. + * Map structure: table_path -> properties + */ + private final Map> tables = new HashMap<>(); + + /** + * Check if namespace exists locally. + */ + private boolean hasNamespace(String id) { + return namespaces.containsKey(id); + } + + /** + * Check if table exists locally. + */ + private boolean hasTable(String id) { + return tables.containsKey(id); + } + + /** + * Create a mock namespace instance. + */ + public static MockLanceNamespace connect(String impl, String location, Object allocator) { + return new MockLanceNamespace(); + } + + /** + * Mock: Create a namespace. + */ + public void createNamespace(Object request) { + String id = extractId(request); + if (hasNamespace(id)) { + throw new RuntimeException("Namespace already exists: " + id); + } + Map props = extractProperties(request); + namespaces.put(id, props != null ? new HashMap<>(props) : new HashMap<>()); + } + + /** + * Mock: List namespaces. + */ + public Object listNamespaces(Object request) { + Set result = new HashSet<>(namespaces.keySet()); + return createResponse(result); + } + + /** + * Mock: Check if namespace exists. + */ + public Object namespaceExists(Object request) { + String id = extractId(request); + if (!hasNamespace(id)) { + throw new RuntimeException("Namespace does not exist: " + id); + } + return null; + } + + /** + * Mock: Drop a namespace. + */ + public void dropNamespace(Object request) { + String id = extractId(request); + if (!hasNamespace(id)) { + throw new RuntimeException("Namespace does not exist: " + id); + } + namespaces.remove(id); + } + + /** + * Mock: Describe a namespace (get metadata). + */ + public Object describeNamespace(Object request) { + String id = extractId(request); + if (!hasNamespace(id)) { + throw new RuntimeException("Namespace does not exist: " + id); + } + Map props = namespaces.get(id); + return createDescribeResponse(props); + } + + /** + * Mock: Create an empty table. + */ + public void createEmptyTable(Object request) { + String id = extractId(request); + if (hasTable(id)) { + throw new RuntimeException("Table already exists: " + id); + } + Map props = extractProperties(request); + tables.put(id, props != null ? new HashMap<>(props) : new HashMap<>()); + } + + /** + * Mock: List tables. + */ + public Object listTables(Object request) { + Set result = new HashSet<>(tables.keySet()); + return createResponse(result); + } + + /** + * Mock: Check if table exists. + */ + public Object tableExists(Object request) { + String id = extractId(request); + if (!hasTable(id)) { + throw new RuntimeException("Table does not exist: " + id); + } + return null; + } + + /** + * Mock: Drop a table. + */ + public void dropTable(Object request) { + String id = extractId(request); + if (!hasTable(id)) { + throw new RuntimeException("Table does not exist: " + id); + } + tables.remove(id); + } + + /** + * Mock: Describe a table (get metadata). + */ + public Object describeTable(Object request) { + String id = extractId(request); + if (!hasTable(id)) { + throw new RuntimeException("Table does not exist: " + id); + } + Map props = tables.get(id); + return createTableDescribeResponse(props); + } + + /** + * Close the connection. + */ + public void close() throws Exception { + // Cleanup + namespaces.clear(); + tables.clear(); + } + + // ==================== Helper Methods ==================== + + /** + * Extract ID from request object. + */ + private String extractId(Object request) { + if (request == null) { + return ""; + } + try { + Object id = request.getClass() + .getMethod("getId") + .invoke(request); + if (id instanceof java.util.List) { + java.util.List list = (java.util.List) id; + return list.isEmpty() ? "" : list.get(0).toString(); + } + return id != null ? id.toString() : ""; + } catch (Exception e) { + return ""; + } + } + + /** + * Extract properties from request object. + */ + private Map extractProperties(Object request) { + if (request == null) { + return null; + } + try { + Object props = request.getClass() + .getMethod("getProperties") + .invoke(request); + if (props instanceof Map) { + return (Map) props; + } + return null; + } catch (Exception e) { + return null; + } + } + + /** + * Create a response object with namespaces. + */ + private Object createResponse(Set namespaces) { + try { + Class responseClass = Class.forName( + "org.lance.namespace.model.ListNamespacesResponse"); + Object response = responseClass.getDeclaredConstructor().newInstance(); + + // Try to set the namespaces using reflection + try { + responseClass.getMethod("setNamespaces", Set.class) + .invoke(response, namespaces); + } catch (Exception e) { + // If setNamespaces fails, try setting as field + java.lang.reflect.Field field = responseClass.getDeclaredField("namespaces"); + field.setAccessible(true); + field.set(response, namespaces); + } + return response; + } catch (Exception e) { + // Fallback: return a simple object with getNamespaces method + return new Object() { + public Set getNamespaces() { + return namespaces; + } + }; + } + } + + /** + * Create a describe response object. + */ + private Object createDescribeResponse(Map properties) { + try { + Class responseClass = Class.forName( + "org.lance.namespace.model.DescribeNamespaceResponse"); + Object response = responseClass.getDeclaredConstructor().newInstance(); + + try { + responseClass.getMethod("setProperties", Map.class) + .invoke(response, properties); + } catch (Exception e) { + // Fallback to field + java.lang.reflect.Field field = responseClass.getDeclaredField("properties"); + field.setAccessible(true); + field.set(response, properties); + } + return response; + } catch (Exception e) { + // Fallback + return new Object() { + public Map getProperties() { + return properties; + } + }; + } + } + + /** + * Create a table describe response object. + */ + private Object createTableDescribeResponse(Map properties) { + try { + Class responseClass = Class.forName( + "org.lance.namespace.model.DescribeTableResponse"); + Object response = responseClass.getDeclaredConstructor().newInstance(); + + try { + responseClass.getMethod("setProperties", Map.class) + .invoke(response, properties); + } catch (Exception e) { + // Fallback to field + java.lang.reflect.Field field = responseClass.getDeclaredField("properties"); + field.setAccessible(true); + field.set(response, properties); + } + + // Try to set table_path + try { + responseClass.getMethod("setTable_path", String.class) + .invoke(response, "/path/to/table"); + } catch (Exception ignored) { + // Optional field + } + + return response; + } catch (Exception e) { + // Fallback + return new Object() { + public Map getProperties() { + return properties; + } + + public String getTable_path() { + return "/path/to/table"; + } + }; + } + } +} From 9483bce23fb5ea6ba642a4509d138e6cd284c8dd Mon Sep 17 00:00:00 2001 From: kaori-seasons Date: Fri, 27 Feb 2026 11:11:16 +0800 Subject: [PATCH 2/3] feat(catalog): implement Lance Namespace Catalog integration - Add BaseLanceNamespaceCatalog with full Flink Catalog API support - Add LanceCatalogFactory for creating namespace adapters - Refactor LanceNamespaceAdapter with improved CRUD operations - Update LanceNamespaceConfig with extra_level and parent support --- .../namespace/BaseLanceNamespaceCatalog.java | 475 +++++++++++++++++- .../namespace/LanceCatalogFactory.java | 176 +++++++ .../namespace/LanceNamespaceAdapter.java | 265 ++++++---- .../namespace/LanceNamespaceConfig.java | 233 +-------- 4 files changed, 798 insertions(+), 351 deletions(-) create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceCatalogFactory.java diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java index 07bf1e6..c288338 100644 --- a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/BaseLanceNamespaceCatalog.java @@ -20,60 +20,325 @@ import org.apache.flink.table.catalog.AbstractCatalog; import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogPartition; +import org.apache.flink.table.catalog.CatalogFunction; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; /** - * Base class for Lance Catalog implementation integrated with Lance Namespace. - * - * This class provides the foundation for catalog operations using a namespace adapter, - * supporting multi-level namespace hierarchies and flexible backend implementations. + * Base Lance Catalog implementation integrated with Lance Namespace. */ public abstract class BaseLanceNamespaceCatalog extends AbstractCatalog { - protected AbstractLanceNamespaceAdapter adapter; + private static final Logger LOG = LoggerFactory.getLogger(BaseLanceNamespaceCatalog.class); + + protected LanceNamespaceAdapter namespaceAdapter; protected LanceNamespaceConfig config; + protected Optional extraLevel; + protected Optional parentPrefix; - public BaseLanceNamespaceCatalog(String catalogName, - AbstractLanceNamespaceAdapter adapter, - LanceNamespaceConfig config) { + public BaseLanceNamespaceCatalog(String catalogName, LanceNamespaceAdapter adapter, LanceNamespaceConfig config) { super(catalogName, "default"); - this.adapter = adapter; - this.config = config; + + this.namespaceAdapter = Objects.requireNonNull(adapter, "Namespace adapter cannot be null"); + this.config = Objects.requireNonNull(config, "Configuration cannot be null"); + + LOG.info("Initializing BaseLanceNamespaceCatalog: {}", catalogName); + + // Configure extra level + if (config.getExtraLevel().isPresent()) { + this.extraLevel = config.getExtraLevel(); + } else if (config.isDirectoryNamespace()) { + this.extraLevel = Optional.of("default"); + } else { + this.extraLevel = Optional.empty(); + } + + // Configure parent prefix + this.parentPrefix = config.getParentArray(); + + LOG.info("Catalog configuration - impl: {}, extraLevel: {}, parentPrefix: {}", + config.getImpl(), extraLevel, parentPrefix); + } + + // ========== Database Operations ========== + + @Override + public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) + throws DatabaseAlreadyExistException, CatalogException { + + LOG.info("Creating database: {} (ignoreIfExists={})", name, ignoreIfExists); + + try { + if (databaseExists(name)) { + if (ignoreIfExists) { + LOG.info("Database already exists, skipping creation: {}", name); + return; + } else { + throw new DatabaseAlreadyExistException(getName(), name); + } + } + + String[] namespacePath = transformDatabaseNameToNamespace(name); + Map properties = database.getProperties(); + namespaceAdapter.createNamespace(properties, namespacePath); + + LOG.info("Database created successfully: {}", name); + } catch (DatabaseAlreadyExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to create database: {}", name, e); + throw new CatalogException("Failed to create database: " + name, e); + } + } + + @Override + public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) + throws DatabaseNotExistException, CatalogException { + + LOG.info("Dropping database: {} (cascade={})", name, cascade); + + try { + if (!databaseExists(name)) { + if (ignoreIfNotExists) { + LOG.info("Database does not exist, skipping drop: {}", name); + return; + } else { + throw new DatabaseNotExistException(getName(), name); + } + } + + String[] namespacePath = transformDatabaseNameToNamespace(name); + namespaceAdapter.dropNamespace(cascade, namespacePath); + + LOG.info("Database dropped successfully: {}", name); + } catch (DatabaseNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to drop database: {}", name, e); + throw new CatalogException("Failed to drop database: " + name, e); + } + } + + @Override + public List listDatabases() throws CatalogException { + LOG.debug("Listing databases"); + + try { + return namespaceAdapter.listNamespaces(); + } catch (Exception e) { + LOG.error("Failed to list databases", e); + throw new CatalogException("Failed to list databases", e); + } + } + + @Override + public CatalogDatabase getDatabase(String name) + throws DatabaseNotExistException, CatalogException { + + LOG.debug("Getting database: {}", name); + + try { + if (!databaseExists(name)) { + throw new DatabaseNotExistException(getName(), name); + } + + String[] namespacePath = transformDatabaseNameToNamespace(name); + Map metadata = namespaceAdapter.getNamespaceMetadata(namespacePath); + + return new org.apache.flink.table.catalog.CatalogDatabaseImpl(metadata, ""); + } catch (DatabaseNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to get database: {}", name, e); + throw new CatalogException("Failed to get database: " + name, e); + } + } + + @Override + public boolean databaseExists(String name) { + LOG.debug("Checking if database exists: {}", name); + + try { + String[] namespacePath = transformDatabaseNameToNamespace(name); + return namespaceAdapter.namespaceExists(namespacePath); + } catch (Exception e) { + LOG.debug("Error checking database existence: {}", name, e); + return false; + } + } + + // ========== Table Operations ========== + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException, CatalogException { + + LOG.info("Creating table: {} (ignoreIfExists={})", tablePath, ignoreIfExists); + + try { + String dbName = tablePath.getDatabaseName(); + String tblName = tablePath.getObjectName(); + + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + + if (tableExists(tablePath)) { + if (ignoreIfExists) { + LOG.info("Table already exists, skipping creation: {}", tablePath); + return; + } else { + throw new TableAlreadyExistException(getName(), tablePath); + } + } + + String[] tableId = transformTableNameToId(dbName, tblName); + Map properties = table.getOptions(); + namespaceAdapter.createEmptyTable(null, properties, tableId); + + LOG.info("Table created successfully: {}", tablePath); + } catch (TableAlreadyExistException | DatabaseNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to create table: {}", tablePath, e); + throw new CatalogException("Failed to create table: " + tablePath, e); + } + } + + @Override + public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + + LOG.info("Dropping table: {}", tablePath); + + try { + if (!tableExists(tablePath)) { + if (ignoreIfNotExists) { + LOG.info("Table does not exist, skipping drop: {}", tablePath); + return; + } else { + throw new TableNotExistException(getName(), tablePath); + } + } + + String[] tableId = transformTableNameToId(tablePath.getDatabaseName(), tablePath.getObjectName()); + namespaceAdapter.dropTable(tableId); + + LOG.info("Table dropped successfully: {}", tablePath); + } catch (TableNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to drop table: {}", tablePath, e); + throw new CatalogException("Failed to drop table: " + tablePath, e); + } + } + + @Override + public List listTables(String databaseName) + throws DatabaseNotExistException, CatalogException { + + LOG.debug("Listing tables in database: {}", databaseName); + + try { + if (!databaseExists(databaseName)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + + String[] namespacePath = transformDatabaseNameToNamespace(databaseName); + return namespaceAdapter.listTables(namespacePath); + } catch (DatabaseNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to list tables in database: {}", databaseName, e); + throw new CatalogException("Failed to list tables in database: " + databaseName, e); + } } - /** - * Abstract method to be implemented by subclasses to create CatalogTable from metadata. - */ + @Override + public CatalogTable getTable(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + + LOG.debug("Getting table: {}", tablePath); + + try { + if (!tableExists(tablePath)) { + throw new TableNotExistException(getName(), tablePath); + } + + String[] tableId = transformTableNameToId(tablePath.getDatabaseName(), tablePath.getObjectName()); + LanceNamespaceAdapter.TableMetadata metadata = namespaceAdapter.getTableMetadata(tableId); + + return createCatalogTable(tablePath.getDatabaseName(), tablePath.getObjectName(), metadata); + } catch (TableNotExistException e) { + throw e; + } catch (Exception e) { + LOG.error("Failed to get table: {}", tablePath, e); + throw new CatalogException("Failed to get table: " + tablePath, e); + } + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + LOG.debug("Checking if table exists: {}", tablePath); + + try { + String[] tableId = transformTableNameToId(tablePath.getDatabaseName(), tablePath.getObjectName()); + return namespaceAdapter.tableExists(tableId); + } catch (Exception e) { + LOG.debug("Error checking table existence: {}", tablePath, e); + return false; + } + } + + // ========== Abstract method to be implemented by subclasses ========== + protected abstract CatalogTable createCatalogTable( String databaseName, String tableName, - AbstractLanceNamespaceAdapter.TableMetadata metadata) throws CatalogException; + LanceNamespaceAdapter.TableMetadata metadata) throws CatalogException; + + // ========== Helper methods ========== - /** - * Transform database name to namespace path array. - */ protected String[] transformDatabaseNameToNamespace(String databaseName) { - java.util.Optional parentPrefix = config.getParentArray(); - java.util.Optional extraLevel = config.getExtraLevel(); + String[] baseNamespace = new String[] {databaseName}; if (parentPrefix.isPresent()) { String[] parent = parentPrefix.get(); - String[] result = new String[parent.length + 1]; + String[] result = new String[parent.length + baseNamespace.length]; System.arraycopy(parent, 0, result, 0, parent.length); - result[parent.length] = databaseName; + System.arraycopy(baseNamespace, 0, result, parent.length, baseNamespace.length); return result; } else if (extraLevel.isPresent()) { - return new String[]{extraLevel.get(), databaseName}; + String[] result = new String[baseNamespace.length + 1]; + result[0] = extraLevel.get(); + System.arraycopy(baseNamespace, 0, result, 1, baseNamespace.length); + return result; } else { - return new String[]{databaseName}; + return baseNamespace; } } - /** - * Transform table name to full table ID (namespace path + table name). - */ protected String[] transformTableNameToId(String databaseName, String tableName) { String[] dbPath = transformDatabaseNameToNamespace(databaseName); String[] result = new String[dbPath.length + 1]; @@ -81,4 +346,162 @@ protected String[] transformTableNameToId(String databaseName, String tableName) result[dbPath.length] = tableName; return result; } + + // ========== Not implemented - Partition operations not supported ========== + + @Override + public void createPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, + CatalogPartition partition, boolean ignoreIfExists) + throws TableNotExistException, PartitionAlreadyExistsException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, + boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public List listPartitions(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public void alterPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, + CatalogPartition newPartition, boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public List listPartitionsByFilter(ObjectPath tablePath, List filters) + throws CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public void alterTableStatistics(ObjectPath tablePath, CatalogTableStatistics tableStatistics, + boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + LOG.debug("Alter table statistics not supported: {}", tablePath); + } + + @Override + public void alterPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, + CatalogTableStatistics partitionStatistics, boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public void alterTableColumnStatistics(ObjectPath tablePath, CatalogColumnStatistics columnStatistics, + boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + LOG.debug("Alter table column statistics not supported: {}", tablePath); + } + + @Override + public void alterPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, + CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + LOG.debug("Get table statistics not supported: {}", tablePath); + return null; + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + LOG.debug("Get table column statistics not supported: {}", tablePath); + return null; + } + + @Override + public CatalogTableStatistics getPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException, CatalogException { + throw new CatalogException("Partition operations are not supported"); + } + + @Override + public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + LOG.debug("Alter table not supported: {}", tablePath); + } + + @Override + public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) + throws CatalogException { + LOG.debug("Rename table not supported: {} -> {}", tablePath, newTableName); + } + + @Override + public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists) + throws DatabaseNotExistException, CatalogException { + LOG.debug("Alter database not supported: {}", name); + } + + // Note: renameDatabase is not part of the standard Flink Catalog interface + // If needed, it should be implemented by subclasses + + @Override + public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } + + @Override + public void createFunction(ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } + + @Override + public void alterFunction(ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } + + @Override + public List listFunctions(String databaseName) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } + + @Override + public CatalogFunction getFunction(ObjectPath functionPath) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } + + @Override + public boolean functionExists(ObjectPath functionPath) + throws CatalogException { + throw new CatalogException("Function operations are not supported"); + } } diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceCatalogFactory.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceCatalogFactory.java new file mode 100644 index 0000000..159f9aa --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceCatalogFactory.java @@ -0,0 +1,176 @@ +/* + * 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.flink.connector.lance.catalog.namespace; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +/** + * Factory for creating and managing Lance Namespace Adapters and Catalogs. + * + * This factory follows the factory pattern used in lance-spark, providing + * a unified way to create LanceNamespaceAdapter instances with different + * implementations (DirectoryNamespace, RestNamespace, etc.). + * + * Example usage: + * 
+ * LanceCatalogFactory factory = new LanceCatalogFactory();
+ * LanceNamespaceConfig config = LanceNamespaceConfig.builder()
+ *     .impl("dir")
+ *     .root("/data/lance")
+ *     .build();
+ * LanceNamespaceAdapter adapter = factory.createAdapter(config);
+ *
+ */ +public class LanceCatalogFactory { + + private static final Logger LOG = LoggerFactory.getLogger(LanceCatalogFactory.class); + + private volatile BufferAllocator sharedAllocator; + + /** + * Create a new factory with a shared buffer allocator. + */ + public LanceCatalogFactory() { + this.sharedAllocator = new RootAllocator(); + LOG.info("Created LanceCatalogFactory with shared RootAllocator"); + } + + /** + * Create a new factory with a custom buffer allocator. + */ + public LanceCatalogFactory(BufferAllocator allocator) { + this.sharedAllocator = Objects.requireNonNull(allocator, "Allocator cannot be null"); + LOG.info("Created LanceCatalogFactory with custom allocator"); + } + + /** + * Create a LanceNamespaceAdapter from configuration. + * + * @param config The namespace configuration + * @return A configured LanceNamespaceAdapter + */ + public LanceNamespaceAdapter createAdapter(LanceNamespaceConfig config) { + LOG.info("Creating LanceNamespaceAdapter with config"); + + Objects.requireNonNull(config, "Configuration cannot be null"); + + try { + Map properties = new HashMap<>(); + properties.put(LanceNamespaceConfig.KEY_IMPL, config.getImpl()); + if (config.getRoot().isPresent()) { + properties.put(LanceNamespaceConfig.KEY_ROOT, config.getRoot().get()); + } + if (config.getUri().isPresent()) { + properties.put(LanceNamespaceConfig.KEY_URI, config.getUri().get()); + } + if (config.getExtraLevel().isPresent()) { + properties.put(LanceNamespaceConfig.KEY_EXTRA_LEVEL, config.getExtraLevel().get()); + } + if (config.getParent().isPresent()) { + properties.put(LanceNamespaceConfig.KEY_PARENT, config.getParent().get()); + } + + return LanceNamespaceAdapter.create(properties); + } catch (Exception e) { + LOG.error("Failed to create adapter", e); + throw new RuntimeException("Failed to create LanceNamespaceAdapter", e); + } + } + + /** + * Create a LanceNamespaceAdapter from properties map. + * + * @param properties The properties map + * @return A configured LanceNamespaceAdapter + */ + public LanceNamespaceAdapter createAdapter(Map properties) { + LOG.info("Creating LanceNamespaceAdapter from properties"); + + Objects.requireNonNull(properties, "Properties cannot be null"); + + LanceNamespaceConfig config = LanceNamespaceConfig.from(properties); + return createAdapter(config); + } + + /** + * Create a LanceNamespaceAdapter with directory namespace implementation. + * + * @param rootPath The root directory path + * @return A configured LanceNamespaceAdapter + */ + public LanceNamespaceAdapter createDirectoryAdapter(String rootPath) { + LOG.info("Creating directory namespace adapter with root: {}", rootPath); + + Objects.requireNonNull(rootPath, "Root path cannot be null"); + + LanceNamespaceConfig config = LanceNamespaceConfig.builder() + .impl("dir") + .root(rootPath) + .build(); + + return createAdapter(config); + } + + /** + * Create a LanceNamespaceAdapter with REST namespace implementation. + * + * @param uri The REST service URI + * @return A configured LanceNamespaceAdapter + */ + public LanceNamespaceAdapter createRestAdapter(String uri) { + LOG.info("Creating REST namespace adapter with URI: {}", uri); + + Objects.requireNonNull(uri, "URI cannot be null"); + + LanceNamespaceConfig config = LanceNamespaceConfig.builder() + .impl("rest") + .uri(uri) + .build(); + + return createAdapter(config); + } + + /** + * Get the shared buffer allocator. + */ + public BufferAllocator getSharedAllocator() { + return sharedAllocator; + } + + /** + * Close the factory and cleanup resources. + */ + public void close() { + LOG.info("Closing LanceCatalogFactory"); + try { + if (sharedAllocator != null) { + sharedAllocator.close(); + } + } catch (Exception e) { + LOG.warn("Error closing shared allocator", e); + } + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java index 94ac759..4c463cc 100644 --- a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapter.java @@ -18,6 +18,8 @@ package org.apache.flink.connector.lance.catalog.namespace; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; import org.lance.namespace.LanceNamespace; import org.lance.namespace.model.CreateNamespaceRequest; import org.lance.namespace.model.CreateEmptyTableRequest; @@ -33,9 +35,6 @@ import org.lance.namespace.model.ListTablesResponse; import org.lance.namespace.model.NamespaceExistsRequest; import org.lance.namespace.model.TableExistsRequest; - -import org.apache.arrow.memory.BufferAllocator; -import org.apache.arrow.memory.RootAllocator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,12 +47,12 @@ import java.util.Set; /** - * Lance Namespace Adapter Implementation. + * Adapter for Lance Namespace API. * - * Directly calls Lance Namespace SDK APIs to implement database and table management. - * Supports both local file system and REST backend implementations. + * Provides unified interface for interacting with Lance Namespace, + * supporting both directory-based and REST-based implementations. */ -public class LanceNamespaceAdapter implements AbstractLanceNamespaceAdapter { +public class LanceNamespaceAdapter implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(LanceNamespaceAdapter.class); @@ -62,12 +61,12 @@ public class LanceNamespaceAdapter implements AbstractLanceNamespaceAdapter { private LanceNamespace namespace; public LanceNamespaceAdapter(BufferAllocator allocator, LanceNamespaceConfig config) { - this.allocator = Objects.requireNonNull(allocator, "BufferAllocator cannot be null"); - this.config = Objects.requireNonNull(config, "LanceNamespaceConfig cannot be null"); + this.allocator = Objects.requireNonNull(allocator, "Allocator cannot be null"); + this.config = Objects.requireNonNull(config, "Config cannot be null"); } /** - * Factory method to create Adapter instance. + * Create adapter from properties. */ public static LanceNamespaceAdapter create(Map properties) { LanceNamespaceConfig config = LanceNamespaceConfig.from(properties); @@ -76,72 +75,80 @@ public static LanceNamespaceAdapter create(Map properties) { } /** - * Initialize Lance Namespace connection. - * Directly calls LanceNamespace.connect() method. + * Initialize the namespace connection. */ - @Override public void init() { try { - if (config.isDirectoryNamespace() && config.getRoot().isPresent()) { - // Call: LanceNamespace.connect("file", root_path, allocator) - LOG.info("Initializing local file system namespace: {}", config.getRoot().get()); - namespace = LanceNamespace.connect("file", config.getRoot().get(), allocator); - } else if (config.isRestNamespace() && config.getUri().isPresent()) { - // Call: LanceNamespace.connect("rest", uri, allocator) - LOG.info("Initializing REST namespace: {}", config.getUri().get()); - namespace = LanceNamespace.connect("rest", config.getUri().get(), allocator); - } else { - throw new IllegalArgumentException("Invalid namespace configuration"); + if (namespace != null) { + return; } - - LOG.info("Lance Namespace connection successful"); + + Map properties = new HashMap<>(); + properties.put(LanceNamespaceConfig.KEY_IMPL, config.getImpl()); + config.getRoot().ifPresent(root -> properties.put(LanceNamespaceConfig.KEY_ROOT, root)); + config.getUri().ifPresent(uri -> properties.put(LanceNamespaceConfig.KEY_URI, uri)); + + namespace = LanceNamespace.connect(config.getImpl(), properties, allocator); + LOG.info("LanceNamespace initialized successfully with impl: {}", config.getImpl()); } catch (Exception e) { - LOG.error("Failed to initialize Lance Namespace", e); - throw new RuntimeException("Failed to initialize Lance Namespace", e); + LOG.error("Failed to initialize LanceNamespace", e); + throw new RuntimeException("Failed to initialize LanceNamespace", e); } } /** - * List all top-level namespaces. + * List all namespaces. */ - @Override public List listNamespaces() { - return listNamespaces(new String[0]); + LOG.debug("Listing root level namespaces"); + return listNamespacesRecursive(new ArrayList<>()); } /** - * List child namespaces under a parent namespace. - * Directly calls: LanceNamespace.listNamespaces(ListNamespacesRequest) + * List namespaces under parent. */ - @Override public List listNamespaces(String... parentNamespace) { + LOG.debug("Listing namespaces under: {}", Arrays.toString(parentNamespace)); + return listNamespacesRecursive(Arrays.asList(parentNamespace)); + } + + /** + * Internal recursive method for listing namespaces. + */ + private List listNamespacesRecursive(List parent) { try { + if (namespace == null) { + init(); + } + ListNamespacesRequest request = new ListNamespacesRequest(); - if (parentNamespace.length > 0) { - request.setId(Arrays.asList(parentNamespace)); + if (!parent.isEmpty()) { + request.setId(parent); } ListNamespacesResponse response = namespace.listNamespaces(request); - if (response.getNamespaces() != null) { Set namespaceSet = response.getNamespaces(); return new ArrayList<>(namespaceSet); } return new ArrayList<>(); - } catch (Exception e) { - LOG.warn("Failed to list namespaces", e); + LOG.warn("Failed to list namespaces under: {}", parent, e); return new ArrayList<>(); } } /** * Check if namespace exists. - * Directly calls: LanceNamespace.namespaceExists(NamespaceExistsRequest) */ - @Override public boolean namespaceExists(String... namespaceId) { + LOG.debug("Checking if namespace exists: {}", Arrays.toString(namespaceId)); + try { + if (namespace == null) { + init(); + } + NamespaceExistsRequest request = new NamespaceExistsRequest(); request.setId(Arrays.asList(namespaceId)); @@ -154,100 +161,112 @@ public boolean namespaceExists(String... namespaceId) { } /** - * Create namespace. - * Directly calls: LanceNamespace.createNamespace(CreateNamespaceRequest) + * Create a namespace. */ - @Override public void createNamespace(Map properties, String... namespaceId) { + LOG.info("Creating namespace: {}", Arrays.toString(namespaceId)); + try { + if (namespace == null) { + init(); + } + CreateNamespaceRequest request = new CreateNamespaceRequest(); request.setId(Arrays.asList(namespaceId)); - - if (properties != null && !properties.isEmpty()) { + if (properties != null) { request.setProperties(properties); } namespace.createNamespace(request); - LOG.info("Namespace created successfully: {}", Arrays.toString(namespaceId)); } catch (Exception e) { - LOG.error("Failed to create namespace", e); + LOG.error("Failed to create namespace: {}", Arrays.toString(namespaceId), e); throw new RuntimeException("Failed to create namespace", e); } } /** - * Drop namespace. - * Directly calls: LanceNamespace.dropNamespace(DropNamespaceRequest) + * Drop a namespace. */ - @Override public void dropNamespace(boolean cascade, String... namespaceId) { + LOG.info("Dropping namespace: {} (cascade={})", Arrays.toString(namespaceId), cascade); + try { + if (namespace == null) { + init(); + } + DropNamespaceRequest request = new DropNamespaceRequest(); request.setId(Arrays.asList(namespaceId)); request.setCascade(cascade); namespace.dropNamespace(request); - LOG.info("Namespace dropped successfully: {}", Arrays.toString(namespaceId)); } catch (Exception e) { - LOG.error("Failed to drop namespace", e); + LOG.error("Failed to drop namespace: {}", Arrays.toString(namespaceId), e); throw new RuntimeException("Failed to drop namespace", e); } } /** * Get namespace metadata. - * Directly calls: LanceNamespace.describeNamespace(DescribeNamespaceRequest) */ - @Override public Map getNamespaceMetadata(String... namespaceId) { + LOG.debug("Getting namespace metadata: {}", Arrays.toString(namespaceId)); + try { + if (namespace == null) { + init(); + } + DescribeNamespaceRequest request = new DescribeNamespaceRequest(); request.setId(Arrays.asList(namespaceId)); DescribeNamespaceResponse response = namespace.describeNamespace(request); - - if (response.getProperties() != null) { - return response.getProperties(); - } - return new HashMap<>(); + return response.getProperties() != null ? response.getProperties() : new HashMap<>(); } catch (Exception e) { - LOG.warn("Failed to get namespace metadata", e); + LOG.warn("Failed to get namespace metadata: {}", Arrays.toString(namespaceId), e); return new HashMap<>(); } } /** - * List all tables in namespace. - * Directly calls: LanceNamespace.listTables(ListTablesRequest) + * List tables in a namespace. */ - @Override public List listTables(String... namespaceId) { + LOG.debug("Listing tables in namespace: {}", Arrays.toString(namespaceId)); + try { + if (namespace == null) { + init(); + } + ListTablesRequest request = new ListTablesRequest(); request.setId(Arrays.asList(namespaceId)); ListTablesResponse response = namespace.listTables(request); - if (response.getTables() != null) { Set tableSet = response.getTables(); return new ArrayList<>(tableSet); } return new ArrayList<>(); } catch (Exception e) { - LOG.warn("Failed to list tables", e); + LOG.warn("Failed to list tables in namespace: {}", Arrays.toString(namespaceId), e); return new ArrayList<>(); } } /** * Check if table exists. - * Directly calls: LanceNamespace.tableExists(TableExistsRequest) */ - @Override public boolean tableExists(String... tableId) { + LOG.debug("Checking if table exists: {}", Arrays.toString(tableId)); + try { + if (namespace == null) { + init(); + } + TableExistsRequest request = new TableExistsRequest(); request.setId(Arrays.asList(tableId)); @@ -260,87 +279,110 @@ public boolean tableExists(String... tableId) { } /** - * Create empty table. - * Directly calls: LanceNamespace.createEmptyTable(CreateEmptyTableRequest) + * Create an empty table. */ - @Override public void createEmptyTable(String location, Map properties, String... tableId) { + LOG.info("Creating empty table: {} at {}", Arrays.toString(tableId), location); + try { + if (namespace == null) { + init(); + } + CreateEmptyTableRequest request = new CreateEmptyTableRequest(); request.setId(Arrays.asList(tableId)); - - // Set table location information - if (location != null) { - request.setPath(location); + request.setLocation(location); + if (properties != null) { + request.setProperties(properties); } namespace.createEmptyTable(request); - LOG.info("Table created successfully: {}", Arrays.toString(tableId)); } catch (Exception e) { - LOG.error("Failed to create table", e); + LOG.error("Failed to create table: {}", Arrays.toString(tableId), e); throw new RuntimeException("Failed to create table", e); } } /** - * Drop table. - * Directly calls: LanceNamespace.dropTable(DropTableRequest) + * Drop a table. */ - @Override public void dropTable(String... tableId) { + LOG.info("Dropping table: {}", Arrays.toString(tableId)); + try { + if (namespace == null) { + init(); + } + DropTableRequest request = new DropTableRequest(); request.setId(Arrays.asList(tableId)); namespace.dropTable(request); - LOG.info("Table dropped successfully: {}", Arrays.toString(tableId)); } catch (Exception e) { - LOG.error("Failed to drop table", e); + LOG.error("Failed to drop table: {}", Arrays.toString(tableId), e); throw new RuntimeException("Failed to drop table", e); } } /** * Get table metadata. - * Directly calls: LanceNamespace.describeTable(DescribeTableRequest) */ - @Override public TableMetadata getTableMetadata(String... tableId) { + LOG.debug("Getting table metadata: {}", Arrays.toString(tableId)); + try { + if (namespace == null) { + init(); + } + DescribeTableRequest request = new DescribeTableRequest(); request.setId(Arrays.asList(tableId)); DescribeTableResponse response = namespace.describeTable(request); - String location = "/path/to/table"; - Map options = new HashMap<>(); - - // Call API to get table path - if (response.getTable_path() != null) { - location = response.getTable_path(); - } - - // Call API to get properties - if (response.getProperties() != null) { - options = response.getProperties(); - } + String location = response.getLocation(); + Map options = response.getProperties() != null + ? response.getProperties() + : new HashMap<>(); return new TableMetadata(location, options); } catch (Exception e) { - LOG.warn("Failed to get table metadata", e); + LOG.warn("Failed to get table metadata: {}", Arrays.toString(tableId), e); return new TableMetadata("/path/to/table", new HashMap<>()); } } - + /** + * Get the underlying Lance Namespace instance. + */ + public LanceNamespace getNamespace() { + if (namespace == null) { + init(); + } + return namespace; + } + /** + * Get the Buffer Allocator. + */ + public BufferAllocator getAllocator() { + return allocator; + } + + /** + * Close the adapter and release resources. + */ @Override - public void close() throws Exception { + public void close() { try { - if (namespace != null) { - namespace.close(); + if (namespace instanceof AutoCloseable) { + try { + ((AutoCloseable) namespace).close(); + } catch (Exception e) { + LOG.debug("Error invoking close() on namespace", e); + } } } catch (Exception e) { LOG.warn("Error during namespace cleanup", e); @@ -350,4 +392,25 @@ public void close() throws Exception { allocator.close(); } } -} + + /** + * Table metadata holder. + */ + public static class TableMetadata { + private final String location; + private final Map storageOptions; + + public TableMetadata(String location, Map storageOptions) { + this.location = location; + this.storageOptions = storageOptions; + } + + public String getLocation() { + return location; + } + + public Map getStorageOptions() { + return storageOptions; + } + } +} \ No newline at end of file diff --git a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java index 031b62f..ad3f541 100644 --- a/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java @@ -32,9 +32,6 @@ * - Implementation-specific parameters (root path, REST URI) * - Extra level configuration (for Spark compatibility) * - Parent prefix support (for Hive 3 compatibility) - * - * All hardcoded strings are managed through constants and enums - * to ensure maintainability and type safety. */ public class LanceNamespaceConfig { @@ -46,105 +43,7 @@ public class LanceNamespaceConfig { public static final String KEY_PARENT = "parent"; public static final String KEY_PARENT_DELIMITER = "parent_delimiter"; - /** - * Default parent path delimiter. - * Extracted as constant to eliminate hardcoding in logic. - */ - public static final String DEFAULT_PARENT_DELIMITER = "."; - - /** - * Enumeration for supported namespace implementations. - * - * This enum manages: - * - Implementation type values (no hardcoding in business logic) - * - Type validation and parsing - * - Easy extension for new implementation types - */ - public enum ImplType { - /** - * Directory-based namespace implementation (local file system). - * Value: "dir" - */ - DIRECTORY("dir"), - - /** - * REST-based namespace implementation (remote server). - * Value: "rest" - */ - REST("rest"); - - private final String typeValue; - - /** - * Constructor for implementation type. - * - * @param typeValue the string representation of this implementation type - */ - ImplType(String typeValue) { - this.typeValue = typeValue; - } - - /** - * Get the string representation of this implementation type. - * - * @return the type value string - */ - public String getTypeValue() { - return typeValue; - } - - /** - * Get ImplType from string value. - * - * @param value the string value to parse - * @return Optional containing the ImplType, or empty if not found - */ - public static Optional fromValue(String value) { - if (value == null) { - return Optional.empty(); - } - for (ImplType type : ImplType.values()) { - if (type.typeValue.equals(value)) { - return Optional.of(type); - } - } - return Optional.empty(); - } - - /** - * Get ImplType from string value, or throw exception if not found. - * - * @param value the string value to parse - * @return the ImplType for this value - * @throws IllegalArgumentException if the value is not recognized - */ - public static ImplType fromValueOrThrow(String value) { - return fromValue(value) - .orElseThrow(() -> new IllegalArgumentException( - "Unknown implementation type: " + value + - ". Supported types: " + getAllValues() - )); - } - - /** - * Get comma-separated list of all supported type values. - * - * @return comma-separated values of all implementation types - */ - public static String getAllValues() { - StringBuilder sb = new StringBuilder(); - for (ImplType type : ImplType.values()) { - if (sb.length() > 0) { - sb.append(", "); - } - sb.append(type.typeValue); - } - return sb.toString(); - } - } - private final String impl; - private final ImplType implType; private final Map properties; private final Optional extraLevel; private final Optional parent; @@ -152,9 +51,6 @@ public static String getAllValues() { /** * Create configuration from properties map. - * - * @param properties the configuration properties map - * @return a new LanceNamespaceConfig instance */ public static LanceNamespaceConfig from(Map properties) { return new LanceNamespaceConfig(properties); @@ -162,18 +58,13 @@ public static LanceNamespaceConfig from(Map properties) { /** * Create builder for configuration. - * - * @return a new Builder instance */ public static Builder builder() { return new Builder(); } /** - * Private constructor for configuration. - * - * @param properties the configuration properties map - * @throws IllegalArgumentException if required configuration is missing + * Private constructor. */ private LanceNamespaceConfig(Map properties) { this.properties = new HashMap<>(Objects.requireNonNull(properties, "Properties cannot be null")); @@ -181,13 +72,9 @@ private LanceNamespaceConfig(Map properties) { // Extract required impl this.impl = properties.get(KEY_IMPL); if (this.impl == null || this.impl.isEmpty()) { - throw new IllegalArgumentException( - "Missing required configuration: " + KEY_IMPL); + throw new IllegalArgumentException("Missing required configuration: " + KEY_IMPL); } - // Validate and extract implementation type (eliminates hardcoding) - this.implType = ImplType.fromValueOrThrow(this.impl); - // Extract optional extra level String extraLevelValue = properties.get(KEY_EXTRA_LEVEL); this.extraLevel = extraLevelValue != null && !extraLevelValue.isEmpty() ? @@ -198,23 +85,12 @@ private LanceNamespaceConfig(Map properties) { this.parent = parentValue != null && !parentValue.isEmpty() ? Optional.of(parentValue) : Optional.empty(); - // Extract parent delimiter with default (eliminates hardcoded default) - this.parentDelimiter = properties.getOrDefault(KEY_PARENT_DELIMITER, DEFAULT_PARENT_DELIMITER); + // Extract parent delimiter + this.parentDelimiter = properties.getOrDefault(KEY_PARENT_DELIMITER, "."); } /** - * Get namespace implementation type as enum. - * - * @return the ImplType enum value - */ - public ImplType getImplType() { - return implType; - } - - /** - * Get namespace implementation type as string. - * - * @return the implementation type string value + * Get namespace implementation type. */ public String getImpl() { return impl; @@ -222,8 +98,6 @@ public String getImpl() { /** * Get all configuration properties. - * - * @return unmodifiable view of the properties map */ public Map getProperties() { return Collections.unmodifiableMap(properties); @@ -231,8 +105,6 @@ public Map getProperties() { /** * Get root path for directory namespace implementation. - * - * @return Optional containing the root path if present */ public Optional getRoot() { return Optional.ofNullable(properties.get(KEY_ROOT)); @@ -240,8 +112,6 @@ public Optional getRoot() { /** * Get URI for REST namespace implementation. - * - * @return Optional containing the URI if present */ public Optional getUri() { return Optional.ofNullable(properties.get(KEY_URI)); @@ -249,8 +119,6 @@ public Optional getUri() { /** * Get extra level configuration (for Spark compatibility). - * - * @return Optional containing the extra level if present */ public Optional getExtraLevel() { return extraLevel; @@ -258,8 +126,6 @@ public Optional getExtraLevel() { /** * Get parent prefix configuration (for Hive 3 compatibility). - * - * @return Optional containing the parent prefix if present */ public Optional getParent() { return parent; @@ -267,8 +133,6 @@ public Optional getParent() { /** * Get parent delimiter. - * - * @return the delimiter string (default: ".") */ public String getParentDelimiter() { return parentDelimiter; @@ -276,37 +140,27 @@ public String getParentDelimiter() { /** * Get parent prefix as array. - * - * @return Optional containing the parent prefix split by delimiter */ public Optional getParentArray() { return parent.map(p -> p.split(java.util.regex.Pattern.quote(parentDelimiter))); } /** - * Check if directory namespace implementation is configured. - * Uses enum comparison instead of hardcoded string comparison. - * - * @return true if this is a directory namespace implementation + * Check if directory namespace implementation. */ public boolean isDirectoryNamespace() { - return implType == ImplType.DIRECTORY; + return "dir".equals(impl); } /** - * Check if REST namespace implementation is configured. - * Uses enum comparison instead of hardcoded string comparison. - * - * @return true if this is a REST namespace implementation + * Check if REST namespace implementation. */ public boolean isRestNamespace() { - return implType == ImplType.REST; + return "rest".equals(impl); } /** * Check if extra level should be automatically configured. - * - * @return true if auto-configuration of extra level is needed */ public boolean shouldAutoConfigureExtraLevel() { return !extraLevel.isPresent() && isDirectoryNamespace(); @@ -318,125 +172,56 @@ public String toString() { "impl='" + impl + '\'' + ", extraLevel=" + extraLevel + ", parent=" + parent + - ", parentDelimiter='" + parentDelimiter + '\'' + ", properties=" + properties + '}'; } /** * Builder for LanceNamespaceConfig. - * - * Provides a fluent interface for building configuration instances. */ public static class Builder { private final Map properties = new HashMap<>(); - /** - * Set the namespace implementation type. - * - * @param impl the implementation type ("dir" or "rest") - * @return this builder instance - */ public Builder impl(String impl) { properties.put(KEY_IMPL, impl); return this; } - /** - * Set the namespace implementation type using enum. - * - * @param implType the ImplType enum value - * @return this builder instance - */ - public Builder impl(ImplType implType) { - properties.put(KEY_IMPL, implType.getTypeValue()); - return this; - } - - /** - * Set the root path for directory implementation. - * - * @param root the root warehouse path - * @return this builder instance - */ public Builder root(String root) { properties.put(KEY_ROOT, root); return this; } - /** - * Set the URI for REST implementation. - * - * @param uri the REST server URI - * @return this builder instance - */ public Builder uri(String uri) { properties.put(KEY_URI, uri); return this; } - /** - * Set the extra level configuration. - * - * @param extraLevel the extra level value - * @return this builder instance - */ public Builder extraLevel(String extraLevel) { properties.put(KEY_EXTRA_LEVEL, extraLevel); return this; } - /** - * Set the parent prefix. - * - * @param parent the parent prefix value - * @return this builder instance - */ public Builder parent(String parent) { properties.put(KEY_PARENT, parent); return this; } - /** - * Set the parent delimiter. - * - * @param delimiter the delimiter string (default: ".") - * @return this builder instance - */ public Builder parentDelimiter(String delimiter) { properties.put(KEY_PARENT_DELIMITER, delimiter); return this; } - /** - * Set a custom property. - * - * @param key the property key - * @param value the property value - * @return this builder instance - */ public Builder property(String key, String value) { properties.put(key, value); return this; } - /** - * Add all properties from a map. - * - * @param props the properties map to add - * @return this builder instance - */ public Builder properties(Map props) { properties.putAll(props); return this; } - /** - * Build the configuration instance. - * - * @return a new LanceNamespaceConfig instance - * @throws IllegalArgumentException if required properties are missing or invalid - */ public LanceNamespaceConfig build() { return new LanceNamespaceConfig(properties); } From 6bc9872349eece9db99e4a3c3856a6cb30dbcca1 Mon Sep 17 00:00:00 2001 From: kaori-seasons Date: Fri, 27 Feb 2026 11:15:56 +0800 Subject: [PATCH 3/3] refactor(test): remove MockLanceNamespace, use real DirectoryNamespace backend - Remove MockLanceNamespace as lance-core natively supports DirectoryNamespace and RestNamespace - Update LanceNamespaceAdapterITCase to use real DirectoryNamespace backend for testing - Add nested RestNamespaceTests enabled via LANCE_REST_URI environment variable - DirectoryNamespace is used for storage tests, RestNamespace for API tests when available --- .../LanceNamespaceAdapterITCase.java | 129 ++++++- .../catalog/namespace/MockLanceNamespace.java | 327 ------------------ 2 files changed, 119 insertions(+), 337 deletions(-) delete mode 100644 src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java diff --git a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java index 2bf08d5..d6526d0 100644 --- a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java +++ b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceAdapterITCase.java @@ -21,7 +21,9 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; import org.junit.jupiter.api.io.TempDir; import java.nio.file.Path; @@ -35,14 +37,22 @@ /** * Lance Namespace Adapter Integration Test. * - * This test class covers all CRUD operations of LanceNamespaceAdapter for Table API, - * including complete lifecycle management of table creation, query, update and deletion. + *

This test class uses the real Lance Namespace backends (DirectoryNamespace and RestNamespace) + * as supported by lance-core. No mocking is used - tests run against actual namespace implementations. * - * Test scope: - * - Namespace management: create, list, check, delete - * - Table management: create, query, check, delete - * - Metadata operations: get namespace and table metadata - * - Error handling: duplicate creation, non-existing resources and other exception scenarios + *

Test backends: + *

    + *
  • DirectoryNamespace: Default backend for local storage, used in all tests
    • + *
  • RestNamespace: REST API backend, enabled when LANCE_REST_URI environment variable is set
    • + *
+ * + *

Test scope: + *

    + *
  • Namespace management: create, list, check, delete
    • + *
  • Table management: create, query, check, delete
    • + *
  • Metadata operations: get namespace and table metadata
    • + *
  • Error handling: duplicate creation, non-existing resources and other exception scenarios
    • + *
*/ @DisplayName("Lance Namespace Adapter Integration Test") class LanceNamespaceAdapterITCase { @@ -351,7 +361,7 @@ void testGetTableMetadata() { adapter.createEmptyTable(tableLocation, tableProperties, namespaceName, tableName); // Execute - AbstractLanceNamespaceAdapter.TableMetadata metadata = + LanceNamespaceAdapter.TableMetadata metadata = adapter.getTableMetadata(namespaceName, tableName); // Verify @@ -442,9 +452,9 @@ void testCompleteTableCrudLifecycle() { assertThat(tables).contains(tableName); // 4. Read - get table metadata - AbstractLanceNamespaceAdapter.TableMetadata metadata = + LanceNamespaceAdapter.TableMetadata tableMetadata = adapter.getTableMetadata(namespaceName, tableName); - assertThat(metadata.getLocation()).contains(tableName); + assertThat(tableMetadata.getLocation()).contains(tableName); // 5. Delete - drop table adapter.dropTable(namespaceName, tableName); @@ -613,4 +623,103 @@ void testAdapterCloseAndResourceCleanup() throws Exception { newAdapter.close(); } } + + // ==================== REST Namespace Tests ==================== + + /** + * REST Namespace backend tests. + * + *

These tests are enabled when the LANCE_REST_URI environment variable is set. + * They verify that the adapter works correctly with REST-based namespace implementations. + * + *

To run these tests, set the environment variable: + * 

+     * export LANCE_REST_URI=http://localhost:8080
+     *
+ */ + @Nested + @DisplayName("REST Namespace Backend Tests") + @EnabledIfEnvironmentVariable(named = "LANCE_REST_URI", matches = ".+") + class RestNamespaceTests { + + private LanceNamespaceAdapter restAdapter; + + @BeforeEach + void setUp() { + String restUri = System.getenv("LANCE_REST_URI"); + + Map properties = new HashMap<>(); + properties.put(LanceNamespaceConfig.KEY_IMPL, "rest"); + properties.put(LanceNamespaceConfig.KEY_URI, restUri); + + restAdapter = LanceNamespaceAdapter.create(properties); + restAdapter.init(); + } + + @AfterEach + void tearDown() throws Exception { + if (restAdapter != null) { + restAdapter.close(); + } + } + + @Test + @DisplayName("Test REST namespace creation and listing") + void testRestNamespaceOperations() { + String testNamespace = "rest_test_db_" + System.currentTimeMillis(); + + try { + // Create namespace + restAdapter.createNamespace(new HashMap<>(), testNamespace); + assertThat(restAdapter.namespaceExists(testNamespace)).isTrue(); + + // List namespaces + List namespaces = restAdapter.listNamespaces(); + assertThat(namespaces).contains(testNamespace); + } finally { + // Cleanup + try { + restAdapter.dropNamespace(true, testNamespace); + } catch (Exception ignored) { + // Ignore cleanup errors + } + } + } + + @Test + @DisplayName("Test REST table operations") + void testRestTableOperations() { + String testNamespace = "rest_table_db_" + System.currentTimeMillis(); + String testTable = "rest_test_table"; + + try { + // Create namespace first + restAdapter.createNamespace(new HashMap<>(), testNamespace); + + // Create table + restAdapter.createEmptyTable(null, new HashMap<>(), testNamespace, testTable); + assertThat(restAdapter.tableExists(testNamespace, testTable)).isTrue(); + + // List tables + List tables = restAdapter.listTables(testNamespace); + assertThat(tables).contains(testTable); + + // Get table metadata + LanceNamespaceAdapter.TableMetadata metadata = + restAdapter.getTableMetadata(testNamespace, testTable); + assertThat(metadata).isNotNull(); + + // Drop table + restAdapter.dropTable(testNamespace, testTable); + assertThat(restAdapter.tableExists(testNamespace, testTable)).isFalse(); + } finally { + // Cleanup + try { + restAdapter.dropNamespace(true, testNamespace); + } catch (Exception ignored) { + // Ignore cleanup errors + } + } + } + } } diff --git a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java b/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java deleted file mode 100644 index 9326bdc..0000000 --- a/src/test/java/org/apache/flink/connector/lance/catalog/namespace/MockLanceNamespace.java +++ /dev/null @@ -1,327 +0,0 @@ -/* - * 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.flink.connector.lance.catalog.namespace; - -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -/** - * Mock implementation of Lance Namespace for testing purposes. - * - * This mock allows tests to run without requiring the actual Lance Namespace - * library to be available. It simulates the basic behavior of the real API. - */ -public class MockLanceNamespace { - - /** - * Storage for namespaces and their properties. - * Map structure: namespace_path -> properties - */ - private final Map> namespaces = new HashMap<>(); - - /** - * Storage for tables and their properties. - * Map structure: table_path -> properties - */ - private final Map> tables = new HashMap<>(); - - /** - * Check if namespace exists locally. - */ - private boolean hasNamespace(String id) { - return namespaces.containsKey(id); - } - - /** - * Check if table exists locally. - */ - private boolean hasTable(String id) { - return tables.containsKey(id); - } - - /** - * Create a mock namespace instance. - */ - public static MockLanceNamespace connect(String impl, String location, Object allocator) { - return new MockLanceNamespace(); - } - - /** - * Mock: Create a namespace. - */ - public void createNamespace(Object request) { - String id = extractId(request); - if (hasNamespace(id)) { - throw new RuntimeException("Namespace already exists: " + id); - } - Map props = extractProperties(request); - namespaces.put(id, props != null ? new HashMap<>(props) : new HashMap<>()); - } - - /** - * Mock: List namespaces. - */ - public Object listNamespaces(Object request) { - Set result = new HashSet<>(namespaces.keySet()); - return createResponse(result); - } - - /** - * Mock: Check if namespace exists. - */ - public Object namespaceExists(Object request) { - String id = extractId(request); - if (!hasNamespace(id)) { - throw new RuntimeException("Namespace does not exist: " + id); - } - return null; - } - - /** - * Mock: Drop a namespace. - */ - public void dropNamespace(Object request) { - String id = extractId(request); - if (!hasNamespace(id)) { - throw new RuntimeException("Namespace does not exist: " + id); - } - namespaces.remove(id); - } - - /** - * Mock: Describe a namespace (get metadata). - */ - public Object describeNamespace(Object request) { - String id = extractId(request); - if (!hasNamespace(id)) { - throw new RuntimeException("Namespace does not exist: " + id); - } - Map props = namespaces.get(id); - return createDescribeResponse(props); - } - - /** - * Mock: Create an empty table. - */ - public void createEmptyTable(Object request) { - String id = extractId(request); - if (hasTable(id)) { - throw new RuntimeException("Table already exists: " + id); - } - Map props = extractProperties(request); - tables.put(id, props != null ? new HashMap<>(props) : new HashMap<>()); - } - - /** - * Mock: List tables. - */ - public Object listTables(Object request) { - Set result = new HashSet<>(tables.keySet()); - return createResponse(result); - } - - /** - * Mock: Check if table exists. - */ - public Object tableExists(Object request) { - String id = extractId(request); - if (!hasTable(id)) { - throw new RuntimeException("Table does not exist: " + id); - } - return null; - } - - /** - * Mock: Drop a table. - */ - public void dropTable(Object request) { - String id = extractId(request); - if (!hasTable(id)) { - throw new RuntimeException("Table does not exist: " + id); - } - tables.remove(id); - } - - /** - * Mock: Describe a table (get metadata). - */ - public Object describeTable(Object request) { - String id = extractId(request); - if (!hasTable(id)) { - throw new RuntimeException("Table does not exist: " + id); - } - Map props = tables.get(id); - return createTableDescribeResponse(props); - } - - /** - * Close the connection. - */ - public void close() throws Exception { - // Cleanup - namespaces.clear(); - tables.clear(); - } - - // ==================== Helper Methods ==================== - - /** - * Extract ID from request object. - */ - private String extractId(Object request) { - if (request == null) { - return ""; - } - try { - Object id = request.getClass() - .getMethod("getId") - .invoke(request); - if (id instanceof java.util.List) { - java.util.List list = (java.util.List) id; - return list.isEmpty() ? "" : list.get(0).toString(); - } - return id != null ? id.toString() : ""; - } catch (Exception e) { - return ""; - } - } - - /** - * Extract properties from request object. - */ - private Map extractProperties(Object request) { - if (request == null) { - return null; - } - try { - Object props = request.getClass() - .getMethod("getProperties") - .invoke(request); - if (props instanceof Map) { - return (Map) props; - } - return null; - } catch (Exception e) { - return null; - } - } - - /** - * Create a response object with namespaces. - */ - private Object createResponse(Set namespaces) { - try { - Class responseClass = Class.forName( - "org.lance.namespace.model.ListNamespacesResponse"); - Object response = responseClass.getDeclaredConstructor().newInstance(); - - // Try to set the namespaces using reflection - try { - responseClass.getMethod("setNamespaces", Set.class) - .invoke(response, namespaces); - } catch (Exception e) { - // If setNamespaces fails, try setting as field - java.lang.reflect.Field field = responseClass.getDeclaredField("namespaces"); - field.setAccessible(true); - field.set(response, namespaces); - } - return response; - } catch (Exception e) { - // Fallback: return a simple object with getNamespaces method - return new Object() { - public Set getNamespaces() { - return namespaces; - } - }; - } - } - - /** - * Create a describe response object. - */ - private Object createDescribeResponse(Map properties) { - try { - Class responseClass = Class.forName( - "org.lance.namespace.model.DescribeNamespaceResponse"); - Object response = responseClass.getDeclaredConstructor().newInstance(); - - try { - responseClass.getMethod("setProperties", Map.class) - .invoke(response, properties); - } catch (Exception e) { - // Fallback to field - java.lang.reflect.Field field = responseClass.getDeclaredField("properties"); - field.setAccessible(true); - field.set(response, properties); - } - return response; - } catch (Exception e) { - // Fallback - return new Object() { - public Map getProperties() { - return properties; - } - }; - } - } - - /** - * Create a table describe response object. - */ - private Object createTableDescribeResponse(Map properties) { - try { - Class responseClass = Class.forName( - "org.lance.namespace.model.DescribeTableResponse"); - Object response = responseClass.getDeclaredConstructor().newInstance(); - - try { - responseClass.getMethod("setProperties", Map.class) - .invoke(response, properties); - } catch (Exception e) { - // Fallback to field - java.lang.reflect.Field field = responseClass.getDeclaredField("properties"); - field.setAccessible(true); - field.set(response, properties); - } - - // Try to set table_path - try { - responseClass.getMethod("setTable_path", String.class) - .invoke(response, "/path/to/table"); - } catch (Exception ignored) { - // Optional field - } - - return response; - } catch (Exception e) { - // Fallback - return new Object() { - public Map getProperties() { - return properties; - } - - public String getTable_path() { - return "/path/to/table"; - } - }; - } - } -}