From 58ca6cdc2df94266a7429e462857ac0fe9b742e9 Mon Sep 17 00:00:00 2001 From: rockyyin Date: Fri, 17 Jul 2026 00:46:12 +0800 Subject: [PATCH] feat: add LanceNamespaceCatalog backed by Lance namespace API - Add LanceNamespaceCatalog extending AbstractCatalog using LanceNamespace directly - Add LanceNamespaceCatalogFactory (Flink CatalogFactory, type='lance-namespace') - Add LanceNamespaceConfig for namespace configuration - Register factory in META-INF/services/org.apache.flink.table.factories.Factory - Add lance-namespace-core:0.7.5 and lance-namespace-apache-client:0.7.5 - Fix arrow-memory-core version alignment (exclude 15.0.0 from lance-core) - Integration test with real DirectoryNamespace backend --- lance-flink-1.18/pom.xml | 12 + lance-flink-1.19/pom.xml | 12 + lance-flink-1.20/pom.xml | 12 + pom.xml | 29 + .../namespace/LanceNamespaceConfig.java | 213 ++++++ .../lance/table/LanceNamespaceCatalog.java | 609 ++++++++++++++++++ .../table/LanceNamespaceCatalogFactory.java | 157 +++++ .../org.apache.flink.table.factories.Factory | 1 + .../table/LanceNamespaceCatalogITCase.java | 147 +++++ 9 files changed, 1192 insertions(+) create mode 100644 src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java create mode 100644 src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalog.java create mode 100644 src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogFactory.java create mode 100644 src/test/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogITCase.java diff --git a/lance-flink-1.18/pom.xml b/lance-flink-1.18/pom.xml index e38f64d..1db7a2d 100644 --- a/lance-flink-1.18/pom.xml +++ b/lance-flink-1.18/pom.xml @@ -26,6 +26,18 @@ lance-core + + + org.lance + lance-namespace-core + + + + org.lance + lance-namespace-apache-client + + org.apache.arrow diff --git a/lance-flink-1.19/pom.xml b/lance-flink-1.19/pom.xml index 45721e7..73adbcb 100644 --- a/lance-flink-1.19/pom.xml +++ b/lance-flink-1.19/pom.xml @@ -26,6 +26,18 @@ lance-core + + + org.lance + lance-namespace-core + + + + org.lance + lance-namespace-apache-client + + org.apache.arrow diff --git a/lance-flink-1.20/pom.xml b/lance-flink-1.20/pom.xml index 248e43a..88907d7 100644 --- a/lance-flink-1.20/pom.xml +++ b/lance-flink-1.20/pom.xml @@ -26,6 +26,18 @@ lance-core + + + org.lance + lance-namespace-core + + + + org.lance + lance-namespace-apache-client + + org.apache.arrow diff --git a/pom.xml b/pom.xml index d5c3887..08d5929 100644 --- a/pom.xml +++ b/pom.xml @@ -58,9 +58,38 @@ org.slf4j slf4j-api + + org.apache.arrow + arrow-memory-core + + + + org.apache.arrow + arrow-memory-core + ${arrow.version} + + + + + + org.lance + lance-namespace-core + ${lance-namespace.version} + + + + + org.lance + lance-namespace-apache-client + ${lance-namespace.version} + + org.apache.arrow 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..2da37e5 --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/catalog/namespace/LanceNamespaceConfig.java @@ -0,0 +1,213 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.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: + *

