Add support for custom connector-provided serialization codecs

Author: tdcmeehan

pom.xml

@@ -2852,6 +2852,26 @@
                     <version>0.5.1</version>
                 </plugin>
 
+                <plugin>
+                    <groupId>pl.project13.maven</groupId>
+                    <artifactId>git-commit-id-plugin</artifactId>
+                    <version>4.0.3</version>
+                    <configuration>
+                        <useNativeGit>true</useNativeGit>
+                        <!-- Include only properties used above to speed up build (https://github.com/git-commit-id/git-commit-id-maven-plugin/issues/462) -->
+                        <includeOnlyProperties>
+                            <includeOnlyProperty>\Qgit.build.time</includeOnlyProperty>
+                            <includeOnlyProperty>\Qgit.commit.id</includeOnlyProperty>
+                            <includeOnlyProperty>\Qgit.commit.id.describe</includeOnlyProperty>
+                            <includeOnlyProperty>\Qgit.commit.id.abbrev</includeOnlyProperty>
+                        </includeOnlyProperties>
+                        <dateFormat>yyyy-MM-dd'T'HH:mm:ssZZ</dateFormat>
+                        <gitDescribe>
+                            <tags>true</tags>
+                        </gitDescribe>
+                    </configuration>
+                </plugin>
+
             </plugins>
         </pluginManagement>
 

presto-delta/src/test/java/com/facebook/presto/delta/TestDeltaTableHandle.java

@@ -25,8 +25,8 @@
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.metadata.FunctionAndTypeManager;
-import com.facebook.presto.metadata.HandleJsonModule;
 import com.facebook.presto.metadata.HandleResolver;
+import com.facebook.presto.metadata.TestingHandleJsonModule;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.type.TypeDeserializer;
 import com.google.common.collect.ImmutableList;
@@ -91,7 +91,7 @@ private JsonCodec<DeltaTableHandle> getJsonCodec()
     {
         Module module = binder -> {
             binder.install(new JsonModule());
-            binder.install(new HandleJsonModule());
+            binder.install(new TestingHandleJsonModule());
             configBinder(binder).bindConfig(FeaturesConfig.class);
             FunctionAndTypeManager functionAndTypeManager = createTestFunctionAndTypeManager();
             binder.bind(TypeManager.class).toInstance(functionAndTypeManager);

presto-docs/src/main/sphinx/admin/properties.rst

@@ -571,6 +571,24 @@ shared across all of the partitioned consumers. Increasing this value may
 improve network throughput for data transferred between stages if the
 network has high latency or if there are many nodes in the cluster.
 
+``use-connector-provided-serialization-codecs``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+Enables the use of custom connector-provided serialization codecs for handles. 
+This feature allows connectors to use their own serialization format for
+handle objects (such as table handles, column handles, and splits) instead
+of standard JSON serialization.
+
+When enabled, connectors that provide a ``ConnectorCodecProvider`` with 
+appropriate codecs will have their handles serialized using custom binary 
+formats, which are then Base64-encoded for transport. Connectors without 
+codec support automatically fall back to standard JSON serialization. 
+Internal Presto handles (prefixed with ``$``) always use JSON serialization 
+regardless of this setting.
+
 .. _task-properties:
 
 Task Properties

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplit.java

@@ -27,8 +27,8 @@
 import com.facebook.presto.hive.metastore.Storage;
 import com.facebook.presto.hive.metastore.StorageFormat;
 import com.facebook.presto.metadata.FunctionAndTypeManager;
-import com.facebook.presto.metadata.HandleJsonModule;
 import com.facebook.presto.metadata.HandleResolver;
+import com.facebook.presto.metadata.TestingHandleJsonModule;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.HostAddress;
 import com.facebook.presto.spi.SplitWeight;
@@ -153,8 +153,8 @@ private JsonCodec<HiveSplit> getJsonCodec()
     {
         Module module = binder -> {
             binder.install(new JsonModule());
-            binder.install(new HandleJsonModule());
             configBinder(binder).bindConfig(FeaturesConfig.class);
+            binder.install(new TestingHandleJsonModule());
             FunctionAndTypeManager functionAndTypeManager = createTestFunctionAndTypeManager();
             binder.bind(TypeManager.class).toInstance(functionAndTypeManager);
             jsonBinder(binder).addDeserializerBinding(Type.class).to(TypeDeserializer.class);

presto-main-base/src/main/java/com/facebook/presto/connector/ConnectorManager.java

@@ -393,6 +393,16 @@ private Connector createConnector(ConnectorId connectorId, ConnectorFactory fact
         }
     }
 
+    public Optional<ConnectorCodecProvider> getConnectorCodecProvider(ConnectorId connectorId)
+    {
+        requireNonNull(connectorId, "connectorId is null");
+        MaterializedConnector materializedConnector = connectors.get(connectorId);
+        if (materializedConnector == null) {
+            return Optional.empty();
+        }
+        return materializedConnector.getConnectorCodecProvider();
+    }
+
     private static class MaterializedConnector
     {
         private final ConnectorId connectorId;

presto-main-base/src/main/java/com/facebook/presto/index/IndexHandleJacksonModule.java

@@ -13,19 +13,50 @@
  */
 package com.facebook.presto.index;
 
+import com.facebook.presto.connector.ConnectorManager;
 import com.facebook.presto.metadata.AbstractTypedJacksonModule;
 import com.facebook.presto.metadata.HandleResolver;
+import com.facebook.presto.spi.ConnectorCodec;
+import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.ConnectorIndexHandle;
+import com.facebook.presto.spi.connector.ConnectorCodecProvider;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import jakarta.inject.Inject;
+import jakarta.inject.Provider;
+
+import java.util.Optional;
+import java.util.function.Function;
 
 public class IndexHandleJacksonModule
         extends AbstractTypedJacksonModule<ConnectorIndexHandle>
 {
     @Inject
-    public IndexHandleJacksonModule(HandleResolver handleResolver)
+    public IndexHandleJacksonModule(
+            HandleResolver handleResolver,
+            Provider<ConnectorManager> connectorManagerProvider,
+            FeaturesConfig featuresConfig)
+    {
+        super(ConnectorIndexHandle.class,
+                handleResolver::getId,
+                handleResolver::getIndexHandleClass,
+                featuresConfig.isUseConnectorProvidedSerializationCodecs(),
+                connectorId -> connectorManagerProvider.get()
+                        .getConnectorCodecProvider(connectorId)
+                        .flatMap(ConnectorCodecProvider::getConnectorIndexHandleCodec));
+    }
+
+    /**
+     * Test-friendly constructor that accepts a codec extractor function directly.
+     */
+    public IndexHandleJacksonModule(
+            HandleResolver handleResolver,
+            FeaturesConfig featuresConfig,
+            Function<ConnectorId, Optional<ConnectorCodec<ConnectorIndexHandle>>> codecExtractor)
     {
         super(ConnectorIndexHandle.class,
                 handleResolver::getId,
-                handleResolver::getIndexHandleClass);
+                handleResolver::getIndexHandleClass,
+                featuresConfig.isUseConnectorProvidedSerializationCodecs(),
+                codecExtractor);
     }
 }