feat: add catalog namespace support and refactor adapter implementation

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<String> listNamespaces();
+
+    /**
+     * List namespaces under a parent namespace.
+     */
+    List<String> listNamespaces(String... parentNamespace);
+
+    /**
+     * Check if a namespace exists.
+     */
+    boolean namespaceExists(String... namespaceId);
+
+    /**
+     * Create a namespace.
+     */
+    void createNamespace(Map<String, String> properties, String... namespaceId);
+
+    /**
+     * Drop a namespace.
+     */
+    void dropNamespace(boolean cascade, String... namespaceId);
+
+    /**
+     * Get namespace metadata.
+     */
+    Map<String, String> getNamespaceMetadata(String... namespaceId);
+
+    /**
+     * List tables in a namespace.
+     */
+    List<String> listTables(String... namespaceId);
+
+    /**
+     * Check if a table exists.
+     */
+    boolean tableExists(String... tableId);
+
+    /**
+     * Create an empty table.
+     */
+    void createEmptyTable(String location, Map<String, String> 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<String, String> storageOptions;
+
+        public TableMetadata(String location, Map<String, String> storageOptions) {
+            this.location = location;
+            this.storageOptions = storageOptions;
+        }
+
+        public String getLocation() {
+            return location;
+        }
+
+        public Map<String, String> getStorageOptions() {
+            return storageOptions;
+        }
+    }
+}

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<String[]> parentPrefix = config.getParentArray();
+        java.util.Optional<String> 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;
+    }
+}