yuqi1129 commented on code in PR #13830: URL: https://github.com/apache/gravitino/pull/13830#discussion_r4266496321
########## flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/lance/LancePropertiesConverter.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.gravitino.flink.connector.lance; + +import com.google.common.collect.ImmutableSet; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import org.apache.commons.lang3.StringUtils; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.gravitino.flink.connector.CatalogPropertiesConverter; +import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter; + +/** + * Converter between Gravitino Lance catalog and table properties and Apache Flink properties. + */ +public class LancePropertiesConverter + implements CatalogPropertiesConverter, SchemaAndTablePropertiesConverter { + + /** Singleton instance of {@link LancePropertiesConverter}. */ + public static final LancePropertiesConverter INSTANCE = new LancePropertiesConverter(); + + private static final Set<String> PASSTHROUGH_CATALOG_KEYS = + ImmutableSet.of( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_BACKEND, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_IMPL, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_WAREHOUSE, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_LOCATION); + + private LancePropertiesConverter() {} + + @Override + public String transformPropertyToGravitinoCatalog(String configKey) { + if (PASSTHROUGH_CATALOG_KEYS.contains(configKey) + || configKey.startsWith(LancePropertiesConstants.LANCE_STORAGE_PREFIX)) { + return configKey; + } + return null; + } + + @Override + public String transformPropertyToFlinkCatalog(String configKey) { + if (PASSTHROUGH_CATALOG_KEYS.contains(configKey) + || configKey.startsWith(LancePropertiesConstants.LANCE_STORAGE_PREFIX)) { + return configKey; + } + return null; + } + + @Override + public Map<String, String> toFlinkCatalogProperties(Map<String, String> gravitinoProperties) { + Map<String, String> flinkProperties = + CatalogPropertiesConverter.super.toFlinkCatalogProperties(gravitinoProperties); + flinkProperties.putIfAbsent( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_BACKEND, + LancePropertiesConstants.LANCE_CATALOG_BACKEND_REST); + flinkProperties.putIfAbsent( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_IMPL, + LancePropertiesConstants.LANCE_CATALOG_BACKEND_REST); + return flinkProperties; + } + + @Override + public Map<String, String> toFlinkTableProperties( + Map<String, String> flinkCatalogProperties, + Map<String, String> gravitinoTableProperties, + ObjectPath tablePath) { + Map<String, String> tableProperties = + new HashMap<>( + gravitinoTableProperties != null ? gravitinoTableProperties : Collections.emptyMap()); + + String location = tableProperties.get(LancePropertiesConstants.LANCE_TABLE_LOCATION); + if (StringUtils.isBlank(location)) { + location = tableProperties.get(LancePropertiesConstants.LANCE_TABLE_PATH); + } + if (StringUtils.isBlank(location) && flinkCatalogProperties != null) { + String baseLocation = + flinkCatalogProperties.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_WAREHOUSE); + if (StringUtils.isBlank(baseLocation)) { + baseLocation = + flinkCatalogProperties.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_LOCATION); + } + if (StringUtils.isNotBlank(baseLocation) && tablePath != null) { + location = + StringUtils.stripEnd(baseLocation, "/") + + "/" + + tablePath.getDatabaseName() + + "/" + + tablePath.getObjectName(); + } + } + + if (StringUtils.isNotBlank(location)) { + tableProperties.put(LancePropertiesConstants.LANCE_TABLE_PATH, location); + tableProperties.put(LancePropertiesConstants.LANCE_TABLE_LOCATION, location); Review Comment: lance-flink's `LanceDynamicTableFactory` only accepts `path`, `read.*`, `write.*`, `index.*` and `vector.*`, and calls `helper.validateExcept(...)` that only skips `hadoop.*` keys. Since the Flink table options here contain `location`, plus every Gravitino table property (e.g. `format`) and `lance.storage.*`, every `SELECT`/`INSERT` would fail with `ValidationException: Unsupported options found for 'lance'`. lance-flink also does not read `lance.storage.*`, so storage settings and credentials are not passed down either. Could you keep only `connector`, `path` and the options lance-flink supports? ########## flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/lance/GravitinoLanceCatalogFactory.java: ########## @@ -0,0 +1,124 @@ +/* + * 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.gravitino.flink.connector.lance; + +import com.google.common.collect.ImmutableSet; +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.gravitino.flink.connector.CatalogPropertiesConverter; +import org.apache.gravitino.flink.connector.DefaultPartitionConverter; +import org.apache.gravitino.flink.connector.PartitionConverter; +import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter; +import org.apache.gravitino.flink.connector.catalog.BaseCatalogFactory; +import org.apache.gravitino.flink.connector.utils.FactoryUtils; + +/** Factory for creating {@link GravitinoLanceCatalog} instances. */ +public abstract class GravitinoLanceCatalogFactory implements BaseCatalogFactory { + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtils.createCatalogFactoryHelper(this, context); + return newCatalog( + context.getName(), + helper.getOptions().get(GravitinoLanceCatalogFactoryOptions.DEFAULT_DATABASE), + schemaAndTablePropertiesConverter(), + partitionConverter(), + context.getOptions()); + } + + /** + * Creates a new version-specific {@link GravitinoLanceCatalog} instance. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param schemaAndTablePropertiesConverter the properties converter + * @param partitionConverter the partition converter + * @param catalogOptions the catalog options + * @return the created catalog instance + */ + protected abstract Catalog newCatalog( + String catalogName, + String defaultDatabase, + SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter, + PartitionConverter partitionConverter, + Map<String, String> catalogOptions); + + @Override + public String factoryIdentifier() { + return GravitinoLanceCatalogFactoryOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return ImmutableSet.of( + GravitinoLanceCatalogFactoryOptions.DEFAULT_DATABASE, + GravitinoLanceCatalogFactoryOptions.URI, + GravitinoLanceCatalogFactoryOptions.CATALOG_BACKEND, + GravitinoLanceCatalogFactoryOptions.IMPL, + GravitinoLanceCatalogFactoryOptions.WAREHOUSE, + GravitinoLanceCatalogFactoryOptions.LOCATION); + } + + @Override + public String gravitinoCatalogProvider() { + return "lakehouse-generic"; + } + + @Override + public boolean supportsProvider(String provider) { Review Comment: Two concerns: - A `lakehouse-generic` catalog can also hold non-Lance tables (e.g. Delta), and `getTable` doesn't check `format`, so those tables would get `connector=lance`. The Lance REST server rejects non-Lance tables explicitly (see "Format boundary" in `docs/lance-rest-integration.md`); something similar is needed here. - `lakehouse-lance` and `lance` are not existing Gravitino catalog providers. ########## flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/lance/GravitinoLanceCatalog.java: ########## @@ -0,0 +1,447 @@ +/* + * 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.gravitino.flink.connector.lance; + +import java.lang.reflect.Constructor; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.commons.lang3.StringUtils; +import org.apache.flink.table.catalog.AbstractCatalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogColumnStatistics; +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.CatalogTableStatistics; +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.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; +import org.apache.flink.table.catalog.exceptions.TablePartitionedException; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.factories.Factory; +import org.apache.flink.util.Preconditions; +import org.apache.gravitino.Catalog; +import org.apache.gravitino.flink.connector.PartitionConverter; +import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter; +import org.apache.gravitino.flink.connector.catalog.BaseCatalog; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Gravitino Lance Catalog implementation for Apache Flink. + * + * <p>All schema and table DDL operations route through Apache Gravitino server, keeping Gravitino + * as the authoritative source of truth. The catalog enables accessing tables without requiring + * manually specified storage paths for every individual table in Flink SQL. + */ +public abstract class GravitinoLanceCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoLanceCatalog.class); + + private final Map<String, String> catalogOptions; + private AbstractCatalog lanceCatalog; + + /** + * Constructs a new {@link GravitinoLanceCatalog}. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param schemaAndTablePropertiesConverter the properties converter + * @param partitionConverter the partition converter + * @param catalogOptions the catalog configuration options + */ + protected GravitinoLanceCatalog( + String catalogName, + String defaultDatabase, + SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter, + PartitionConverter partitionConverter, + Map<String, String> catalogOptions) { + this( + catalogName, + defaultDatabase, + schemaAndTablePropertiesConverter, + partitionConverter, + catalogOptions, + null); + } + + /** + * Constructs a new {@link GravitinoLanceCatalog} with an explicit inner catalog. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param schemaAndTablePropertiesConverter the properties converter + * @param partitionConverter the partition converter + * @param catalogOptions the catalog configuration options + * @param innerCatalog an explicit inner catalog instance + */ + protected GravitinoLanceCatalog( + String catalogName, + String defaultDatabase, + SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter, + PartitionConverter partitionConverter, + Map<String, String> catalogOptions, + AbstractCatalog innerCatalog) { + super( + catalogName, + catalogOptions, + defaultDatabase, + schemaAndTablePropertiesConverter, + partitionConverter); + this.catalogOptions = catalogOptions; + this.lanceCatalog = innerCatalog; + } + + @Override + public void open() throws CatalogException { + if (lanceCatalog == null) { + String uri = getUri(); + Preconditions.checkArgument( + StringUtils.isNotBlank(uri), + "Lance REST catalog '%s' requires a non-blank 'uri' property", + catalogName()); + this.lanceCatalog = createInnerCatalog(getName(), getDefaultDatabase(), catalogOptions); + } + super.open(); + } + + @Override + public Optional<Factory> getFactory() { + if (lanceCatalog != null) { + return lanceCatalog.getFactory(); + } + return Optional.empty(); + } + + @Override + protected AbstractCatalog realCatalog() { + Preconditions.checkState( + lanceCatalog != null, "Catalog '%s' has not been opened", getName()); + return lanceCatalog; + } + + /** + * Returns the Lance REST server endpoint URI configured for this catalog. + * + * @return the Lance REST server URI + */ + public String getUri() { + String uri = catalogOptions.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + uri = catalogOptions.get("lance.uri"); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + try { + Catalog gravitinoCatalog = catalog(); + if (gravitinoCatalog != null && gravitinoCatalog.properties() != null) { + uri = + gravitinoCatalog.properties().get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + return gravitinoCatalog.properties().get("lance.uri"); + } + } catch (Exception e) { + LOG.debug("Unable to load catalog info from Gravitino server during URI lookup", e); + } + return uri; + } + + /** + * Creates the inner Lance catalog delegate. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param properties the catalog properties + * @return the created inner catalog + */ + protected AbstractCatalog createInnerCatalog( + String catalogName, String defaultDatabase, Map<String, String> properties) { + try { + Class<?> clazz = + Class.forName("org.apache.flink.connector.lance.table.LanceNamespaceCatalog"); Review Comment: lance-flink's `LanceNamespaceCatalog` has no `(String, String)` constructor; its constructor is `(name, defaultDatabase, LanceNamespace, BufferAllocator, Map)`. So this lookup never matches, and the catalog always falls back to `GenericLanceInnerCatalog`, with only a debug log. Note that `uri` is not passed to the inner catalog at all, so the Lance REST endpoint is never used. ########## flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/lance/LancePropertiesConverter.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.gravitino.flink.connector.lance; + +import com.google.common.collect.ImmutableSet; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import org.apache.commons.lang3.StringUtils; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.gravitino.flink.connector.CatalogPropertiesConverter; +import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter; + +/** + * Converter between Gravitino Lance catalog and table properties and Apache Flink properties. + */ +public class LancePropertiesConverter + implements CatalogPropertiesConverter, SchemaAndTablePropertiesConverter { + + /** Singleton instance of {@link LancePropertiesConverter}. */ + public static final LancePropertiesConverter INSTANCE = new LancePropertiesConverter(); + + private static final Set<String> PASSTHROUGH_CATALOG_KEYS = + ImmutableSet.of( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_BACKEND, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_IMPL, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_WAREHOUSE, + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_LOCATION); + + private LancePropertiesConverter() {} + + @Override + public String transformPropertyToGravitinoCatalog(String configKey) { + if (PASSTHROUGH_CATALOG_KEYS.contains(configKey) + || configKey.startsWith(LancePropertiesConstants.LANCE_STORAGE_PREFIX)) { + return configKey; + } + return null; + } + + @Override + public String transformPropertyToFlinkCatalog(String configKey) { + if (PASSTHROUGH_CATALOG_KEYS.contains(configKey) + || configKey.startsWith(LancePropertiesConstants.LANCE_STORAGE_PREFIX)) { + return configKey; + } + return null; + } + + @Override + public Map<String, String> toFlinkCatalogProperties(Map<String, String> gravitinoProperties) { + Map<String, String> flinkProperties = + CatalogPropertiesConverter.super.toFlinkCatalogProperties(gravitinoProperties); + flinkProperties.putIfAbsent( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_BACKEND, + LancePropertiesConstants.LANCE_CATALOG_BACKEND_REST); + flinkProperties.putIfAbsent( + LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_IMPL, + LancePropertiesConstants.LANCE_CATALOG_BACKEND_REST); + return flinkProperties; + } + + @Override + public Map<String, String> toFlinkTableProperties( + Map<String, String> flinkCatalogProperties, + Map<String, String> gravitinoTableProperties, + ObjectPath tablePath) { + Map<String, String> tableProperties = + new HashMap<>( + gravitinoTableProperties != null ? gravitinoTableProperties : Collections.emptyMap()); + + String location = tableProperties.get(LancePropertiesConstants.LANCE_TABLE_LOCATION); + if (StringUtils.isBlank(location)) { + location = tableProperties.get(LancePropertiesConstants.LANCE_TABLE_PATH); + } + if (StringUtils.isBlank(location) && flinkCatalogProperties != null) { + String baseLocation = + flinkCatalogProperties.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_WAREHOUSE); + if (StringUtils.isBlank(baseLocation)) { + baseLocation = + flinkCatalogProperties.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_LOCATION); + } + if (StringUtils.isNotBlank(baseLocation) && tablePath != null) { + location = + StringUtils.stripEnd(baseLocation, "/") + + "/" + + tablePath.getDatabaseName() + + "/" + + tablePath.getObjectName(); + } + } + + if (StringUtils.isNotBlank(location)) { + tableProperties.put(LancePropertiesConstants.LANCE_TABLE_PATH, location); + tableProperties.put(LancePropertiesConstants.LANCE_TABLE_LOCATION, location); + } + + tableProperties.put( + LancePropertiesConstants.FLINK_CONNECTOR, LancePropertiesConstants.LANCE_CONNECTOR_TYPE); + + if (flinkCatalogProperties != null) { + for (Map.Entry<String, String> entry : flinkCatalogProperties.entrySet()) { + if (entry.getKey().startsWith(LancePropertiesConstants.LANCE_STORAGE_PREFIX)) { + tableProperties.putIfAbsent(entry.getKey(), entry.getValue()); + } + } + } + + return tableProperties; + } + + @Override + public Map<String, String> toGravitinoTableProperties(Map<String, String> flinkProperties) { + Map<String, String> properties = + new HashMap<>(flinkProperties != null ? flinkProperties : Collections.emptyMap()); + String path = properties.get(LancePropertiesConstants.LANCE_TABLE_PATH); + if (StringUtils.isNotBlank(path) + && !properties.containsKey(LancePropertiesConstants.LANCE_TABLE_LOCATION)) { + properties.put(LancePropertiesConstants.LANCE_TABLE_LOCATION, path); + } + properties.putIfAbsent( + LancePropertiesConstants.LANCE_TABLE_FORMAT, LancePropertiesConstants.LANCE_TABLE_FORMAT); Review Comment: `LANCE_TABLE_FORMAT` is `"lance"`, so this writes the property `lance=lance`, not `format=lance`. The generic catalog requires `format` (`Table.PROPERTY_TABLE_FORMAT`, required in `GenericTablePropertiesMetadata`), so `CREATE TABLE` from Flink is rejected by the server. Also, the Flink-only options (`connector`, `path`) are copied into the Gravitino table properties as is. Suggest `putIfAbsent(Table.PROPERTY_TABLE_FORMAT, "lance")` and filtering out Flink-only keys. ########## docs/flink-connector/flink-catalog-lance.md: ########## @@ -0,0 +1,91 @@ +--- +title: "Flink Connector: Lance REST Catalog" +slug: "/flink-connector/flink-catalog-lance" +keyword: "flink connector lance rest catalog" +license: "This software is licensed under the Apache License version 2." +--- + +## Introduction + +The Apache Gravitino Flink connector supports accessing Lance REST catalogs without requiring users to manually specify storage paths for every table in Flink SQL. +Tables and schemas are managed through Apache Gravitino, which serves as the authoritative metadata source and automatically populates table locations and Lance connector properties for query execution. + +## Capabilities + +### DDL and DML Operations + +- `CREATE CATALOG` +- `CREATE DATABASE` / `DROP DATABASE` +- `CREATE TABLE` / `DROP TABLE` / `ALTER TABLE` +- `SELECT` queries across tables without manually configuring `'connector' = 'lance'` or `'path'` on every query. + +### Automatic Table Path Resolution + +When querying or creating tables in a Gravitino Lance catalog: +1. The table's storage location is automatically retrieved from Gravitino metadata (or derived from the catalog's warehouse location: `<warehouse>/<database>/<table>`). +2. Flink table scan options are automatically populated with `'connector' = 'lance'` and `'path' = '<location>'`. +3. Storage configurations prefixed with `lance.storage.` configured on the catalog are automatically propagated to table options. + +## Getting Started + +### Prerequisites + +* Flink 1.19, 1.20, or 2.1 Review Comment: lance-flink only has Flink 1.18/1.19/1.20 modules, so the 2.1 support (and the v2.1 module registration) can't work for reads and writes yet. Also, `uri` is documented as required below, but the code doesn't use it. ########## flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/lance/GravitinoLanceCatalog.java: ########## @@ -0,0 +1,447 @@ +/* + * 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.gravitino.flink.connector.lance; + +import java.lang.reflect.Constructor; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.commons.lang3.StringUtils; +import org.apache.flink.table.catalog.AbstractCatalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogColumnStatistics; +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.CatalogTableStatistics; +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.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; +import org.apache.flink.table.catalog.exceptions.TablePartitionedException; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.factories.Factory; +import org.apache.flink.util.Preconditions; +import org.apache.gravitino.Catalog; +import org.apache.gravitino.flink.connector.PartitionConverter; +import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter; +import org.apache.gravitino.flink.connector.catalog.BaseCatalog; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Gravitino Lance Catalog implementation for Apache Flink. + * + * <p>All schema and table DDL operations route through Apache Gravitino server, keeping Gravitino + * as the authoritative source of truth. The catalog enables accessing tables without requiring + * manually specified storage paths for every individual table in Flink SQL. + */ +public abstract class GravitinoLanceCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoLanceCatalog.class); + + private final Map<String, String> catalogOptions; + private AbstractCatalog lanceCatalog; + + /** + * Constructs a new {@link GravitinoLanceCatalog}. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param schemaAndTablePropertiesConverter the properties converter + * @param partitionConverter the partition converter + * @param catalogOptions the catalog configuration options + */ + protected GravitinoLanceCatalog( + String catalogName, + String defaultDatabase, + SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter, + PartitionConverter partitionConverter, + Map<String, String> catalogOptions) { + this( + catalogName, + defaultDatabase, + schemaAndTablePropertiesConverter, + partitionConverter, + catalogOptions, + null); + } + + /** + * Constructs a new {@link GravitinoLanceCatalog} with an explicit inner catalog. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param schemaAndTablePropertiesConverter the properties converter + * @param partitionConverter the partition converter + * @param catalogOptions the catalog configuration options + * @param innerCatalog an explicit inner catalog instance + */ + protected GravitinoLanceCatalog( + String catalogName, + String defaultDatabase, + SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter, + PartitionConverter partitionConverter, + Map<String, String> catalogOptions, + AbstractCatalog innerCatalog) { + super( + catalogName, + catalogOptions, + defaultDatabase, + schemaAndTablePropertiesConverter, + partitionConverter); + this.catalogOptions = catalogOptions; + this.lanceCatalog = innerCatalog; + } + + @Override + public void open() throws CatalogException { + if (lanceCatalog == null) { + String uri = getUri(); + Preconditions.checkArgument( + StringUtils.isNotBlank(uri), + "Lance REST catalog '%s' requires a non-blank 'uri' property", + catalogName()); + this.lanceCatalog = createInnerCatalog(getName(), getDefaultDatabase(), catalogOptions); + } + super.open(); + } + + @Override + public Optional<Factory> getFactory() { + if (lanceCatalog != null) { + return lanceCatalog.getFactory(); + } + return Optional.empty(); + } + + @Override + protected AbstractCatalog realCatalog() { + Preconditions.checkState( + lanceCatalog != null, "Catalog '%s' has not been opened", getName()); + return lanceCatalog; + } + + /** + * Returns the Lance REST server endpoint URI configured for this catalog. + * + * @return the Lance REST server URI + */ + public String getUri() { + String uri = catalogOptions.get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + uri = catalogOptions.get("lance.uri"); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + try { + Catalog gravitinoCatalog = catalog(); + if (gravitinoCatalog != null && gravitinoCatalog.properties() != null) { + uri = + gravitinoCatalog.properties().get(LancePropertiesConstants.GRAVITINO_LANCE_CATALOG_URI); + if (StringUtils.isNotBlank(uri)) { + return uri; + } + return gravitinoCatalog.properties().get("lance.uri"); + } + } catch (Exception e) { + LOG.debug("Unable to load catalog info from Gravitino server during URI lookup", e); + } + return uri; + } + + /** + * Creates the inner Lance catalog delegate. + * + * @param catalogName the catalog name + * @param defaultDatabase the default database name + * @param properties the catalog properties + * @return the created inner catalog + */ + protected AbstractCatalog createInnerCatalog( + String catalogName, String defaultDatabase, Map<String, String> properties) { + try { + Class<?> clazz = + Class.forName("org.apache.flink.connector.lance.table.LanceNamespaceCatalog"); + for (Constructor<?> ctor : clazz.getConstructors()) { + if (ctor.getParameterCount() == 2 + && ctor.getParameterTypes()[0].equals(String.class) + && ctor.getParameterTypes()[1].equals(String.class)) { + return (AbstractCatalog) ctor.newInstance(catalogName, defaultDatabase); + } + } + } catch (Throwable t) { + LOG.debug( + "Native LanceNamespaceCatalog not found or could not be loaded, using fallback inner catalog", + t); + } + return new GenericLanceInnerCatalog(catalogName, defaultDatabase); + } + + /** Minimal fallback inner catalog used when native Lance Flink connector is absent. */ + static class GenericLanceInnerCatalog extends AbstractCatalog { Review Comment: All write methods of this fallback are no-ops and the reads return empty results, so whatever `BaseCatalog` delegates to `realCatalog()` (partitions, functions, statistics) silently "succeeds". I'd rather fail explicitly, or drop the inner catalog entirely if metadata is meant to go through the Gravitino API. ########## flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/lance/TestGravitinoLanceCatalog.java: ########## @@ -0,0 +1,330 @@ +/* + * 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.gravitino.flink.connector.lance; + +import com.google.common.collect.ImmutableMap; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.gravitino.Catalog; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.flink.connector.DefaultPartitionConverter; +import org.apache.gravitino.flink.connector.utils.CatalogCompat; +import org.apache.gravitino.rel.Column; +import org.apache.gravitino.rel.Table; +import org.apache.gravitino.rel.TableCatalog; +import org.apache.gravitino.rel.ViewCatalog; +import org.apache.gravitino.rel.types.Types; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +public abstract class TestGravitinoLanceCatalog { + + protected abstract CatalogCompat catalogCompat(); + + private static class TestableGravitinoLanceCatalog extends GravitinoLanceCatalog { + + private final CatalogCompat catalogCompat; + private final Catalog mockGravitinoCatalog; + + TestableGravitinoLanceCatalog( + String catalogName, + String defaultDatabase, + Map<String, String> catalogOptions, + Catalog mockGravitinoCatalog, + CatalogCompat catalogCompat) { + super( + catalogName, + defaultDatabase, + LancePropertiesConverter.INSTANCE, + DefaultPartitionConverter.INSTANCE, + catalogOptions); + this.mockGravitinoCatalog = mockGravitinoCatalog; + this.catalogCompat = catalogCompat; + } + + @Override + protected CatalogCompat catalogCompat() { + return catalogCompat; + } + + @Override + protected Catalog catalog() { + return mockGravitinoCatalog; + } + } + + @Test + void testOpenWithValidUriSucceeds() { + Map<String, String> options = ImmutableMap.of("uri", "http://localhost:8090/lance";); + Catalog mockGravitino = Mockito.mock(Catalog.class); + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + + Assertions.assertDoesNotThrow(catalog::open); + Assertions.assertEquals("http://localhost:8090/lance";, catalog.getUri()); + Assertions.assertNotNull(catalog.realCatalog()); + } + + @Test + void testOpenWithMissingUriThrows() { + Map<String, String> options = Collections.emptyMap(); + Catalog mockGravitino = Mockito.mock(Catalog.class); + Mockito.when(mockGravitino.properties()).thenReturn(Collections.emptyMap()); + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + + IllegalArgumentException ex = + Assertions.assertThrows(IllegalArgumentException.class, catalog::open); + Assertions.assertTrue(ex.getMessage().contains("requires a non-blank 'uri' property")); + } + + @Test + void testOpenWithBlankUriThrows() { + Map<String, String> options = ImmutableMap.of("uri", " "); + Catalog mockGravitino = Mockito.mock(Catalog.class); + Mockito.when(mockGravitino.properties()).thenReturn(Collections.emptyMap()); + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + + IllegalArgumentException ex = + Assertions.assertThrows(IllegalArgumentException.class, catalog::open); + Assertions.assertTrue(ex.getMessage().contains("requires a non-blank 'uri' property")); + } + + @Test + void testOpenWithUriFromGravitinoCatalog() { + Map<String, String> options = Collections.emptyMap(); + Catalog mockGravitino = Mockito.mock(Catalog.class); + Mockito.when(mockGravitino.properties()) + .thenReturn(ImmutableMap.of("uri", "http://server-uri:8090/lance";)); + + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + + Assertions.assertDoesNotThrow(catalog::open); + Assertions.assertEquals("http://server-uri:8090/lance";, catalog.getUri()); + } + + @Test + void testRealCatalogRequiresOpen() { + Map<String, String> options = ImmutableMap.of("uri", "http://localhost:8090/lance";); + Catalog mockGravitino = Mockito.mock(Catalog.class); + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + + IllegalStateException ex = + Assertions.assertThrows(IllegalStateException.class, catalog::realCatalog); + Assertions.assertTrue(ex.getMessage().contains("has not been opened")); + } + + @Test + void testGetTableInjectsConnectorAndPath() throws Exception { + Map<String, String> options = ImmutableMap.of("uri", "http://localhost:8090/lance";); + Catalog mockGravitino = Mockito.mock(Catalog.class); + TableCatalog mockTableCatalog = Mockito.mock(TableCatalog.class); + Mockito.when(mockGravitino.asTableCatalog()).thenReturn(mockTableCatalog); + + Table mockTable = Mockito.mock(Table.class); + Mockito.when(mockTable.columns()) + .thenReturn( + new Column[] { + Column.of("id", Types.IntegerType.get(), "id column"), + Column.of("name", Types.StringType.get(), "name column") + }); + Mockito.when(mockTable.comment()).thenReturn("test table comment"); + Mockito.when(mockTable.index()).thenReturn(new org.apache.gravitino.rel.indexes.Index[0]); + Mockito.when(mockTable.distribution()) + .thenReturn(org.apache.gravitino.rel.expressions.distributions.Distributions.NONE); + Mockito.when(mockTable.partitioning()) + .thenReturn(new org.apache.gravitino.rel.expressions.transforms.Transform[0]); + Mockito.when(mockTable.properties()) + .thenReturn(ImmutableMap.of("location", "s3://lance-bucket/warehouse/my_db/users")); + + NameIdentifier ident = NameIdentifier.of("my_db", "users"); + Mockito.when(mockTableCatalog.loadTable(ident)).thenReturn(mockTable); + + TestableGravitinoLanceCatalog catalog = + new TestableGravitinoLanceCatalog( + "test_lance", "default", options, mockGravitino, catalogCompat()); + catalog.open(); + + CatalogBaseTable table = catalog.getTable(new ObjectPath("my_db", "users")); + Assertions.assertNotNull(table); + Map<String, String> tableOptions = table.getOptions(); + + // Verifies that Flink SQL user does not need to configure path or connector! + Assertions.assertEquals("lance", tableOptions.get("connector")); + Assertions.assertEquals("s3://lance-bucket/warehouse/my_db/users", tableOptions.get("path")); + Assertions.assertEquals( + "s3://lance-bucket/warehouse/my_db/users", tableOptions.get("location")); Review Comment: This asserts `location` in the Flink table options, which lance-flink's factory rejects (see the comment on `LancePropertiesConverter`). Since all tests are mock based, could you add at least one test that creates, writes and reads a table with lance-flink on the classpath against a real Gravitino server? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
