Copilot commented on code in PR #11113:
URL: https://github.com/apache/gravitino/pull/11113#discussion_r3296851879
##########
lance/lance-common/src/main/java/org/apache/gravitino/lance/common/config/LanceConfig.java:
##########
@@ -60,6 +61,13 @@ public class LanceConfig extends Config implements
OverwriteDefaultConfig {
.stringConf()
.createWithDefault(GRAVITINO_URI);
+ public static final ConfigEntry<Boolean> INTERNAL_AUX_MODE =
+ new ConfigBuilder(CONFIG_INTERNAL_AUX_MODE)
+ .doc("Internal runtime flag indicating whether Lance REST runs as an
auxiliary service")
+ .version(ConfigConstants.VERSION_1_1_0)
Review Comment:
`INTERNAL_AUX_MODE` is described as an internal runtime flag, but it is
currently declared as a public config entry. Consider marking it as internal
via `new ConfigBuilder(...).internal()` so it doesn’t show up as a user-facing
configuration option (and helps avoid accidental external use).
##########
lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java:
##########
@@ -135,4 +168,413 @@ public Catalog loadAndValidateLakehouseCatalog(String
catalogName) {
}
return catalog;
}
+
+ String[] listSchemas(Catalog catalog) throws NoSuchCatalogException {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return java.util.Arrays.stream(
+ schemaDispatcher.listSchemas(Namespace.of(metalakeName,
catalog.name())))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return java.util.Arrays.stream(
+ ((org.apache.gravitino.connector.SupportsSchemas)
ops).listSchemas(Namespace.of()))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
Review Comment:
Avoid using fully qualified `org.apache.gravitino.connector.SupportsSchemas`
in code. Add an import for `org.apache.gravitino.connector.SupportsSchemas` and
use the simple name in the instanceof/cast, per the repo import guidelines
(AGENTS.md).
##########
lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java:
##########
@@ -135,4 +168,413 @@ public Catalog loadAndValidateLakehouseCatalog(String
catalogName) {
}
return catalog;
}
+
+ String[] listSchemas(Catalog catalog) throws NoSuchCatalogException {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return java.util.Arrays.stream(
+ schemaDispatcher.listSchemas(Namespace.of(metalakeName,
catalog.name())))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
+ }
Review Comment:
Avoid fully qualified `java.util.Arrays` usage here; repo style prefers
normal imports unless there is a real name conflict (AGENTS.md imports
guideline). Import `java.util.Arrays` and use `Arrays.stream(...)` instead to
keep the code consistent and more readable.
##########
lance/lance-common/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceNamespaceWrapper.java:
##########
@@ -229,4 +254,236 @@ public void testLanceConfigWithAllClientProperties() {
Assertions.assertEquals("10000",
allConfig.get("gravitino.client.connectionTimeoutMs"));
Assertions.assertEquals("60000",
allConfig.get("gravitino.client.socketTimeoutMs"));
}
+
+ @Test
+ public void testCreateCatalogFetcherUsesHttpClientInStandaloneMode() {
+ LanceConfig lanceConfig =
+ new LanceConfig(
+ ImmutableMap.of(
+ LanceConfig.METALAKE_NAME.getKey(), "test_metalake",
+ LanceConfig.NAMESPACE_BACKEND_URI.getKey(),
"http://localhost:8090";,
+ LanceConfig.INTERNAL_AUX_MODE.getKey(), "false"));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig);
+
+ GravitinoLanceNamespaceWrapper.CatalogFetcher fetcher =
+ wrapper.createCatalogFetcher("test_metalake");
+
+ Assertions.assertEquals("HttpCatalogFetcher",
fetcher.getClass().getSimpleName());
+ Assertions.assertDoesNotThrow(fetcher::close);
+ }
+
+ @Test
+ public void testLoadAndValidateLakehouseCatalogUsesCatalogFetcher() {
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper();
+ Catalog expectedCatalog = createCatalogProxy(Catalog.Type.RELATIONAL,
"lakehouse-generic");
+ wrapper.setCatalogFetcher(
+ new GravitinoLanceNamespaceWrapper.CatalogFetcher() {
+ @Override
+ public Catalog[] listCatalogsInfo() {
+ return new Catalog[0];
+ }
+
+ @Override
+ public Catalog loadCatalog(String catalogName) {
+ return expectedCatalog;
+ }
+
+ @Override
+ public Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Catalog alterCatalog(String catalogName, CatalogChange...
changes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean dropCatalog(String catalogName, boolean force) {
+ throw new UnsupportedOperationException();
+ }
+ });
+
+ Assertions.assertSame(expectedCatalog,
wrapper.loadAndValidateLakehouseCatalog("test_catalog"));
+ }
+
+ @Test
+ public void testLoadSchemaUsesSchemaDispatcherInAuxMode() throws Exception {
+ Schema expectedSchema = createSchemaProxy();
+ AtomicReference<NameIdentifier> loadedSchemaIdent = new
AtomicReference<>();
+ SchemaDispatcher schemaDispatcher =
+ (SchemaDispatcher)
+ Proxy.newProxyInstance(
+ SchemaDispatcher.class.getClassLoader(),
+ new Class<?>[] {SchemaDispatcher.class},
+ (proxy, method, args) -> {
+ if ("loadSchema".equals(method.getName())) {
+ loadedSchemaIdent.set((NameIdentifier) args[0]);
+ return expectedSchema;
+ }
+
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ return null;
+ });
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(),
+ "catalogDispatcher",
+ Proxy.newProxyInstance(
+ CatalogDispatcher.class.getClassLoader(),
+ new Class<?>[] {CatalogDispatcher.class},
+ (proxy, method, args) -> null),
+ true);
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "catalogManager",
allocateCatalogManager(), true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "schemaDispatcher",
schemaDispatcher, true);
+
+ LanceConfig lanceConfig =
+ new LanceConfig(
+ ImmutableMap.of(
+ LanceConfig.METALAKE_NAME.getKey(), "test_metalake",
+ LanceConfig.INTERNAL_AUX_MODE.getKey(), "true"));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig);
+ wrapper.asNamespaceOps();
+
+ Schema actualSchema =
+ wrapper.loadSchema(
+ createCatalogProxy(Catalog.Type.RELATIONAL, "lakehouse-generic"),
"test_schema");
+
+ Assertions.assertSame(expectedSchema, actualSchema);
+ Assertions.assertEquals(
+ NameIdentifierUtil.ofSchema("test_metalake", "test_catalog",
"test_schema"),
+ loadedSchemaIdent.get());
+ }
+
+ @Test
+ public void testAsTableCatalogUsesTableDispatcherInAuxMode() throws
Exception {
+ Table expectedTable = createTableProxy();
+ AtomicReference<NameIdentifier> loadedTableIdent = new AtomicReference<>();
+ AtomicReference<Namespace> listedNamespace = new AtomicReference<>();
+ TableDispatcher tableDispatcher =
+ (TableDispatcher)
+ Proxy.newProxyInstance(
+ TableDispatcher.class.getClassLoader(),
+ new Class<?>[] {TableDispatcher.class},
+ (proxy, method, args) -> {
+ if ("loadTable".equals(method.getName())) {
+ loadedTableIdent.set((NameIdentifier) args[0]);
+ return expectedTable;
+ }
+ if ("listTables".equals(method.getName())) {
+ listedNamespace.set((Namespace) args[0]);
+ return new NameIdentifier[]
{NameIdentifier.of("test_schema", "test_table")};
+ }
+
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ return null;
+ });
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(),
+ "catalogDispatcher",
+ Proxy.newProxyInstance(
+ CatalogDispatcher.class.getClassLoader(),
+ new Class<?>[] {CatalogDispatcher.class},
+ (proxy, method, args) -> null),
+ true);
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "catalogManager",
allocateCatalogManager(), true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "tableDispatcher",
tableDispatcher, true);
+
+ LanceConfig lanceConfig =
+ new LanceConfig(
+ ImmutableMap.of(
+ LanceConfig.METALAKE_NAME.getKey(), "test_metalake",
+ LanceConfig.INTERNAL_AUX_MODE.getKey(), "true"));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig);
+ wrapper.asTableOps();
+
+ TableCatalog tableCatalog =
+ wrapper.asTableCatalog(createCatalogProxy(Catalog.Type.RELATIONAL,
"lakehouse-generic"));
+
+ Assertions.assertSame(
+ expectedTable, tableCatalog.loadTable(NameIdentifier.of("test_schema",
"test_table")));
+ Assertions.assertArrayEquals(
+ new NameIdentifier[] {NameIdentifier.of("test_schema", "test_table")},
+ tableCatalog.listTables(Namespace.of("test_schema")));
+ Assertions.assertEquals(
+ NameIdentifierUtil.ofTable("test_metalake", "test_catalog",
"test_schema", "test_table"),
+ loadedTableIdent.get());
+ Assertions.assertEquals(
+ Namespace.of("test_metalake", "test_catalog", "test_schema"),
listedNamespace.get());
+ }
+
+ private Schema createSchemaProxy() {
+ return (Schema)
+ Proxy.newProxyInstance(
+ Schema.class.getClassLoader(),
+ new Class<?>[] {Schema.class},
+ (proxy, method, args) -> null);
+ }
+
+ private Table createTableProxy() {
+ return (Table)
+ Proxy.newProxyInstance(
+ Table.class.getClassLoader(),
+ new Class<?>[] {Table.class},
+ (proxy, method, args) -> null);
+ }
+
+ private CatalogManager allocateCatalogManager() throws Exception {
+ Object unsafe =
+ FieldUtils.readDeclaredStaticField(Class.forName("sun.misc.Unsafe"),
"theUnsafe", true);
+ return (CatalogManager)
+ unsafe
+ .getClass()
+ .getMethod("allocateInstance", Class.class)
+ .invoke(unsafe, CatalogManager.class);
Review Comment:
This test uses `sun.misc.Unsafe.allocateInstance` to fabricate a
`CatalogManager`. That can be brittle across JDKs/modules and may fail under
stricter runtime access rules. If possible, refactor the production code so
aux-mode doesn’t require `GravitinoEnv.catalogManager()` for these paths
(allowing tests to avoid `CatalogManager` entirely), or provide a supported
test helper to initialize `GravitinoEnv` components instead of relying on
Unsafe.
##########
lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java:
##########
@@ -135,4 +168,413 @@ public Catalog loadAndValidateLakehouseCatalog(String
catalogName) {
}
return catalog;
}
+
+ String[] listSchemas(Catalog catalog) throws NoSuchCatalogException {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return java.util.Arrays.stream(
+ schemaDispatcher.listSchemas(Namespace.of(metalakeName,
catalog.name())))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return java.util.Arrays.stream(
+ ((org.apache.gravitino.connector.SupportsSchemas)
ops).listSchemas(Namespace.of()))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
+ }
+ }
+
+ return catalog.asSchemas().listSchemas();
+ }
+
+ boolean schemaExists(Catalog catalog, String schemaName) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.schemaExists(schemaIdent(catalog.name(),
schemaName));
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return ((org.apache.gravitino.connector.SupportsSchemas) ops)
+ .schemaExists(NameIdentifier.of(schemaName));
+ }
+ }
+
+ return catalog.asSchemas().schemaExists(schemaName);
+ }
+
+ Schema loadSchema(Catalog catalog, String schemaName) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.loadSchema(schemaIdent(catalog.name(),
schemaName));
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return ((org.apache.gravitino.connector.SupportsSchemas) ops)
+ .loadSchema(NameIdentifier.of(schemaName));
+ }
+ }
+
+ return catalog.asSchemas().loadSchema(schemaName);
+ }
+
+ Schema createSchema(
+ Catalog catalog, String schemaName, String comment, Map<String, String>
properties) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.createSchema(
+ schemaIdent(catalog.name(), schemaName), comment, properties);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return ((org.apache.gravitino.connector.SupportsSchemas) ops)
+ .createSchema(NameIdentifier.of(schemaName), comment, properties);
+ }
+ }
+
+ return catalog.asSchemas().createSchema(schemaName, comment, properties);
+ }
+
+ Schema alterSchema(Catalog catalog, String schemaName, SchemaChange...
changes) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.alterSchema(schemaIdent(catalog.name(),
schemaName), changes);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return ((org.apache.gravitino.connector.SupportsSchemas) ops)
+ .alterSchema(NameIdentifier.of(schemaName), changes);
+ }
+ }
+
+ return catalog.asSchemas().alterSchema(schemaName, changes);
+ }
+
+ boolean dropSchema(Catalog catalog, String schemaName, boolean cascade) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.dropSchema(schemaIdent(catalog.name(),
schemaName), cascade);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof org.apache.gravitino.connector.SupportsSchemas) {
+ return ((org.apache.gravitino.connector.SupportsSchemas) ops)
+ .dropSchema(NameIdentifier.of(schemaName), cascade);
+ }
+ }
+
+ return catalog.asSchemas().dropSchema(schemaName, cascade);
+ }
+
+ TableCatalog asTableCatalog(Catalog catalog) {
+ TableDispatcher tableDispatcher = currentTableDispatcher();
+ if (tableDispatcher != null) {
+ return new InternalTableCatalogAdapter(catalog.name(), tableDispatcher);
+ }
+
+ if (catalog instanceof BaseCatalog) {
+ CatalogOperations ops = ((BaseCatalog<?>) catalog).ops();
+ if (ops instanceof TableCatalog) {
+ return (TableCatalog) ops;
+ }
+ }
+
+ return catalog.asTableCatalog();
+ }
+
+ private NameIdentifier schemaIdent(String catalogName, String schemaName) {
+ return NameIdentifierUtil.ofSchema(metalakeName, catalogName, schemaName);
+ }
+
+ private NameIdentifier tableIdent(String catalogName, NameIdentifier ident) {
+ return NameIdentifierUtil.ofTable(
+ metalakeName, catalogName, ident.namespace().level(0), ident.name());
+ }
+
+ private Namespace tableNamespace(String catalogName, Namespace namespace) {
+ return Namespace.of(metalakeName, catalogName, namespace.level(0));
+ }
+
+ private SchemaDispatcher currentSchemaDispatcher() {
+ if (!config().isAuxMode()) {
+ return null;
+ }
+
+ return GravitinoEnv.getInstance().schemaDispatcher();
+ }
+
+ private TableDispatcher currentTableDispatcher() {
+ if (!config().isAuxMode()) {
+ return null;
+ }
+
+ return GravitinoEnv.getInstance().tableDispatcher();
+ }
+
+ @VisibleForTesting
+ CatalogFetcher createCatalogFetcher(String metalakeName) {
+ return config().isAuxMode()
+ ? new InternalCatalogFetcher(metalakeName)
+ : new HttpCatalogFetcher(
+ config().get(NAMESPACE_BACKEND_URI), metalakeName, config(),
extractClientProperties());
+ }
+
+ @VisibleForTesting
+ void setCatalogFetcher(CatalogFetcher catalogFetcher) {
+ this.catalogFetcher = catalogFetcher;
+ }
+
+ private Map<String, String> extractClientProperties() {
+ Map<String, String> clientProperties = new HashMap<>();
+ config()
+ .getAllConfig()
+ .forEach(
+ (key, value) -> {
+ if (key.startsWith("gravitino.client.")) {
+ clientProperties.put(key, value);
+ LOG.info("Applying client config: {} = {}", key, value);
+ }
+ });
+ return clientProperties;
+ }
+
+ @VisibleForTesting
+ static GravitinoClient createGravitinoClient(
+ String uri, String metalakeName, LanceConfig config, Map<String, String>
clientProperties) {
+ ClientBuilder builder =
GravitinoClient.builder(uri).withMetalake(metalakeName);
+ builder.withClientConfig(clientProperties);
+ return builder.build();
+ }
+
+ interface CatalogFetcher extends Closeable {
+
+ Catalog[] listCatalogsInfo() throws NoSuchMetalakeException;
+
+ Catalog loadCatalog(String catalogName) throws NoSuchCatalogException;
+
+ Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties)
+ throws NoSuchMetalakeException, CatalogAlreadyExistsException;
+
+ Catalog alterCatalog(String catalogName, CatalogChange... changes)
+ throws NoSuchCatalogException;
+
+ boolean dropCatalog(String catalogName, boolean force)
+ throws NonEmptyEntityException, CatalogInUseException;
+
+ @Override
+ default void close() {}
+ }
+
+ private class InternalTableCatalogAdapter implements TableCatalog {
+ private final String catalogName;
+ private final TableDispatcher dispatcher;
+
+ private InternalTableCatalogAdapter(String catalogName, TableDispatcher
dispatcher) {
+ this.catalogName = catalogName;
+ this.dispatcher = dispatcher;
+ }
+
+ @Override
+ public NameIdentifier[] listTables(Namespace namespace) throws
NoSuchSchemaException {
+ return dispatcher.listTables(tableNamespace(catalogName, namespace));
+ }
+
+ @Override
+ public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
+ return dispatcher.loadTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public Table createTable(
+ NameIdentifier ident,
+ Column[] columns,
+ String comment,
+ Map<String, String> properties,
+ Transform[] partitions,
+ Distribution distribution,
+ SortOrder[] sortOrders,
+ Index[] indexes)
+ throws NoSuchSchemaException, TableAlreadyExistsException {
+ return dispatcher.createTable(
+ tableIdent(catalogName, ident),
+ columns,
+ comment,
+ properties,
+ partitions,
+ distribution,
+ sortOrders,
+ indexes);
+ }
+
+ @Override
+ public Table alterTable(NameIdentifier ident, TableChange... changes)
+ throws NoSuchTableException, IllegalArgumentException {
+ return dispatcher.alterTable(tableIdent(catalogName, ident), changes);
+ }
+
+ @Override
+ public boolean dropTable(NameIdentifier ident) {
+ return dispatcher.dropTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public boolean purgeTable(NameIdentifier ident) throws
UnsupportedOperationException {
+ return dispatcher.purgeTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public boolean tableExists(NameIdentifier ident) {
+ return dispatcher.tableExists(tableIdent(catalogName, ident));
+ }
+ }
+
+ private static class InternalCatalogFetcher implements CatalogFetcher {
+ private final String metalakeName;
+ private final CatalogDispatcher catalogDispatcher;
+ private final CatalogManager catalogManager;
+
+ private InternalCatalogFetcher(String metalakeName) {
+ this.metalakeName = metalakeName;
+ CatalogDispatcher dispatcher =
GravitinoEnv.getInstance().catalogDispatcher();
+ Preconditions.checkState(
+ dispatcher != null,
+ "CatalogDispatcher is not available. Internal catalog fetcher
requires Gravitino server mode.");
+ this.catalogDispatcher = dispatcher;
+ this.catalogManager = GravitinoEnv.getInstance().catalogManager();
Review Comment:
`InternalCatalogFetcher` treats `catalogManager` as optional (null-check in
`loadCatalog`), but the constructor unconditionally calls
`GravitinoEnv.getInstance().catalogManager()`, which throws if the env isn't
initialized (and also means `catalogManager` can never be null in normal
execution). Either require `CatalogManager` explicitly (and remove the dead
null-branch), or retrieve it in a way that allows it to be absent and fall back
to `catalogDispatcher` as intended.
--
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]