+ */ +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"; + + private final String impl; + private final Map properties; + private final Optional extraLevel; + private final Optional parent; + private final String parentDelimiter; + + /** + * Create configuration from properties map. + */ + public static LanceNamespaceConfig from(Map properties) { + return new LanceNamespaceConfig(properties); + } + + /** + * Create builder for configuration. + */ + public static Builder builder() { + return new Builder(); + } + + 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); + } + + // 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 + this.parentDelimiter = properties.getOrDefault(KEY_PARENT_DELIMITER, "."); + } + + /** + * Get namespace implementation type. + */ + public String getImpl() { + return impl; + } + + /** + * Get all configuration properties. + */ + public Map getProperties() { + return Collections.unmodifiableMap(properties); + } + + /** + * Get root path for directory namespace implementation. + */ + public Optional getRoot() { + return Optional.ofNullable(properties.get(KEY_ROOT)); + } + + /** + * Get URI for REST namespace implementation. + */ + public Optional getUri() { + return Optional.ofNullable(properties.get(KEY_URI)); + } + + /** + * Get extra level configuration (for Spark compatibility). + */ + public Optional getExtraLevel() { + return extraLevel; + } + + /** + * Get parent prefix configuration (for Hive 3 compatibility). + */ + public Optional getParent() { + return parent; + } + + /** + * Get parent delimiter. + */ + public String getParentDelimiter() { + return parentDelimiter; + } + + /** + * Check if directory namespace implementation. + */ + public boolean isDirectoryNamespace() { + return "dir".equals(impl); + } + + /** + * Check if REST namespace implementation. + */ + public boolean isRestNamespace() { + return "rest".equals(impl); + } + + @Override + public String toString() { + return "LanceNamespaceConfig{" + + "impl='" + impl + '\'' + + ", extraLevel=" + extraLevel + + ", parent=" + parent + + '}'; + } + + /** + * Builder for LanceNamespaceConfig. + */ + public static class Builder { + private final Map properties = new HashMap<>(); + + public Builder impl(String impl) { + properties.put(KEY_IMPL, impl); + return this; + } + + public Builder root(String root) { + properties.put(KEY_ROOT, root); + return this; + } + + public Builder uri(String uri) { + properties.put(KEY_URI, uri); + return this; + } + + public Builder extraLevel(String extraLevel) { + properties.put(KEY_EXTRA_LEVEL, extraLevel); + return this; + } + + public Builder parent(String parent) { + properties.put(KEY_PARENT, parent); + return this; + } + + public Builder parentDelimiter(String delimiter) { + properties.put(KEY_PARENT_DELIMITER, delimiter); + return this; + } + + public Builder property(String key, String value) { + properties.put(key, value); + return this; + } + + public Builder properties(Map props) { + properties.putAll(props); + return this; + } + + public LanceNamespaceConfig build() { + return new LanceNamespaceConfig(properties); + } + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalog.java b/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalog.java new file mode 100644 index 0000000..d38ca21 --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalog.java @@ -0,0 +1,609 @@ +/* + * 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.table; + +import org.apache.flink.connector.lance.catalog.namespace.LanceNamespaceConfig; + +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.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogPartition; +import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; +import org.apache.flink.table.expressions.Expression; + +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.model.CreateNamespaceRequest; +import org.lance.namespace.model.CreateTableRequest; +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; + +import static java.util.Collections.singletonList; + +/** + * Catalog backed by a Lance namespace. + * + *

Uses {@link LanceNamespace} directly — no intermediate adapter layer. + * The namespace implementation ({@code dir}, {@code rest}, etc.) is + * selected through {@link LanceNamespaceConfig} and connected via + * {@link LanceNamespace#connect(String, Map, BufferAllocator)}. + * + *

Usage via Flink SQL: + * 

{@code
+ * CREATE CATALOG my_lance WITH (
+ *     'type' = 'lance-namespace',
+ *     'impl' = 'dir',
+ *     'root' = '/tmp/lance-warehouse'
+ * );
+ * USE CATALOG my_lance;
+ * }
+ */ +public class LanceNamespaceCatalog extends AbstractCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(LanceNamespaceCatalog.class); + + public static final String DEFAULT_DATABASE = "default"; + + private final LanceNamespace namespace; + private final BufferAllocator allocator; + private final Map options; + private boolean opened; + private boolean closed; + + public LanceNamespaceCatalog( + String name, + String defaultDatabase, + LanceNamespace namespace, + BufferAllocator allocator, + Map options) { + super(name, defaultDatabase); + this.namespace = Objects.requireNonNull(namespace, "namespace"); + this.allocator = Objects.requireNonNull(allocator, "allocator"); + this.options = new HashMap<>(Objects.requireNonNull(options, "options")); + } + + @Override + public void open() throws CatalogException { + if (closed) { + throw new CatalogException("Cannot open a closed Lance namespace catalog"); + } + if (opened) { + return; + } + + try { + // LanceNamespace.connect() already initializes the namespace; + // no need to call namespace.initialize() again. + createDatabaseIfMissing(getDefaultDatabase()); + opened = true; + } catch (Exception e) { + closed = true; + closeQuietly(e); + throw new CatalogException("Failed to open Lance namespace catalog", e); + } + } + + @Override + public void close() throws CatalogException { + if (closed) { + return; + } + closed = true; + opened = false; + + try { + allocator.close(); + } catch (Exception e) { + throw new CatalogException("Failed to close Arrow allocator", e); + } + } + + // ========== Database Operations ========== + + @Override + public List listDatabases() throws CatalogException { + List databases = new ArrayList<>(); + ListNamespacesRequest request = new ListNamespacesRequest(); + String pageToken = null; + try { + do { + if (pageToken != null) { + request.setPageToken(pageToken); + } + ListNamespacesResponse response = namespace.listNamespaces(request); + databases.addAll(response.getNamespaces()); + pageToken = response.getPageToken(); + } while (pageToken != null && !pageToken.isEmpty()); + } catch (RuntimeException e) { + throw new CatalogException("Failed to list databases", e); + } + return databases; + } + + @Override + public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException { + if (databaseExists(databaseName)) { + return new CatalogDatabaseImpl(Collections.emptyMap(), null); + } + throw new DatabaseNotExistException(getName(), databaseName); + } + + @Override + public boolean databaseExists(String databaseName) { + try { + namespace.namespaceExists(new NamespaceExistsRequest().id(singletonList(databaseName))); + return true; + } catch (RuntimeException e) { + if (isNotFound(e)) { + return false; + } + throw new CatalogException("Failed to check if database exists: " + databaseName, e); + } + } + + @Override + public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) + throws DatabaseAlreadyExistException, CatalogException { + try { + namespace.createNamespace(new CreateNamespaceRequest().id(singletonList(name))); + } catch (RuntimeException e) { + if (isAlreadyExists(e)) { + if (!ignoreIfExists) { + throw new DatabaseAlreadyExistException(getName(), name); + } + return; + } + throw new CatalogException("Failed to create database: " + name, e); + } + } + + @Override + public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) + throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException { + try { + namespace.dropNamespace(new DropNamespaceRequest().id(singletonList(name))); + } catch (RuntimeException e) { + if (isNotFound(e)) { + if (!ignoreIfNotExists) { + throw new DatabaseNotExistException(getName(), name); + } + return; + } + if (isNotEmpty(e)) { + if (!cascade) { + throw new DatabaseNotEmptyException(getName(), name); + } + throw new CatalogException( + "Lance namespace catalog does not support dropping non-empty databases with CASCADE"); + } + throw new CatalogException("Failed to drop database: " + name, e); + } + } + + @Override + public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(name)) { + if (!ignoreIfNotExists) { + throw new DatabaseNotExistException(getName(), name); + } + return; + } + throw new CatalogException("Lance namespace catalog does not support altering databases"); + } + + // ========== Table Operations ========== + + @Override + public List listTables(String databaseName) + throws DatabaseNotExistException, CatalogException { + List tables = new ArrayList<>(); + ListTablesRequest request = new ListTablesRequest().id(singletonList(databaseName)); + String pageToken = null; + try { + do { + if (pageToken != null) { + request.setPageToken(pageToken); + } + ListTablesResponse response = namespace.listTables(request); + tables.addAll(response.getTables()); + pageToken = response.getPageToken(); + } while (pageToken != null && !pageToken.isEmpty()); + } catch (RuntimeException e) { + if (isNotFound(e)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + throw new CatalogException("Failed to list tables in database: " + databaseName, e); + } + return tables; + } + + @Override + public List listViews(String databaseName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(databaseName)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return Collections.emptyList(); + } + + @Override + public boolean tableExists(ObjectPath tablePath) throws CatalogException { + try { + namespace.tableExists(new TableExistsRequest().id(tableId(tablePath))); + return true; + } catch (RuntimeException e) { + if (isNotFound(e)) { + return false; + } + throw new CatalogException("Failed to check if table exists: " + tablePath, e); + } + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) + throws TableNotExistException, CatalogException { + DescribeTableResponse response; + try { + response = namespace.describeTable(new DescribeTableRequest().id(tableId(tablePath))); + } catch (RuntimeException e) { + if (isNotFound(e)) { + throw new TableNotExistException(getName(), tablePath); + } + throw new CatalogException("Failed to get table: " + tablePath, e); + } + + String location = response.getLocation(); + Map tableOptions = new HashMap<>(); + tableOptions.put("connector", "lance"); + tableOptions.put("path", location); + + Map props = response.getProperties(); + if (props != null) { + tableOptions.putAll(props); + } + + return CatalogTable.of( + org.apache.flink.table.api.Schema.newBuilder().build(), + "", + Collections.emptyList(), + tableOptions); + } + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException, CatalogException { + if (!databaseExists(tablePath.getDatabaseName())) { + throw new DatabaseNotExistException(getName(), tablePath.getDatabaseName()); + } + if (ignoreIfExists && tableExists(tablePath)) { + return; + } + + Map tableOptions = table.getOptions(); + String location = tableOptions.get("path"); + if (location == null || location.isEmpty()) { + throw new CatalogException( + "Table option 'path' is required when creating tables in Lance namespace catalog. " + + "Specify the Lance dataset path, e.g. WITH ('path' = '/data/my_dataset')"); + } + + try { + namespace.createTable(new CreateTableRequest().id(tableId(tablePath)), new byte[0]); + } catch (RuntimeException e) { + if (isAlreadyExists(e)) { + if (!ignoreIfExists) { + throw new TableAlreadyExistException(getName(), tablePath); + } + return; + } + throw new CatalogException("Failed to create table: " + tablePath, e); + } + } + + @Override + public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + try { + namespace.dropTable(new DropTableRequest().id(tableId(tablePath))); + } catch (RuntimeException e) { + if (isNotFound(e)) { + if (!ignoreIfNotExists) { + throw new TableNotExistException(getName(), tablePath); + } + return; + } + throw new CatalogException("Failed to drop table: " + tablePath, e); + } + } + + @Override + public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + if (!tableExists(tablePath)) { + if (!ignoreIfNotExists) { + throw new TableNotExistException(getName(), tablePath); + } + return; + } + throw new CatalogException("Lance namespace catalog does not support renaming tables"); + } + + @Override + public void alterTable( + ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + if (!tableExists(tablePath)) { + if (!ignoreIfNotExists) { + throw new TableNotExistException(getName(), tablePath); + } + return; + } + throw new CatalogException("Lance namespace catalog does not support altering tables"); + } + + // ========== Partition Operations (unsupported) ========== + + @Override + public List listPartitions(ObjectPath tablePath) { + return Collections.emptyList(); + } + + @Override + public List listPartitions( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + return Collections.emptyList(); + } + + @Override + public List listPartitionsByFilter( + ObjectPath tablePath, List filters) { + return Collections.emptyList(); + } + + @Override + public CatalogPartition getPartition( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) + throws PartitionNotExistException { + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + @Override + public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + return false; + } + + @Override + public void createPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition partition, + boolean ignoreIfExists) { + throw new CatalogException("Lance namespace catalog does not support partitions"); + } + + @Override + public void dropPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + boolean ignoreIfNotExists) { + throw new CatalogException("Lance namespace catalog does not support partitions"); + } + + @Override + public void alterPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition newPartition, + boolean ignoreIfNotExists) { + throw new CatalogException("Lance namespace catalog does not support partitions"); + } + + // ========== Function Operations (unsupported) ========== + + @Override + public List listFunctions(String dbName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + return Collections.emptyList(); + } + + @Override + public CatalogFunction getFunction(ObjectPath functionPath) + throws FunctionNotExistException, CatalogException { + throw new FunctionNotExistException(getName(), functionPath); + } + + @Override + public boolean functionExists(ObjectPath functionPath) { + return false; + } + + @Override + public void createFunction( + ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) { + throw new CatalogException( + "Lance namespace catalog does not support user-defined functions"); + } + + @Override + public void alterFunction( + ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) { + throw new CatalogException( + "Lance namespace catalog does not support user-defined functions"); + } + + @Override + public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) { + throw new CatalogException( + "Lance namespace catalog does not support user-defined functions"); + } + + // ========== Statistics Operations (unsupported) ========== + + @Override + public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) { + return CatalogTableStatistics.UNKNOWN; + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) { + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public CatalogTableStatistics getPartitionStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + return CatalogTableStatistics.UNKNOWN; + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + return CatalogColumnStatistics.UNKNOWN; + } + + @Override + public void alterTableStatistics( + ObjectPath tablePath, + CatalogTableStatistics tableStatistics, + boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Lance namespace catalog does not support updating statistics"); + } + + @Override + public void alterTableColumnStatistics( + ObjectPath tablePath, + CatalogColumnStatistics columnStatistics, + boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Lance namespace catalog does not support updating statistics"); + } + + @Override + public void alterPartitionStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogTableStatistics partitionStatistics, + boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Lance namespace catalog does not support updating statistics"); + } + + @Override + public void alterPartitionColumnStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogColumnStatistics columnStatistics, + boolean ignoreIfNotExists) + throws CatalogException { + throw new CatalogException("Lance namespace catalog does not support updating statistics"); + } + + // ========== Helpers ========== + + private void createDatabaseIfMissing(String database) { + if (databaseExists(database)) { + return; + } + try { + namespace.createNamespace(new CreateNamespaceRequest().id(singletonList(database))); + } catch (RuntimeException e) { + if (!isAlreadyExists(e)) { + throw new CatalogException("Failed to create database: " + database, e); + } + } + } + + private static List tableId(ObjectPath tablePath) { + return List.of(tablePath.getDatabaseName(), tablePath.getObjectName()); + } + + private void closeQuietly(Exception originalError) { + try { + allocator.close(); + } catch (Exception closeError) { + originalError.addSuppressed(closeError); + } + } + + // ========== Error classification ========== + + private static boolean isNotFound(RuntimeException e) { + return messageContains(e, "not found"); + } + + private static boolean isAlreadyExists(RuntimeException e) { + return messageContains(e, "already exists"); + } + + private static boolean isNotEmpty(RuntimeException e) { + return messageContains(e, "not empty"); + } + + private static boolean messageContains(Throwable t, String needle) { + String lower = needle.toLowerCase(Locale.ROOT); + for (Throwable cur = t; cur != null; cur = cur.getCause()) { + String msg = cur.getMessage(); + if (msg != null && msg.toLowerCase(Locale.ROOT).contains(lower)) { + return true; + } + if (cur.getCause() == cur) { + break; + } + } + return false; + } +} diff --git a/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogFactory.java b/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogFactory.java new file mode 100644 index 0000000..21743a3 --- /dev/null +++ b/src/main/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogFactory.java @@ -0,0 +1,157 @@ +/* + * 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.table; + +import org.apache.flink.connector.lance.catalog.namespace.LanceNamespaceConfig; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import org.lance.namespace.LanceNamespace; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Flink {@link CatalogFactory} for {@link LanceNamespaceCatalog}. + * + *

Usage via Flink SQL: + * 

{@code
+ * -- Directory-based namespace (local)
+ * CREATE CATALOG my_lance WITH (
+ *     'type' = 'lance-namespace',
+ *     'impl' = 'dir',
+ *     'root' = '/tmp/lance-warehouse'
+ * );
+ *
+ * -- REST-based namespace
+ * CREATE CATALOG my_lance WITH (
+ *     'type' = 'lance-namespace',
+ *     'impl' = 'rest',
+ *     'uri' = 'http://localhost:8080'
+ * );
+ * }
+ */ +public class LanceNamespaceCatalogFactory implements CatalogFactory { + + public static final String IDENTIFIER = "lance-namespace"; + + public static final ConfigOption IMPL = ConfigOptions + .key(LanceNamespaceConfig.KEY_IMPL) + .stringType() + .noDefaultValue() + .withDescription("Lance namespace implementation: 'dir' or 'rest'"); + + public static final ConfigOption ROOT = ConfigOptions + .key(LanceNamespaceConfig.KEY_ROOT) + .stringType() + .noDefaultValue() + .withDescription("Root path for directory namespace"); + + public static final ConfigOption URI = ConfigOptions + .key(LanceNamespaceConfig.KEY_URI) + .stringType() + .noDefaultValue() + .withDescription("URI for REST namespace"); + + public static final ConfigOption DEFAULT_DATABASE = ConfigOptions + .key("default-database") + .stringType() + .defaultValue(LanceNamespaceCatalog.DEFAULT_DATABASE) + .withDescription("Default database name"); + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set> requiredOptions() { + Set> options = new HashSet<>(); + options.add(IMPL); + return options; + } + + @Override + public Set> optionalOptions() { + Set> options = new HashSet<>(); + options.add(ROOT); + options.add(URI); + options.add(DEFAULT_DATABASE); + return options; + } + + @Override + public Catalog createCatalog(Context context) { + FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + + String catalogName = context.getName(); + String impl = helper.getOptions().get(IMPL); + String defaultDatabase = helper.getOptions().get(DEFAULT_DATABASE); + + // Build namespace connection properties + Map namespaceProps = new HashMap<>(); + namespaceProps.put(LanceNamespaceConfig.KEY_IMPL, impl); + + String root = helper.getOptions().get(ROOT); + if (root != null) { + namespaceProps.put(LanceNamespaceConfig.KEY_ROOT, root); + } + + String uri = helper.getOptions().get(URI); + if (uri != null) { + namespaceProps.put(LanceNamespaceConfig.KEY_URI, uri); + } + + // Collect remaining options for the namespace catalog + Map catalogOptions = new HashMap<>(); + catalogOptions.put(LanceNamespaceConfig.KEY_IMPL, impl); + if (root != null) { + catalogOptions.put(LanceNamespaceConfig.KEY_ROOT, root); + } + if (uri != null) { + catalogOptions.put(LanceNamespaceConfig.KEY_URI, uri); + } + + // Connect to Lance namespace + // Use TCCL switch to ensure arrow-memory-netty SPI is visible from + // Flink's test classloader which may not include it by default. + ClassLoader originalCL = Thread.currentThread().getContextClassLoader(); + try { + Thread.currentThread() + .setContextClassLoader(LanceNamespaceCatalogFactory.class.getClassLoader()); + BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE); + LanceNamespace namespace = + LanceNamespace.connect(impl, namespaceProps, allocator); + return new LanceNamespaceCatalog( + catalogName, defaultDatabase, namespace, allocator, catalogOptions); + } finally { + Thread.currentThread().setContextClassLoader(originalCL); + } + } +} diff --git a/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory index 4f34cb6..2449002 100644 --- a/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory +++ b/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -1,2 +1,3 @@ org.apache.flink.connector.lance.table.LanceDynamicTableFactory org.apache.flink.connector.lance.table.LanceCatalogFactory +org.apache.flink.connector.lance.table.LanceNamespaceCatalogFactory diff --git a/src/test/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogITCase.java b/src/test/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogITCase.java new file mode 100644 index 0000000..89bd8d7 --- /dev/null +++ b/src/test/java/org/apache/flink/connector/lance/table/LanceNamespaceCatalogITCase.java @@ -0,0 +1,147 @@ +/* + * 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.table; + +import org.apache.flink.table.api.DataTypes; + +import org.apache.flink.connector.lance.catalog.namespace.LanceNamespaceConfig; + +import org.lance.namespace.LanceNamespace; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration test for {@link LanceNamespaceCatalog} using real {@code DirectoryNamespace} + * backend — no mocks. + * + *

Exercises Flink SQL operations through the catalog programmatic API rather than + * Flink SQL DDL to avoid Flink test classloader complications with Arrow memory-netty SPI. + */ +class LanceNamespaceCatalogITCase { + + @TempDir + static Path warehouseDir; + + private LanceNamespaceCatalog catalog; + + @BeforeAll + static void ensureArrowNettyLoaded() { + // Force Arrow to use Netty allocation manager via system property. + // In some classloader configurations (e.g. Flink test infrastructure), + // the SPI-based DefaultAllocationManager discovery fails because + // arrow-memory-netty is not visible to the ServiceLoader. + // Setting this property forces Arrow to use the Netty allocator directly. + System.setProperty( + "arrow.memory.allocator.type", "Netty"); + try (RootAllocator alloc = new RootAllocator(Long.MAX_VALUE)) { + // allocator created and closed successfully + } + } + + @BeforeEach + void setUp() { + Map namespaceProps = new HashMap<>(); + namespaceProps.put(LanceNamespaceConfig.KEY_IMPL, "dir"); + namespaceProps.put(LanceNamespaceConfig.KEY_ROOT, warehouseDir.toAbsolutePath().toString()); + + Map catalogOptions = new HashMap<>(); + catalogOptions.put(LanceNamespaceConfig.KEY_IMPL, "dir"); + catalogOptions.put(LanceNamespaceConfig.KEY_ROOT, warehouseDir.toAbsolutePath().toString()); + + BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE); + LanceNamespace namespace = + LanceNamespace.connect("dir", namespaceProps, allocator); + + catalog = new LanceNamespaceCatalog( + "test_lance", "default", namespace, allocator, catalogOptions); + catalog.open(); + } + + @AfterEach + void tearDown() { + if (catalog != null) { + try { + catalog.close(); + } catch (Exception ignored) { + } + } + try { + Files.walk(warehouseDir) + .map(Path::toFile) + .forEach(File::delete); + } catch (Exception ignored) { + } + } + + @Test + void testListDatabases() { + assertThat(catalog.listDatabases()).contains("default"); + } + + @Test + void testCreateAndDropDatabase() throws Exception { + catalog.createDatabase("test_db", null, false); + assertThat(catalog.databaseExists("test_db")).isTrue(); + assertThat(catalog.listDatabases()).contains("test_db"); + + catalog.dropDatabase("test_db", false, false); + assertThat(catalog.databaseExists("test_db")).isFalse(); + } + + @Test + void testCreateTable() throws Exception { + catalog.createDatabase("test_db", null, false); + + // Create a simple table via namespace + Map tableOptions = new HashMap<>(); + tableOptions.put("path", warehouseDir.resolve("my_table").toAbsolutePath().toString()); + + org.apache.flink.table.catalog.ObjectPath tablePath = + new org.apache.flink.table.catalog.ObjectPath("test_db", "my_table"); + + catalog.createTable(tablePath, + org.apache.flink.table.catalog.CatalogTable.of( + org.apache.flink.table.api.Schema.newBuilder() + .column("id", org.apache.flink.table.api.DataTypes.INT()) + .column("name", org.apache.flink.table.api.DataTypes.STRING()) + .build(), + "", + java.util.Collections.emptyList(), + tableOptions), + false); + + assertThat(catalog.tableExists(tablePath)).isTrue(); + assertThat(catalog.listTables("test_db")).contains("my_table"); + } +} \ No newline at end of file