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

Author: tdcmeehan

RFC-0013-java-connector-federaton.md

@@ -0,0 +1,325 @@
+# **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:
+
+1. **On startup**:
+   - 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
+
+2. **Flight server subprocess**:
+   - Uses standard Presto configuration structure from the etc directory:
+     - `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.)
+   - Binds to the port specified by `arrow-flight.colocated.port` on localhost
+
+#### presto-flight-server (new)
+Flight server that loads connectors via standard Presto plugin mechanism:
+
+```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 mechanism 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 ConnectorFlightServer 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
+
+#### 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
+```
+
+The Flight server etc directory contains standard Presto configuration files:
+- `jvm.config`: JVM options for the Flight server process
+- `config.properties`: Flight server configuration (catalog.config-dir, plugin.dir, etc.)
+- Catalog directory: Connector configurations (jdbc.properties, mongodb.properties, etc.)
+
+#### 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
+```
+
+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
+
+* 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