Copilot commented on code in PR #9735:
URL: https://github.com/apache/gravitino/pull/9735#discussion_r2726779156
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java:
##########
@@ -45,6 +45,8 @@
public class GravitinoConnectorFactory implements ConnectorFactory {
private static final Logger LOG =
LoggerFactory.getLogger(GravitinoConnectorFactory.class);
+ private static final int MIN_SUPPORT_TRINO_SPI_VERSION = 435;
+ private static final int MAX_SUPPORT_TRINO_SPI_VERSION = 440;
Review Comment:
The supported Trino version range is very narrow (435-440, only 6 versions).
This seems restrictive and may cause compatibility issues for users.
Additionally, the comment on line 173 mentions "compatiablity problem" with a
typo ("compatiablity" should be "compatibility").
Consider widening the supported version range or document the reasoning for
this narrow range. Also fix the typo in the warning message.
##########
.github/workflows/trino-integration-test.yml:
##########
@@ -1,11 +1,8 @@
name: Trino Integration Test
# Controls when the workflow will run
-on:
- push:
- branches: [ "main", "branch-*" ]
- pull_request:
- branches: [ "main", "branch-*" ]
+# Disable it temporarily.
+on: {}
Review Comment:
The CI workflow for Trino integration tests has been completely disabled by
setting `on: {}`. The comment says "Disable it temporarily", but there's no
issue reference or timeline for re-enabling it.
Add a reference to a tracking issue and a clear plan for when this will be
re-enabled. Disabling integration tests can lead to undetected regressions.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java:
##########
@@ -117,13 +137,70 @@ public Connector create(
}
GravitinoStoredProcedureFactory gravitinoStoredProcedureFactory =
new GravitinoStoredProcedureFactory(catalogConnectorManager,
metalake);
- return new GravitinoSystemConnector(gravitinoStoredProcedureFactory);
+ return createSystemConnector(gravitinoStoredProcedureFactory);
}
}
- @VisibleForTesting
- Supplier<GravitinoAdminClient> clientProvider() {
- return () -> null;
+ protected GravitinoConnector createConnector(CatalogConnectorContext
connectorContext) {
+ return new GravitinoConnector(connectorContext);
+ }
+
+ protected GravitinoSystemConnector createSystemConnector(
+ GravitinoStoredProcedureFactory storedProcedureFactory) {
+ return new GravitinoSystemConnector(storedProcedureFactory);
+ }
+
+ protected String getTrinoCatalogName(String metalakeName, String
catalogName) {
+ return "\"" + metalakeName + "." + catalogName + "\"";
+ }
+
+ private void checkTrinoSpiVersion(ConnectorContext context, GravitinoConfig
config) {
+ String spiVersion = context.getSpiVersion();
+
+ trinoVersion = Integer.parseInt(spiVersion);
+ if (trinoVersion < getMinSupportTrinoSpiVersion()
+ || trinoVersion > getMaxSupportTrinoSpiVersion()) {
+ Boolean skipTrinoVersionValidation =
config.isSkipTrinoVersionValidation();
+ if (!skipTrinoVersionValidation) {
+ String errmsg =
+ String.format(
+ "Unsupported Trino-%s version. The Supported version for the
Gravitino-Trino-connector from Trino-%d to Trino-%d."
+ + "Maybe you can set
gravitino.trino.skip-version-validation to skip version validation.",
+ trinoVersion, getMinSupportTrinoSpiVersion(),
getMaxSupportTrinoSpiVersion());
+ throw new
TrinoException(GravitinoErrorCode.GRAVITINO_UNSUPPORTED_TRINO_VERSION, errmsg);
+ } else {
+ LOG.warn(
+ "The version {} has not undergone thorough testing with Gravitino,
there may be compatiablity problem.",
Review Comment:
The typo "compatiablity" should be "compatibility".
```suggestion
"The version {} has not undergone thorough testing with
Gravitino, there may be compatibility problem.",
```
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java:
##########
@@ -222,11 +173,13 @@ private boolean checkCatalogExist(String name) {
ResultSet rs = statement.getResultSet();
while (rs.next()) {
String catalogName = rs.getString(1);
- if (catalogName.equals(name) || catalogName.equals("\"" + name +
"\"")) {
+ if (name.equals(catalogName) || name.equals("\"" + catalogName +
"\"")) {
return true;
}
}
return false;
+ } catch (SQLException e) {
+ throw e;
Review Comment:
These catch blocks catch SQLException and immediately rethrow it, which is
redundant. If the purpose is to handle SQLException differently from other
exceptions, the rethrow should either be removed, or the SQLException should be
wrapped in a different exception type.
Remove the redundant catch-and-rethrow blocks or add meaningful processing
before rethrowing.
##########
trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/GravitinoMockServer.java:
##########
@@ -603,6 +618,42 @@ void doAlterTable(
}
}
+ private void addColumn(
+ ConnectorMetadata metadata, ConnectorTableHandle tableHandle,
ColumnMetadata columnMetadata) {
+ if (trinoVersion < 468) {
+ try {
+ metadata
+ .getClass()
+ .getMethod(
+ "addColumn",
+ ConnectorSession.class,
+ ConnectorTableHandle.class,
+ ColumnMetadata.class)
+ .invoke(metadata, null, tableHandle, columnMetadata);
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException("Failed to invoke legacy addColumn by
reflection", e);
+ }
+ } else {
+ try {
+ Class<?> columnPositionClass =
Class.forName("io.trino.spi.connector.ColumnPosition");
+ Class<?> lastPositionClass =
Class.forName("io.trino.spi.connector.ColumnPosition$Last");
+ Object position =
lastPositionClass.getDeclaredConstructor().newInstance();
+
+ metadata
+ .getClass()
+ .getMethod(
+ "addColumn",
+ ConnectorSession.class,
+ ConnectorTableHandle.class,
+ ColumnMetadata.class,
+ columnPositionClass)
+ .invoke(metadata, null, tableHandle, columnMetadata, position);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to invoke addColumn with
ColumnPosition", e);
+ }
+ }
Review Comment:
The reflection code for handling different Trino versions uses a hard-coded
version check (trinoVersion < 468). However, this version number (468) falls
outside the supported version range defined in GravitinoConnectorFactory
(435-440). This suggests the version ranges may be inconsistent across the
codebase.
Verify that version checks are consistent across the codebase and align with
the documented supported version range.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -330,6 +332,10 @@ public List<GravitinoCatalog> getCatalogs() {
/** Shuts down the catalog connector manager. */
public void shutdown() {
LOG.info("Gravitino CatalogConnectorManager shutdown.");
+ if (catalogRegister != null) {
+ catalogRegister.close();
+ }
+ executorService.shutdown();
throw new NotImplementedException();
Review Comment:
The shutdown method calls `catalogRegister.close()` and
`executorService.shutdown()`, then throws a NotImplementedException. This means
these cleanup operations will always be followed by an exception, which
suggests incomplete implementation.
Either complete the shutdown implementation or remove the
NotImplementedException if shutdown is complete. Throwing
NotImplementedException after performing cleanup operations is confusing.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java:
##########
@@ -222,11 +173,13 @@ private boolean checkCatalogExist(String name) {
ResultSet rs = statement.getResultSet();
while (rs.next()) {
String catalogName = rs.getString(1);
- if (catalogName.equals(name) || catalogName.equals("\"" + name +
"\"")) {
+ if (name.equals(catalogName) || name.equals("\"" + catalogName +
"\"")) {
Review Comment:
The catalog name comparison logic has reversed operands that could cause
incorrect matching. The condition checks if `name.equals(catalogName) ||
name.equals("\"" + catalogName + "\"")`, but based on the context where `name`
is the parameter passed to checkCatalogExist and `catalogName` comes from the
result set, this should be `catalogName.equals(name) || ("\"" + catalogName +
"\"").equals(name)` to handle quoted catalog names correctly.
Verify the catalog name matching logic is correct for both quoted and
unquoted catalog names.
##########
trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector.java:
##########
@@ -189,38 +151,51 @@ public void testAlterTable() throws Exception {
createTestTable(fullTableName1);
- // test add column and drop column, but the memory connector is not
supported these operations.
- assertQueryFails(
- String.format("alter table %s add column if not exists c varchar",
fullTableName1),
- format("This connector does not support adding columns"));
-
- assertQueryFails(
- String.format("alter table %s drop column a", fullTableName1),
- format("This connector does not support dropping columns"));
-
// test set table comment
assertUpdate(String.format("comment on table %s is 'test table comments'",
fullTableName1));
assertThat((String) computeScalar("show create table " + fullTableName1))
.contains("COMMENT 'test table comments'");
- // test rename column, but the memory connector is not supported these
operations.
- assertQueryFails(
- String.format("alter table %s rename column a to c ", fullTableName1),
- format("This connector does not support renaming columns"));
-
- assertQueryFails(
- String.format("alter table %s alter column a set DATA TYPE int",
fullTableName1),
- format("This connector does not support setting column types"));
-
// test set column comment
assertUpdate(String.format("comment on column %s.a is 'test column
comments'", fullTableName1));
assertThat((String) computeScalar("show create table " + fullTableName1))
.contains("COMMENT 'test column comments'");
+ // test add column and drop column, but the memory connector is not
supported these operations.
+ if (trinoVersion < 469) {
+ assertQueryFails(
+ String.format("alter table %s add column if not exists c varchar",
fullTableName1),
+ "This connector does not support adding columns");
+ } else {
+ assertUpdate(
+ String.format("alter table %s add column if not exists c varchar",
fullTableName1));
+ assertThat((String) computeScalar("show create table " + fullTableName1))
+ .contains("c varchar");
+ }
+
+ assertQueryFails(
+ String.format("alter table %s drop column a", fullTableName1),
+ "This connector does not support dropping columns");
+
+ // test rename column, but the memory connector is not supported these
operations.
+ if (trinoVersion < 452) {
+ assertQueryFails(
+ String.format("alter table %s rename column b to d ",
fullTableName1),
+ "This connector does not support renaming columns");
+ } else {
+ assertUpdate(String.format("alter table %s rename column b to d ",
fullTableName1));
+ assertThat((String) computeScalar("show create table " + fullTableName1))
+ .contains("d integer");
+ }
Review Comment:
The test checks for version-specific behavior using version numbers (452,
469) that are well outside the supported version range defined in
GravitinoConnectorFactory (435-440). This indicates that either:
1. The supported version range in GravitinoConnectorFactory is incorrect
2. These tests are testing future/unsupported versions
3. The version range needs to be updated
Verify the correct supported version range and ensure consistency between
the factory's version checks and the test version checks.
```suggestion
// test add column and drop column, but the memory connector does not
support these operations.
assertQueryFails(
String.format("alter table %s add column if not exists c varchar",
fullTableName1),
"This connector does not support adding columns");
assertQueryFails(
String.format("alter table %s drop column a", fullTableName1),
"This connector does not support dropping columns");
// test rename column, but the memory connector does not support these
operations.
assertQueryFails(
String.format("alter table %s rename column b to d ",
fullTableName1),
"This connector does not support renaming columns");
```
##########
trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/AbstractGravitinoConnectorTest.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.trino.connector;
+
+import io.trino.plugin.memory.MemoryPlugin;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.client.GravitinoAdminClient;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import org.awaitility.Awaitility;
+
+abstract class AbstractGravitinoConnectorTest extends
AbstractTestQueryFramework {
+
+ GravitinoMockServer server;
+ int trinoVersion;
+
+ @Override
+ protected QueryRunner createQueryRunner() throws Exception {
+ GravitinoAdminClient gravitinoClient = createGravitinoClient();
+ try {
+
+ DistributedQueryRunner queryRunner = createTrinoQueryRunner();
+
+ GravitinoPlugin gravitinoPlugin = createGravitinoPulgin(gravitinoClient);
+ queryRunner.installPlugin(gravitinoPlugin);
+
+ configureCatalogs(queryRunner, gravitinoClient);
+
+
GravitinoConnectorPluginManager.instance(this.getClass().getClassLoader())
+ .installPlugin("memory", new MemoryPlugin());
+ CatalogConnectorManager catalogConnectorManager =
+ gravitinoPlugin.getCatalogConnectorManager();
+ trinoVersion = gravitinoPlugin.getTrinoVersion();
+ server.setCatalogConnectorManager(catalogConnectorManager, trinoVersion);
+
+ // Wait for the catalog to be created. Wait for at least 30 seconds.
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() -> !catalogConnectorManager.getCatalogs().isEmpty());
+
+ return queryRunner;
+ } catch (Exception e) {
+ throw new RuntimeException("Create query runner failed", e);
+ }
+ }
+
+ protected GravitinoPlugin createGravitinoPulgin(GravitinoAdminClient client)
{
Review Comment:
The method name has a typo: "createGravitinoPulgin" should be
"createGravitinoPlugin".
##########
trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/AbstractGravitinoConnectorTest.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.trino.connector;
+
+import io.trino.plugin.memory.MemoryPlugin;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.client.GravitinoAdminClient;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import org.awaitility.Awaitility;
+
+abstract class AbstractGravitinoConnectorTest extends
AbstractTestQueryFramework {
+
+ GravitinoMockServer server;
+ int trinoVersion;
+
+ @Override
+ protected QueryRunner createQueryRunner() throws Exception {
+ GravitinoAdminClient gravitinoClient = createGravitinoClient();
+ try {
+
+ DistributedQueryRunner queryRunner = createTrinoQueryRunner();
+
+ GravitinoPlugin gravitinoPlugin = createGravitinoPulgin(gravitinoClient);
Review Comment:
The method name "createGravitinoPulgin" has a typo and should be
"createGravitinoPlugin".
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java:
##########
@@ -86,14 +87,15 @@ public ConnectorMetadata getMetadata(
ConnectorMetadata internalMetadata =
internalConnector.getMetadata(session,
gravitinoTransactionHandle.getInternalHandle());
Preconditions.checkArgument(internalMetadata != null, "Internal metadata
must not be null");
+ return createGravitinoMetadata(
+ connectorMetadata, catalogConnectorContext.getMetadataAdapter(),
internalMetadata);
+ }
- GravitinoMetalake metalake = catalogConnectorContext.getMetalake();
-
- CatalogConnectorMetadata catalogConnectorMetadata =
- new CatalogConnectorMetadata(metalake, catalogIdentifier);
-
- return new GravitinoMetadata(
- catalogConnectorMetadata,
catalogConnectorContext.getMetadataAdapter(), internalMetadata);
+ protected GravitinoMetadata createGravitinoMetadata(
+ CatalogConnectorMetadata catalogConnectorMetadata,
+ CatalogConnectorMetadataAdapter metadataAdapter,
+ ConnectorMetadata internalMetadata) {
+ throw new RuntimeException("Should be overridden in subclass");
Review Comment:
This method throws a generic RuntimeException with a message "Should be
overridden in subclass". However, GravitinoConnector is not abstract, and the
createConnector method in GravitinoConnectorFactory directly instantiates
GravitinoConnector (line 145), not a subclass. This will cause a
RuntimeException to be thrown at runtime when getMetadata is called.
Either make GravitinoConnector abstract, or ensure that createConnector in
GravitinoConnectorFactory creates a concrete subclass that overrides this
method.
```suggestion
return catalogConnectorMetadata;
```
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/util/json/BlockJsonSerde.java:
##########
@@ -67,9 +67,7 @@ public void serialize(
throws IOException {
// Encoding name is length prefixed as are many block encodings
SliceOutput output =
- new DynamicSliceOutput(
- toIntExact(
- block.getSizeInBytes() + block.getEncodingName().length() +
(2 * Integer.BYTES)));
+ new DynamicSliceOutput(toIntExact(block.getSizeInBytes() + (2 *
Integer.BYTES) + 1024));
Review Comment:
The block size calculation changed from using
`block.getEncodingName().length()` to a fixed `1024` bytes. This change may
allocate too much or too little memory depending on the actual encoding name
length and could affect performance or cause buffer issues.
Document why this change was made and verify that 1024 bytes is sufficient
for all use cases, or use a more precise calculation.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -131,15 +135,13 @@ public void config(GravitinoConfig config,
GravitinoAdminClient client) {
* @throws Exception if the catalog connector manager fails to start
*/
public void start(ConnectorContext context) throws Exception {
- catalogRegister.init(context, config);
- if (catalogRegister.isCoordinator()) {
- executorService.scheduleWithFixedDelay(
- this::loadMetalake,
- metadataUpdateIntervalSecond,
- metadataUpdateIntervalSecond,
- TimeUnit.SECONDS);
- }
-
+ catalogRegister.init(config);
+ executorService.scheduleWithFixedDelay(
+ this::loadMetalake,
+ metadataUpdateIntervalSecond,
+ metadataUpdateIntervalSecond,
+ TimeUnit.SECONDS);
+ gravitinoClient.close();
Review Comment:
The `gravitinoClient.close()` call immediately after scheduling the
`loadMetalake` task will close the client that is still needed by the scheduled
task. The `loadMetalake()` method uses `gravitinoClient.listMetalakes()` and
`gravitinoClient.loadMetalake()` on lines 160 and 190, which will fail if the
client is closed.
Remove the `gravitinoClient.close()` call or ensure it's only called during
shutdown.
```suggestion
```
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoSplitSource.java:
##########
@@ -49,11 +49,15 @@ public CompletableFuture<ConnectorSplitBatch>
getNextBatch(int maxSize) {
.thenApply(
batch -> {
List<ConnectorSplit> list =
-
batch.getSplits().stream().map(GravitinoSplit::new).collect(Collectors.toList());
+
batch.getSplits().stream().map(this::createSplit).collect(Collectors.toList());
return new ConnectorSplitBatch(list, batch.isNoMoreSplits());
});
}
+ protected ConnectorSplit createSplit(ConnectorSplit split) {
+ throw new RuntimeException("Should be overridden in subclass");
Review Comment:
This method throws a generic RuntimeException with message "Should be
overridden in subclass". However, GravitinoSplitSource is not abstract, and
createSplitSource in GravitinoSplitManager (line 60) directly instantiates
GravitinoSplitSource. This will cause a RuntimeException when splits are
processed.
Either make GravitinoSplitSource abstract, or ensure createSplitSource
creates a concrete subclass that overrides this method.
```suggestion
return split;
```
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java:
##########
@@ -252,6 +205,8 @@ private void executeSql(String sql) {
// check the catalog is already created
statement.execute(sql);
return;
+ } catch (SQLException e) {
+ throw e;
Review Comment:
The catch block catches SQLException and immediately rethrows it without any
additional processing, making the catch block redundant.
Remove this redundant catch-and-rethrow block or add meaningful processing.
--
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]
