This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch trino-irc-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit fb42cd6b993bea466eb8725f4ccadffd8f225a45 Author: Yuhui <[email protected]> AuthorDate: Fri Aug 21 10:02:07 2026 +0800 [Cherry-pick to branch-1.3] [#12526] feat(trino-connector): Support Starburst SPI compatibility (#12527) --- .../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); + } +}
