Add RFC-0013 for Java connector federation in C++ workers

Author: tdcmeehan

RFC-0013-java-connector-federaton.md

@@ -0,0 +1,327 @@
+# **RFC-0013 for Java Connector Federation in C++ Workers**
+
+## Java Connector Federation via Arrow Flight
+
+Proposers
+
+* Tim Meehan
+* Bryan Cutler
+* Pratik Dabre
+
+## Related Issues
+
+* [PR #25242](https://github.com/prestodb/presto/pull/25242): Custom Serialization Framework (ConnectorCodecProvider)
+* Arrow Flight Connector in presto-native-execution
+
+## Summary
+
+Enable C++ workers to execute queries using existing single-connection Java connector implementations through Arrow Flight RPC, providing a migration path for connectors not yet ported to C++.
+
+## Background
+
+Presto's C++ workers provide performance improvements but require connectors to be rewritten in C++. Many connectors (JDBC, MongoDB, Elasticsearch) are single-connection systems where the data transfer, not computation, is the bottleneck. Rewriting these connectors in C++ does not improve performance, yet they require a complete reimplementation.
+
+The recent addition of `ConnectorCodecProvider` ([PR #25242](https://github.com/prestodb/presto/pull/25242)) enables custom serialization of connector data structures. Combined with the existing Arrow Flight connector in C++, this provides a path to federation.
+
+### Goals
+
+* Enable single-connection Java connectors to work with C++ workers without code duplication
+* Optimize for connectors where all splits share a single database connection
+
+### Non-goals
+
+* Multi-split connectors (Hive, Iceberg) where parallelism and C++ execution improve performance
+* Modifying the coordinator-worker protocol
+
+## Proposed Implementation
+
+### 1. Architecture Overview
+
+```mermaid
+graph LR
+    subgraph "Coordinator (Java)"
+        A[JDBC Connector<br/>generates split]
+    end
+    
+    subgraph "C++ Worker"
+        B[Arrow Flight<br/>Connector<br/>forwards ticket]
+    end
+    
+    subgraph "Flight Server (Java)"
+        C[JDBC Connector<br/>execute split]
+    end
+    
+    A -->|ArrowSplit<br/>with ticket| B
+    B -->|gRPC/Flight<br/>protocol| C
+    C -->|Arrow<br/>RecordBatches| B
+```
+
+### 2. Module Changes
+
+#### presto-connector-flight-common (new)
+Common library providing Arrow Flight wrapping (example pseudocode):
+
+```java
+// One-line integration for any connector
+public class ArrowFlightCodecProvider implements ConnectorCodecProvider {
+    public static ConnectorCodecProvider wrap(
+            ConnectorCodecProvider original,
+            ArrowFlightFederationConfig config) {
+        if (!config.isEnabled()) {
+            return original;
+        }
+        return new ArrowFlightCodecProvider(original, config);
+    }
+
+    public Optional<ConnectorCodec<ConnectorSplit>> getConnectorSplitCodec() {
+        return Optional.of(new ArrowFlightWrapperCodec(
+                originalCodec,
+                connectorId));
+    }
+}
+
+public class ArrowFlightWrapperCodec implements ConnectorCodec<ConnectorSplit> {
+    private final String connectorId;
+
+    public byte[] serialize(ConnectorSplit split) {
+        byte[] originalBytes = originalCodec.serialize(split);
+
+        // Create ticket with connector ID prefix to identify the connector
+        ByteBuffer ticketBuffer = ByteBuffer.allocate(
+                4 + connectorId.length() + originalBytes.length);
+        ticketBuffer.putInt(connectorId.length());
+        ticketBuffer.put(connectorId.getBytes(UTF_8));
+        ticketBuffer.put(originalBytes);
+
+        // No locations added - C++ worker will use locally configured Flight server
+        FlightEndpoint.Builder endpoint = FlightEndpoint.newBuilder()
+                .setTicket(FlightTicket.newBuilder()
+                        .setTicket(ByteString.copyFrom(ticketBuffer.array())));
+
+        ArrowSplit arrowSplit = new ArrowSplit();
+        arrowSplit.setFlightEndpointBytes(Base64.encode(endpoint.build().toByteArray()));
+        return arrowFlightCodec.serialize(arrowSplit);
+    }
+}
+
+```
+
+#### Integration for any single-connection Connector
+```java
+// JDBC Connector
+public class JdbcConnector implements Connector {
+    @Inject ArrowFlightFederationConfig flightConfig;
+
+    @Override
+    public ConnectorCodecProvider getConnectorCodecProvider() {
+        // Enable federation
+        return ArrowFlightCodecProvider.wrap(
+                getDefaultJdbcCodecProvider(),
+                flightConfig);
+    }
+}
+
+// MongoDB Connector - identical pattern
+public class MongoConnector implements Connector {
+    @Inject ArrowFlightFederationConfig flightConfig;
+
+    @Override
+    public ConnectorCodecProvider getConnectorCodecProvider() {
+        return ArrowFlightCodecProvider.wrap(
+                getDefaultMongoCodecProvider(),
+                flightConfig);
+    }
+}
+```
+
+#### presto-native-execution changes
+
+##### Custom Serialization Support
+
+Override the `deserialize` method to parse the binary format that matches the Java ArrowSplit codec's `serialize` method:
+
+```cpp
+class ArrowConnectorProtocol : public ConnectorProtocolTemplate<
+    ArrowTableHandle,
+    ArrowTableLayoutHandle,
+    ArrowColumnHandle,
+    NotImplemented,
+    NotImplemented,
+    ArrowSplit,
+    NotImplemented,
+    ArrowTransactionHandle,
+    NotImplemented,
+    NotImplemented> {
+public:
+  void deserialize(
+      const std::string& thrift,
+      std::shared_ptr<ConnectorSplit>& proto) const override {
+    // Parse binary format matching Java's ArrowSplit codec serialize method
+    auto arrowSplit = std::make_shared<ArrowSplit>();
+    // Binary format: [schemaName_len][schemaName][tableName_len][tableName][flightEndpointBytes]
+    // Implementation details...
+    proto = arrowSplit;
+  }
+};
+```
+
+The deserialization path in `ProtocolToThrift.cpp` already calls `getConnectorProtocol(connectorId).deserialize()`.
+
+The existing Arrow Flight connector already uses the locally configured `arrow-flight.server` and `arrow-flight.server.port` properties. When no locations are provided in the FlightEndpoint, it automatically falls back to these configured values.
+
+##### Colocated Flight Server Process Management
+
+When `arrow-flight.colocated=true`, the C++ worker manages a Flight server subprocess:
+
+- Automatically configures `arrow-flight.server=localhost` and uses `arrow-flight.colocated.port`
+- Spawns a Java subprocess running `presto-flight-server` with JVM options from `${etc.dir}/jvm.config`
+- Restarts the Flight server if it crashes
+- Terminates the subprocess on worker shutdown
+
+#### presto-flight-server (new)
+Flight server that loads connectors using Presto's plugin loading approach, adapted from the main Presto server implementation.
+
+The Flight server uses standard Presto configuration structure:
+- `jvm.config`: JVM heap size and flags (e.g., `-Xmx4G -XX:+UseG1GC`)
+- `config.properties`: Flight server configuration including:
+  - `catalog.config-dir`: Path to catalog configurations (default: `etc/catalog`)
+  - `plugin.dir`: Path to plugin directory
+  - Other Flight server-specific properties
+- Catalog directory (as specified by `catalog.config-dir`): Connector configurations (jdbc.properties, mongodb.properties, etc.)
+
+```java
+public class FlightServerPluginManager {
+    private final Map<String, ConnectorFactory> connectorFactories = new ConcurrentHashMap<>();
+    private final Map<String, Connector> connectors = new ConcurrentHashMap<>();
+
+    public void loadPlugins() throws Exception {
+        // Use same approach as PluginManager to load from installed plugins dir
+        for (File pluginDir : listFiles(installedPluginsDir)) {
+            loadPlugin(pluginDir);
+        }
+    }
+
+    private void loadPlugin(File pluginDir) {
+        URLClassLoader pluginClassLoader = buildClassLoader(pluginDir);
+        ServiceLoader<Plugin> serviceLoader = ServiceLoader.load(Plugin.class, pluginClassLoader);
+
+        for (Plugin plugin : serviceLoader) {
+            for (ConnectorFactory factory : plugin.getConnectorFactories()) {
+                log.info("Registering connector %s", factory.getName());
+                connectorFactories.put(factory.getName(), factory);
+            }
+        }
+    }
+
+    public Connector getConnector(String connectorName) {
+        // Create connector instances from factories as needed
+        return connectors.computeIfAbsent(connectorName, name -> {
+            ConnectorFactory factory = connectorFactories.get(name);
+            return factory.create(name, config, connectorContext);
+        });
+    }
+}
+
+public class ConnectorFlightProducer extends FlightProducer {
+    private final FlightServerPluginManager pluginManager;
+
+    @Override
+    public void getStream(CallContext context, Ticket ticket,
+            ServerStreamListener listener) {
+        // Ticket format: [connectorId_length][connectorId][split_bytes]
+        ByteBuffer buffer = ByteBuffer.wrap(ticket.getBytes());
+        int connectorIdLength = buffer.getInt();
+        byte[] connectorIdBytes = new byte[connectorIdLength];
+        buffer.get(connectorIdBytes);
+        String connectorId = new String(connectorIdBytes, UTF_8);
+
+        // Get connector from plugin manager
+        Connector connector = pluginManager.getConnector(connectorId);
+        ConnectorCodec<ConnectorSplit> codec = connector.getConnectorCodecProvider()
+                .getConnectorSplitCodec()
+                .orElse(defaultJsonCodec);
+
+        // Deserialize and execute
+        byte[] splitBytes = new byte[buffer.remaining()];
+        buffer.get(splitBytes);
+        ConnectorSplit split = codec.deserialize(splitBytes);
+
+        ConnectorPageSource pageSource = connector.getPageSource(split);
+        streamPagesAsArrow(pageSource, listener);
+    }
+}
+```
+
+### 3. Configuration
+
+#### Authentication
+
+Arrow Flight supports pluggable authentication. For the initial implementation, authentication is not required for colocated deployments since the Flight server only accepts localhost connections. Remote deployments requiring authentication would need to implement custom authenticators based on their security requirements.
+
+#### Colocated Deployment Mode
+
+```properties
+# C++ worker configuration
+arrow-flight.colocated=true
+arrow-flight.colocated.port=8815  # Port for the colocated Flight server
+# arrow-flight.server is automatically set to localhost
+# Setting arrow-flight.server is an error when colocated=true
+
+# Flight server configuration directory
+arrow-flight.federation.etc.dir=/path/to/flight-server/etc
+```
+
+When `arrow-flight.federation.etc.dir` is configured, it points to a directory containing the configuration files for the presto-flight-server process (see presto-flight-server section above for details).
+
+#### Remote Deployment Mode
+
+```properties
+# C++ worker configuration
+arrow-flight.colocated=false  # Default
+arrow-flight.server=flight-server.internal  # Should be set for federation
+arrow-flight.server.port=8815  # Should be set for federation
+
+# Custom authenticator configuration would be added here
+# based on the authenticator implementation
+```
+
+When `arrow-flight.colocated=true`, setting `arrow-flight.server` throws a configuration error since the server is automatically set to localhost and the port is configured via `arrow-flight.colocated.port`. For this federation architecture to work when `colocated=false`, both `arrow-flight.server` and `arrow-flight.server.port` should be configured since the FlightEndpoints sent by the coordinator do not include locations.
+
+### 4. Protocol Flow
+
+1. Coordinator's JDBC connector creates `JdbcSplit`
+2. `ArrowFlightWrapperCodec` serializes as `ArrowSplit` with embedded ticket (no locations)
+3. C++ worker receives `ArrowSplit` via custom serialization protocol
+4. Arrow Flight connector extracts ticket, sends to locally configured Flight server
+5. Flight server deserializes ticket as `JdbcSplit`, executes query
+6. Results stream back as Arrow RecordBatches
+
+## Metrics
+
+* Flight protocol overhead (latency comparison between C++ worker and Flight server metrics)
+* Data transfer throughput between C++ worker and Flight server
+
+## Other Approaches Considered
+
+**JNI Integration**: Direct JVM embedding in C++ workers. Rejected due to memory management and debugging requirements.
+
+**Custom Protocol**: Thrift-based connector protocol. Arrow Flight chosen instead for columnar-native transport, existing ecosystem support, and potential for future cross-language connector implementations.
+
+## Adoption Plan
+
+* New configuration properties:
+  - Per-connector: `arrow-flight-federation.enabled`
+  - Per C++ worker: 
+    - `arrow-flight.colocated`: Whether to spawn local Flight server subprocess
+    - `arrow-flight.colocated.port`: Port for colocated Flight server (used when colocated=true)
+    - `arrow-flight.server` and `arrow-flight.server.port`: Should be set when colocated=false for federation to work
+    - `arrow-flight.federation.etc.dir`: Configuration directory for colocated Flight server
+
+* Single-connection connectors can be migrated individually
+* Documentation: Update each connector's documentation as it gains Flight support
+
+## Test Plan
+
+* Unit tests for split serialization
+* Integration tests for end-to-end query execution through Flight server
+* Tests for both colocated and remote Flight server configurations