This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 7e26ce9ade [#12526] feat(trino-connector): Support Starburst SPI
compatibility (#12527)
7e26ce9ade is described below
commit 7e26ce9ade193292c4b97b99b2257b6b492e5821
Author: Yuhui <[email protected]>
AuthorDate: Fri Aug 21 10:02:07 2026 +0800
[#12526] feat(trino-connector): Support Starburst SPI compatibility (#12527)
### What changes were proposed in this pull request?
Add compatibility handling for Starburst SPI variants in the Trino
connector, including
`BlockEncodingManager` instantiation, SPI version parsing, and newer
DynamicFilter/ConnectorFactory
SPI methods.
### Why are the changes needed?
Starburst SPI differs from baseline open-source Trino in several
interfaces and internal
constructor signatures. Without these compatibility methods, the
Gravitino connector may
fail to load or execute on Starburst.
Fix: #12526
### Does this PR introduce any user-facing change?
The Gravitino Trino connector can run with compatible Starburst SPI
distributions. No new
configuration or public Gravitino API is introduced.
### How was this patch tested?
```text
./gradlew spotlessApply
./gradlew :trino-connector:trino-connector:test -PskipITs
```
Both commands completed successfully.
---
.../trino/connector/GravitinoConnectorFactory.java | 48 +++++++-
.../trino/connector/GravitinoDynamicFilter.java | 16 +++
.../trino/connector/util/json/JsonCodec.java | 123 ++++++++++++++++++++-
.../connector/TestGravitinoConnectorFactory.java | 74 +++++++++++++
.../trino/connector/util/json/TestJsonCodec.java | 105 ++++++++++++++++++
5 files changed, 362 insertions(+), 4 deletions(-)
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
index 1420a60c23..5315342713 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.trino.connector;
import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
+import static
org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT;
import static
org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR;
import com.google.common.annotations.VisibleForTesting;
@@ -28,7 +29,12 @@ import io.trino.spi.TrinoException;
import io.trino.spi.connector.Connector;
import io.trino.spi.connector.ConnectorContext;
import io.trino.spi.connector.ConnectorFactory;
+import java.util.Locale;
import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.client.GravitinoAdminClient;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorContext;
@@ -48,6 +54,9 @@ 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 = Integer.MAX_VALUE;
+ private static final Pattern TRINO_SPI_VERSION_PATTERN =
Pattern.compile("^(\\d+)");
+ private static final Set<String> SECURITY_SENSITIVE_PROPERTY_SUFFIXES =
+ Set.of("password", "secret", "token", "credential", "accesskey",
"secretkey", "privatekey");
/** The default connector name. */
public static final String DEFAULT_CONNECTOR_NAME = "gravitino";
@@ -142,6 +151,17 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
}
}
+ // Note: this method is not annotated with @Override because it does not
exist in the
+ // ConnectorFactory interface of the baseline open-source Trino SPI version
this connector
+ // compiles against. Some newer Trino/Starburst SPI versions declare it as
an abstract method,
+ // where it is dispatched at runtime by signature, providing cross-version
compatibility.
+ public Set<String> getSecuritySensitivePropertyNames(
+ String catalogName, Map<String, String> config, ConnectorContext
context) {
+ return config.keySet().stream()
+ .filter(GravitinoConnectorFactory::isSecuritySensitivePropertyName)
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
protected GravitinoConnector createConnector(CatalogConnectorContext
connectorContext) {
throw new TrinoException(NOT_SUPPORTED, "Should be overridden in
subclass");
}
@@ -157,7 +177,7 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
private void checkTrinoSpiVersion(ConnectorContext context, GravitinoConfig
config) {
String spiVersion = context.getSpiVersion();
- trinoVersion = Integer.parseInt(spiVersion);
+ trinoVersion = parseTrinoSpiVersion(spiVersion);
// check catalog name with metalake are supported in this trino version
if (!config.singleMetalakeMode() && !supportCatalogNameWithMetalake()) {
@@ -192,6 +212,32 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
}
}
+ @VisibleForTesting
+ static boolean isSecuritySensitivePropertyName(String propertyName) {
+ String normalizedPropertyName =
propertyName.toLowerCase(Locale.ROOT).replaceAll("[._-]", "");
+ return
SECURITY_SENSITIVE_PROPERTY_SUFFIXES.stream().anyMatch(normalizedPropertyName::endsWith);
+ }
+
+ @VisibleForTesting
+ static int parseTrinoSpiVersion(String spiVersion) {
+ Matcher matcher = TRINO_SPI_VERSION_PATTERN.matcher(spiVersion);
+ if (!matcher.find()) {
+ throw new TrinoException(
+ GRAVITINO_ILLEGAL_ARGUMENT,
+ String.format("Invalid Trino SPI version '%s': expected leading
digits", spiVersion));
+ }
+
+ try {
+ return Integer.parseInt(matcher.group(1));
+ } catch (NumberFormatException e) {
+ throw new TrinoException(
+ GRAVITINO_ILLEGAL_ARGUMENT,
+ String.format(
+ "Invalid Trino SPI version '%s': numeric version is out of
range", spiVersion),
+ e);
+ }
+ }
+
protected boolean supportCatalogNameWithMetalake() {
return true;
}
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoDynamicFilter.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoDynamicFilter.java
index bfc4f430e1..da75a1128d 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoDynamicFilter.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoDynamicFilter.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.trino.connector;
import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.DynamicFilter;
import io.trino.spi.predicate.TupleDomain;
+import java.lang.reflect.Method;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
@@ -55,6 +57,20 @@ class GravitinoDynamicFilter implements DynamicFilter {
return delegate.isAwaitable();
}
+ // Note: this method is not annotated with @Override because it does not
exist in the
+ // DynamicFilter interface of the baseline open-source Trino SPI version
this connector compiles
+ // against. It is present in newer Trino/Starburst SPI versions, where it is
dispatched at
+ // runtime by signature, providing cross-version compatibility. When the
delegate implements the
+ // same method, its value is forwarded so the underlying SPI's timeout
preference is preserved.
+ public OptionalLong getPreferredDynamicFilterTimeout() {
+ try {
+ Method method =
delegate.getClass().getMethod("getPreferredDynamicFilterTimeout");
+ return (OptionalLong) method.invoke(delegate);
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ return OptionalLong.empty();
+ }
+ }
+
@Override
public TupleDomain<ColumnHandle> getCurrentPredicate() {
return
delegate.getCurrentPredicate().transformKeys(GravitinoHandle::unWrap);
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/util/json/JsonCodec.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/util/json/JsonCodec.java
index 0566858223..00de547b70 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/util/json/JsonCodec.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/util/json/JsonCodec.java
@@ -30,6 +30,7 @@ import com.fasterxml.jackson.datatype.guava.GuavaModule;
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
+import com.google.common.annotations.VisibleForTesting;
import io.airlift.json.RecordAutoDetectModule;
import io.trino.spi.TrinoException;
import io.trino.spi.block.Block;
@@ -47,15 +48,22 @@ import io.trino.spi.type.StandardTypes;
import io.trino.spi.type.Type;
import io.trino.spi.type.TypeManager;
import io.trino.spi.type.TypeSignature;
+import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.Set;
import java.util.function.Function;
import org.apache.gravitino.trino.connector.GravitinoConnectorPluginManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Utility class for JSON serialization and deserialization in the Gravitino
Trino connector.
* Provides functionality for handling various Trino-specific types and plugin
classes.
*/
public class JsonCodec {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JsonCodec.class);
private static volatile ObjectMapper mapper;
private static volatile Type jsonType;
@@ -103,14 +111,123 @@ public class JsonCodec {
static BlockEncodingSerde createBlockEncodingSerde(TypeManager typeManager)
throws Exception {
ClassLoader classLoader = typeManager.getClass().getClassLoader();
- Class blockEncodingManagerClass =
+ Class<?> blockEncodingManagerClass =
classLoader.loadClass("io.trino.metadata.BlockEncodingManager");
- Class internalBlockEncodingSerdeClass =
+ Class<?> internalBlockEncodingSerdeClass =
classLoader.loadClass("io.trino.metadata.InternalBlockEncodingSerde");
+ Object blockEncodingManager =
+ instantiateBlockEncodingManager(blockEncodingManagerClass,
classLoader);
return (BlockEncodingSerde)
internalBlockEncodingSerdeClass
.getConstructor(blockEncodingManagerClass, TypeManager.class)
-
.newInstance(blockEncodingManagerClass.getConstructor().newInstance(),
typeManager);
+ .newInstance(blockEncodingManager, typeManager);
+ }
+
+ /**
+ * Instantiate BlockEncodingManager across Trino/Starburst variants. OSS
Trino exposes a public
+ * no-arg constructor; Starburst replaces it with
BlockEncodingManager(FeaturesConfig) (used to
+ * gate type-specific encodings via feature flags). Newer Trino branches
additionally publish a
+ * {@code Set<BlockEncoding>} variant for Guice multibindings. We probe each
known signature in
+ * order, then fall back to a generic constructor scan that fills unknown
reference parameters
+ * with default values as a last-resort compatibility mechanism.
+ */
+ @VisibleForTesting
+ static Object instantiateBlockEncodingManager(
+ Class<?> blockEncodingManagerClass, ClassLoader classLoader) throws
Exception {
+ try {
+ Object instance =
blockEncodingManagerClass.getConstructor().newInstance();
+ LOG.debug("Instantiated BlockEncodingManager with its public no-argument
constructor");
+ return instance;
+ } catch (NoSuchMethodException ignored) {
+ // fall through to parameterized variants
+ }
+
+ try {
+ Class<?> featuresConfigClass =
classLoader.loadClass("io.trino.FeaturesConfig");
+ Constructor<?> ctor =
blockEncodingManagerClass.getConstructor(featuresConfigClass);
+ Object featuresConfig =
featuresConfigClass.getConstructor().newInstance();
+ Object instance = ctor.newInstance(featuresConfig);
+ LOG.debug("Instantiated BlockEncodingManager with FeaturesConfig");
+ return instance;
+ } catch (NoSuchMethodException | ClassNotFoundException ignored) {
+ // fall through
+ }
+
+ try {
+ Constructor<?> setCtor =
blockEncodingManagerClass.getConstructor(Set.class);
+ Object instance = setCtor.newInstance(Collections.emptySet());
+ LOG.debug("Instantiated BlockEncodingManager with an empty BlockEncoding
set");
+ return instance;
+ } catch (NoSuchMethodException ignored) {
+ // fall through to last-resort scan
+ }
+
+ NoSuchMethodException lastError = null;
+ for (Constructor<?> ctor :
blockEncodingManagerClass.getDeclaredConstructors()) {
+ if (!ctor.trySetAccessible()) {
+ continue;
+ }
+
+ Class<?>[] paramTypes = ctor.getParameterTypes();
+ Object[] args = new Object[paramTypes.length];
+ boolean resolvable = true;
+ for (int i = 0; i < paramTypes.length && resolvable; i++) {
+ if (Set.class.isAssignableFrom(paramTypes[i])) {
+ args[i] = Collections.emptySet();
+ } else if (paramTypes[i].isPrimitive()) {
+ args[i] = defaultPrimitive(paramTypes[i]);
+ } else {
+ Object defaultInstance = tryDefaultInstance(paramTypes[i]);
+ if (defaultInstance == null) {
+ resolvable = false;
+ } else {
+ args[i] = defaultInstance;
+ }
+ }
+ }
+ if (!resolvable) {
+ continue;
+ }
+
+ try {
+ Object instance = ctor.newInstance(args);
+ LOG.debug("Instantiated BlockEncodingManager with fallback constructor
{}", ctor);
+ return instance;
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ lastError =
+ new NoSuchMethodException(
+ "Failed invoking BlockEncodingManager constructor " + ctor +
": " + e.getMessage());
+ }
+ }
+
+ throw new NoSuchMethodException(
+ "No usable BlockEncodingManager constructor found in "
+ + blockEncodingManagerClass.getName()
+ + (lastError == null ? "" : " (last error: " +
lastError.getMessage() + ")"));
+ }
+
+ private static Object tryDefaultInstance(Class<?> type) {
+ try {
+ Constructor<?> ctor = type.getDeclaredConstructor();
+ if (!ctor.trySetAccessible()) {
+ return null;
+ }
+ return ctor.newInstance();
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ return null;
+ }
+ }
+
+ private static Object defaultPrimitive(Class<?> type) {
+ if (type == boolean.class) return false;
+ if (type == char.class) return '\0';
+ if (type == byte.class) return (byte) 0;
+ if (type == short.class) return (short) 0;
+ if (type == int.class) return 0;
+ if (type == long.class) return 0L;
+ if (type == float.class) return 0f;
+ if (type == double.class) return 0d;
+ throw new IllegalArgumentException("Unsupported primitive type: " + type);
}
static void registerHandleSerializationModule(ObjectMapper objectMapper) {
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactory.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactory.java
new file mode 100644
index 0000000000..0b0c71c6b6
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactory.java
@@ -0,0 +1,74 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.trino.spi.TrinoException;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+class TestGravitinoConnectorFactory {
+
+ @Test
+ void testSecuritySensitivePropertyNames() {
+ GravitinoConnectorFactory factory = new GravitinoConnectorFactory(null);
+ Map<String, String> config =
+ Map.ofEntries(
+ Map.entry("trino.jdbc.password", "password"),
+ Map.entry("gravitino.client.oauth2.clientSecret", "secret"),
+ Map.entry("hive.s3.aws-access-key", "access-key"),
+ Map.entry("hive.s3.aws-secret-key", "secret-key"),
+ Map.entry("authentication.token", "token"),
+ Map.entry("oauth2.credential", "credential"),
+ Map.entry("tls.private_key", "private-key"),
+ Map.entry("gravitino.uri", "uri"));
+
+ Set<String> sensitivePropertyNames =
+ factory.getSecuritySensitivePropertyNames("catalog", config, null);
+
+ assertThat(sensitivePropertyNames)
+ .containsExactlyInAnyOrder(
+ "trino.jdbc.password",
+ "gravitino.client.oauth2.clientSecret",
+ "hive.s3.aws-access-key",
+ "hive.s3.aws-secret-key",
+ "authentication.token",
+ "oauth2.credential",
+ "tls.private_key");
+ assertThatThrownBy(() -> sensitivePropertyNames.add("another.password"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ void testParseTrinoSpiVersion() {
+
assertThat(GravitinoConnectorFactory.parseTrinoSpiVersion("478")).isEqualTo(478);
+
assertThat(GravitinoConnectorFactory.parseTrinoSpiVersion("478-e")).isEqualTo(478);
+
assertThat(GravitinoConnectorFactory.parseTrinoSpiVersion("478.1-vendor")).isEqualTo(478);
+ }
+
+ @Test
+ void testRejectInvalidTrinoSpiVersion() {
+ assertThatThrownBy(() ->
GravitinoConnectorFactory.parseTrinoSpiVersion("starburst-478"))
+ .isInstanceOf(TrinoException.class)
+ .hasMessage("Invalid Trino SPI version 'starburst-478': expected
leading digits");
+ }
+}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/util/json/TestJsonCodec.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/util/json/TestJsonCodec.java
new file mode 100644
index 0000000000..1b76b3000b
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/util/json/TestJsonCodec.java
@@ -0,0 +1,105 @@
+/*
+ * 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.util.json;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.trino.FeaturesConfig;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+class TestJsonCodec {
+
+ static class NoArgManager {}
+
+ static class FeaturesConfigManager {
+ final FeaturesConfig featuresConfig;
+
+ public FeaturesConfigManager(FeaturesConfig featuresConfig) {
+ this.featuresConfig = featuresConfig;
+ }
+ }
+
+ static class SetManager {
+ final Set<?> encodings;
+
+ public SetManager(Set<?> encodings) {
+ this.encodings = encodings;
+ }
+ }
+
+ static class FallbackManager {
+ final int flag;
+
+ private FallbackManager(int flag) {
+ this.flag = flag;
+ }
+ }
+
+ static class UnresolvableDependency {
+ UnresolvableDependency(String required) {}
+ }
+
+ static class UnresolvableManager {
+ final UnresolvableDependency dependency;
+
+ private UnresolvableManager(UnresolvableDependency dependency) {
+ this.dependency = dependency;
+ }
+ }
+
+ private static final ClassLoader CLASS_LOADER =
TestJsonCodec.class.getClassLoader();
+
+ @Test
+ void testPrefersNoArgConstructor() throws Exception {
+ Object instance =
JsonCodec.instantiateBlockEncodingManager(NoArgManager.class, CLASS_LOADER);
+ assertThat(instance).isInstanceOf(NoArgManager.class);
+ }
+
+ @Test
+ void testFallsBackToFeaturesConfigConstructor() throws Exception {
+ Object instance =
+ JsonCodec.instantiateBlockEncodingManager(FeaturesConfigManager.class,
CLASS_LOADER);
+ assertThat(instance).isInstanceOfSatisfying(FeaturesConfigManager.class, m
-> {});
+ }
+
+ @Test
+ void testFallsBackToSetConstructor() throws Exception {
+ Object instance =
JsonCodec.instantiateBlockEncodingManager(SetManager.class, CLASS_LOADER);
+ assertThat(instance)
+ .isInstanceOfSatisfying(SetManager.class, m ->
assertThat(m.encodings).isEmpty());
+ }
+
+ @Test
+ void testFallsBackToReflectiveScanForOtherConstructors() throws Exception {
+ Object instance =
+ JsonCodec.instantiateBlockEncodingManager(FallbackManager.class,
CLASS_LOADER);
+ assertThat(instance)
+ .isInstanceOfSatisfying(FallbackManager.class, m ->
assertThat(m.flag).isEqualTo(0));
+ }
+
+ @Test
+ void testThrowsWhenNoConstructorIsResolvable() {
+ assertThatThrownBy(
+ () ->
+
JsonCodec.instantiateBlockEncodingManager(UnresolvableManager.class,
CLASS_LOADER))
+ .isInstanceOf(NoSuchMethodException.class);
+ }
+}
