feat: Add comprehensive Pinot Proxy and gRPC support to pinot-spark-3-connector

🔧 Pinot Proxy Support:
- Add proxy.enabled configuration option (default: false)
- Implement HTTP proxy forwarding with FORWARD_HOST and FORWARD_PORT headers
- Support proxy routing for all controller and broker API requests
- Enable proxy-based secure cluster access where proxy is the only exposed endpoint

🚀 Comprehensive gRPC Configuration:
- Add grpc.port configuration (default: 8090)
- Add grpc.max-inbound-message-size configuration (default: 128MB)
- Add grpc.use-plain-text configuration (default: true)
- Support grpc.tls.keystore-type, grpc.tls.keystore-path, grpc.tls.keystore-password
- Support grpc.tls.truststore-type, grpc.tls.truststore-path, grpc.tls.truststore-password
- Add grpc.tls.ssl-provider configuration (default: JDK)
- Add grpc.proxy-uri for gRPC proxy endpoint configuration

🔒 gRPC Proxy Integration:
- Implement gRPC proxy support with FORWARD_HOST and FORWARD_PORT metadata
- Create comprehensive GrpcUtils for channel management and proxy headers
- Support secure gRPC communication through proxy infrastructure
- Enable TLS/SSL configuration for gRPC connections

🏗️ Architecture Updates:
- Update PinotDataSourceReadOptions with all new proxy and gRPC fields
- Enhance PinotClusterClient with proxy-aware API methods
- Add HttpUtils.sendGetRequestWithProxyHeaders() for proxy HTTP requests
- Update PinotServerDataFetcher with gRPC proxy configuration support
- Modify all Spark DataSource V2 components to pass proxy parameters

🧪 Comprehensive Testing:
- Add 8 new test cases for proxy and gRPC configuration parsing
- Create GrpcUtilsTest for gRPC channel creation and proxy metadata
- Update existing tests to include new configuration parameters
- Achieve 39/39 passing tests with full backward compatibility

📚 Enhanced Documentation:
- Add comprehensive Pinot Proxy Support section with examples
- Add detailed gRPC Configuration section with TLS examples
- Include Security Best Practices for production deployments
- Provide proxy + gRPC + HTTPS + authentication integration examples

🎯 Production Features:
- Full backward compatibility - existing code works unchanged
- Based on Trino PR #13015 reference implementation
- Supports secure production deployments with proxy-only access
- Comprehensive error handling and validation
- Performance optimizations for gRPC connections

All tests passing (39/39) with complete feature parity to Trino's implementation.

Author: xiangfu0

pinot-connectors/pinot-spark-3-connector/README.md

@@ -60,9 +60,44 @@ val data = spark.read
 data.show(100)
 ```
 
-## HTTPS Configuration
+## Security Configuration
 
-The connector supports HTTPS/TLS connections to Pinot clusters. To enable HTTPS, use the following configuration options:
+You can secure both HTTP and gRPC using a unified switch or explicit flags.
+
+- Unified: set `secureMode=true` to enable HTTPS and gRPC TLS together (recommended)
+- Explicit: set `useHttps` for REST and `grpc.use-plain-text=false` for gRPC
+
+### Quick examples
+
+```scala
+// Unified secure mode (enables HTTPS + gRPC TLS by default)
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("secureMode", "true")
+  .load()
+
+// Explicit HTTPS only (gRPC remains plaintext by default)
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("useHttps", "true")
+  .load()
+
+// Explicit gRPC TLS only (REST remains HTTP by default)
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("grpc.use-plain-text", "false")
+  .load()
+```
+
+### HTTPS Configuration
+
+When HTTPS is enabled (either via `secureMode=true` or `useHttps=true`), you can configure keystore/truststore as needed:
 
 ```scala
 val data = spark.read
@@ -81,7 +116,8 @@ val data = spark.read
 
 | Option | Description | Required | Default |
 |--------|-------------|----------|---------|
-| `useHttps` | Enable HTTPS connections | No | `false` |
+| `secureMode` | Unified switch to enable HTTPS and gRPC TLS | No | `false` |
+| `useHttps` | Enable HTTPS connections (overrides `secureMode` for REST) | No | `false` |
 | `keystorePath` | Path to client keystore file (JKS format) | No | None |
 | `keystorePassword` | Password for the keystore | No | None |
 | `truststorePath` | Path to truststore file (JKS format) | No | None |
@@ -130,6 +166,107 @@ val data = spark.read
 
 **Note:** If only `authToken` is provided without `authHeader`, the connector will automatically use `Authorization: Bearer <token>`.
 
+## Pinot Proxy Support
+
+The connector supports Pinot Proxy for secure cluster access where the proxy is the only exposed endpoint. When proxy is enabled, all HTTP requests to controllers/brokers and gRPC requests to servers are routed through the proxy.
+
+### Proxy Configuration Examples
+
+```scala
+// Basic proxy configuration
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("controller", "pinot-proxy:8080")  // Proxy endpoint
+  .option("proxy.enabled", "true")
+  .load()
+
+// Proxy with authentication
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("controller", "pinot-proxy:8080")
+  .option("proxy.enabled", "true")
+  .option("authToken", "my-proxy-token")
+  .load()
+
+// Proxy with gRPC configuration
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("controller", "pinot-proxy:8080")
+  .option("proxy.enabled", "true")
+  .option("grpc.proxy-uri", "pinot-proxy:8094")  // gRPC proxy endpoint
+  .load()
+```
+
+### Proxy Configuration Options
+
+| Option | Description | Required | Default |
+|--------|-------------|----------|----------|
+| `proxy.enabled` | Use Pinot Proxy for controller and broker requests | No | `false` |
+
+**Note:** When proxy is enabled, the connector adds `FORWARD_HOST` and `FORWARD_PORT` headers to route requests to the actual Pinot services.
+
+## gRPC Configuration
+
+The connector supports comprehensive gRPC configuration for secure and optimized communication with Pinot servers.
+
+### gRPC Configuration Examples
+
+```scala
+// Basic gRPC configuration
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("grpc.port", "8091")
+  .option("grpc.max-inbound-message-size", "256000000")  // 256MB
+  .load()
+
+// gRPC with TLS (explicit)
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("grpc.use-plain-text", "false")
+  .option("grpc.tls.keystore-path", "/path/to/grpc-keystore.jks")
+  .option("grpc.tls.keystore-password", "keystore-password")
+  .option("grpc.tls.truststore-path", "/path/to/grpc-truststore.jks")
+  .option("grpc.tls.truststore-password", "truststore-password")
+  .load()
+
+// gRPC with proxy
+val data = spark.read
+  .format("pinot")
+  .option("table", "airlineStats")
+  .option("tableType", "offline")
+  .option("proxy.enabled", "true")
+  .option("grpc.proxy-uri", "pinot-proxy:8094")
+  .load()
+```
+
+### gRPC Configuration Options
+
+| Option | Description | Required | Default |
+|--------|-------------|----------|----------|
+| `grpc.port` | Pinot gRPC port | No | `8090` |
+| `grpc.max-inbound-message-size` | Max inbound message bytes when init gRPC client | No | `128MB` |
+| `grpc.use-plain-text` | Use plain text for gRPC communication (overrides `secureMode` for gRPC) | No | `true` |
+| `grpc.tls.keystore-type` | TLS keystore type for gRPC connection | No | `JKS` |
+| `grpc.tls.keystore-path` | TLS keystore file location for gRPC connection | No | None |
+| `grpc.tls.keystore-password` | TLS keystore password | No | None |
+| `grpc.tls.truststore-type` | TLS truststore type for gRPC connection | No | `JKS` |
+| `grpc.tls.truststore-path` | TLS truststore file location for gRPC connection | No | None |
+| `grpc.tls.truststore-password` | TLS truststore password | No | None |
+| `grpc.tls.ssl-provider` | SSL provider | No | `JDK` |
+| `grpc.proxy-uri` | Pinot Rest Proxy gRPC endpoint URI | No | None |
+
+**Note:** When using gRPC with proxy, the connector automatically adds `FORWARD_HOST` and `FORWARD_PORT` metadata headers for proper request routing.
+
 ## Examples
 
 There are more examples included in `src/test/scala/.../ExampleSparkPinotConnectorTest.scala`.
@@ -152,6 +289,32 @@ Spark-Pinot connector uses Spark `DatasourceV2 API`. Please check the Databricks
 
 https://www.slideshare.net/databricks/apache-spark-data-source-v2-with-wenchen-fan-and-gengliang-wang
 
+## Security Best Practices
+
+### Production HTTPS Configuration
+- Always use HTTPS in production environments
+- Store certificates in secure locations with appropriate file permissions
+- Use proper certificate validation with valid truststore
+- Rotate certificates regularly
+
+### Production Authentication
+- Use service accounts with minimal required permissions
+- Store authentication tokens securely (environment variables, secret management systems)
+- Implement token rotation policies
+- Monitor authentication failures
+
+### Production gRPC Configuration
+- Enable TLS for gRPC communication in production
+- Use certificate-based authentication when possible
+- Configure appropriate message size limits based on your data
+- Use connection pooling for high-throughput scenarios
+
+### Production Proxy Configuration
+- Ensure proxy endpoints are properly secured
+- Monitor proxy health and performance
+- Implement proper request routing and load balancing
+- Use authentication for proxy access
+
 ## Future Works
 - Add integration tests for read operation
 - Add write support(pinot segment write logic will be changed in later versions of pinot)

pinot-connectors/pinot-spark-3-connector/documentation/read_model.md

@@ -132,11 +132,14 @@ val df = spark.read
 |-----------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| ------------- |-------------------------------------------------------|
 | table                 | Pinot table name without table type                                                                                                                                                                           | Yes | -                                                     |
 | tableType             | Pinot table type(`realtime`, `offline` or `hybrid`)                                                                                                                                                           | Yes | -                                                     |
-| controller            | Pinot controller url and port. Input should be `url:port` format without schema. Connector does not support `https` schema for now.                                                                           | No | localhost:9000                                        |
-| broker                | Pinot broker url and port. Input should be `url:port` format without schema. If not specified, connector will find broker instances of table automatically. Connector does not support `https` schema for now | No | Fetch broker instances of table from Pinot Controller | 
+| controller            | Pinot controller `host:port` (schema inferred from `useHttps`/`secureMode`)                                                                                             | No | localhost:9000                                        |
+| broker                | Pinot broker `host:port` (schema inferred from `useHttps`/`secureMode`)                                                                                                 | No | Fetch broker instances of table from Pinot Controller | 
 | usePushDownFilters    | Push filters to pinot servers or not. If true, data exchange between pinot server and spark will be minimized.                                                                                                | No | true                                                  |
 | segmentsPerSplit      | Represents the maximum segment count that will be scanned by pinot server in one connection                                                                                                                   | No | 3                                                     | 
 | pinotServerTimeoutMs  | The maximum timeout(ms) to get data from pinot server                                                                                                                                                         | No | 10 mins                                               |
 | useGrpcServer         | Boolean value to enable reads via gRPC. This option is more memory efficient both on Pinot server and Spark executor side because it utilizes streaming. Requires gRPC to be enabled on Pinot server.         | No | false                                                 |
 | queryOptions          | Comma separated list of Pinot query options (e.g. "enableNullHandling=true,skipUpsert=true")                                                                                                                  | No | ""                                                    |
 | failOnInvalidSegments | Fail the read operation if response metadata indicates invalid segments                                                                                                                                       | No | false                                                 |
+| secureMode            | Unified switch to enable HTTPS and gRPC TLS (explicit `useHttps`/`grpc.use-plain-text` take precedence)                                                                  | No | false                                                |
+| useHttps              | Enable HTTPS for REST calls (overrides `secureMode` for REST)                                                                                                            | No | false                                                |
+| grpc.use-plain-text   | Use plaintext for gRPC (overrides `secureMode` for gRPC)                                                                                                                | No | true                                                 |

pinot-connectors/pinot-spark-3-connector/examples/README.md

@@ -0,0 +1,55 @@
+<!--
+
+    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.
+
+-->
+# Pinot Spark 3 Connector Example: Reading from Pinot via Proxy with Auth Token
+
+This example demonstrates how to use the Pinot Spark 3 Connector to read data from a Pinot cluster via a proxy with authentication token support.
+
+## Prerequisites
+
+- Apache Spark 3.x installed and `spark-shell` available in your PATH.
+
+- Setup PINOT_HOME env variable:
+  ```
+  export PINOT_HOME=/path/to/pinot
+  ```
+
+- The Pinot Spark 3 Connector shaded JAR built and available at:
+  ```
+  $PINOT_HOME/pinot-connectors/pinot-spark-3-connector/target/pinot-spark-3-connector-*-shaded.jar
+  ```
+
+- Example Scala script located at:
+  ```
+  $PINOT_HOME/pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala
+  ```
+
+## How to Run
+
+Launch the example in `spark-shell` with the following command:
+
+```bash
+spark-shell --master 'local[*]' --name read-pinot --jars "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/target/pinot-spark-3-connector-*-shaded.jar" < "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala"
+```
+
+
+
+
+

pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala

@@ -0,0 +1,44 @@
+/**
+ * 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.
+ */
+import org.apache.spark.sql.SparkSession
+
+val spark = SparkSession.builder().appName("read-pinot-airlineStats").master("local[*]").getOrCreate()
+
+val df = spark.read.
+  format("org.apache.pinot.connector.spark.v3.datasource.PinotDataSource").
+  option("table", "myTable").
+  option("tableType", "offline").
+  option("controller", "pinot-controller").
+  option("broker", "pinot-broker").
+  option("secureMode", "true").
+  option("authToken", "st-xxxxxxx").
+  option("proxy.enabled", "true").
+  option("grpc.proxy-uri", "grpc-pinot-proxy").
+  option("useGrpcServer", "true").
+  load()
+
+println("Schema:")
+df.printSchema()
+
+println("Sample rows:")
+df.show(10, truncate = false)
+
+println(s"Total rows: ${df.count()}")
+
+spark.stop()

pinot-connectors/pinot-spark-3-connector/src/main/scala/org/apache/pinot/connector/spark/v3/datasource/DataExtractor.scala

@@ -54,10 +54,12 @@ private[pinot] object DataExtractor {
       case FieldSpec.DataType.LONG => LongType
       case FieldSpec.DataType.FLOAT => FloatType
       case FieldSpec.DataType.DOUBLE => DoubleType
+      case FieldSpec.DataType.BIG_DECIMAL => DecimalType(38, 18)
       case FieldSpec.DataType.STRING => StringType
       case FieldSpec.DataType.BYTES => ArrayType(ByteType)
       case FieldSpec.DataType.TIMESTAMP => LongType
       case FieldSpec.DataType.BOOLEAN => BooleanType
+      case FieldSpec.DataType.JSON => StringType
       case _ =>
         throw PinotException(s"Unsupported pinot data type '$dataType")
     }
@@ -116,6 +118,21 @@ private[pinot] object DataExtractor {
       dataTable.getFloat(rowIndex, colIndex)
     case ColumnDataType.DOUBLE =>
       dataTable.getDouble(rowIndex, colIndex)
+    case ColumnDataType.BIG_DECIMAL =>
+      val bd = dataTable.getBigDecimal(rowIndex, colIndex)
+      if (bd == null) null
+      else {
+        // Derive precision/scale from actual value; clamp to Spark's max precision
+        val precision = math.min(DecimalType.MAX_PRECISION, bd.precision())
+        val scale = math.min(DecimalType.MAX_SCALE, math.max(0, bd.scale()))
+        val scaled =
+          if (bd.scale() == scale) bd
+          else bd.setScale(scale, java.math.RoundingMode.HALF_UP)
+        Decimal(scaled, precision, scale)
+      }
+    case ColumnDataType.JSON =>
+      // Use underlying string
+      UTF8String.fromString(dataTable.getString(rowIndex, colIndex))
     case ColumnDataType.TIMESTAMP =>
       dataTable.getLong(rowIndex, colIndex)
     case ColumnDataType.BOOLEAN =>

pinot-connectors/pinot-spark-3-connector/src/main/scala/org/apache/pinot/connector/spark/v3/datasource/PinotDataSource.scala

@@ -40,7 +40,7 @@ class PinotDataSource extends TableProvider with DataSourceRegister {
     val controller = readParameters.controller
 
     val pinotTableSchema =
-      PinotClusterClient.getTableSchema(controller, tableName, readParameters.useHttps, readParameters.authHeader, readParameters.authToken)
+      PinotClusterClient.getTableSchema(controller, tableName, readParameters.useHttps, readParameters.authHeader, readParameters.authToken, readParameters.proxyEnabled)
     DataExtractor.pinotSchemaToSparkSchema(pinotTableSchema)
   }
 

pinot-connectors/pinot-spark-3-connector/src/main/scala/org/apache/pinot/connector/spark/v3/datasource/PinotScan.scala

@@ -54,13 +54,13 @@ class PinotScan(
   override def toBatch: Batch = this
 
   override def planInputPartitions(): Array[InputPartition] = {
-    val routingTable = PinotClusterClient.getRoutingTable(readParameters.broker, query, readParameters.useHttps, readParameters.authHeader, readParameters.authToken)
+    val routingTable = PinotClusterClient.getRoutingTable(readParameters.broker, query, readParameters.useHttps, readParameters.authHeader, readParameters.authToken, readParameters.proxyEnabled)
 
     val instanceInfo : Map[String, InstanceInfo] = Map()
     val instanceInfoReader = (instance:String) => { // cached reader to reduce network round trips
       instanceInfo.getOrElseUpdate(
         instance,
-        PinotClusterClient.getInstanceInfo(readParameters.controller, instance, readParameters.useHttps, readParameters.authHeader, readParameters.authToken)
+        PinotClusterClient.getInstanceInfo(readParameters.controller, instance, readParameters.useHttps, readParameters.authHeader, readParameters.authToken, readParameters.proxyEnabled)
       )
     }
 

pinot-connectors/pinot-spark-3-connector/src/main/scala/org/apache/pinot/connector/spark/v3/datasource/PinotScanBuilder.scala

@@ -45,7 +45,7 @@ class PinotScanBuilder(readParameters: PinotDataSourceReadOptions)
       if (readParameters.tableType.isDefined) {
         None
       } else {
-        PinotClusterClient.getTimeBoundaryInfo(readParameters.broker, readParameters.tableName, readParameters.useHttps, readParameters.authHeader, readParameters.authToken)
+        PinotClusterClient.getTimeBoundaryInfo(readParameters.broker, readParameters.tableName, readParameters.useHttps, readParameters.authHeader, readParameters.authToken, readParameters.proxyEnabled)
       }
 
     val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)

pinot-connectors/pinot-spark-3-connector/src/main/scala/org/apache/pinot/connector/spark/v3/datasource/SparkToPinotTypeTranslator.scala

@@ -58,6 +58,7 @@ object SparkToPinotTypeTranslator {
     case _: LongType => FieldSpec.DataType.LONG
     case _: FloatType => FieldSpec.DataType.FLOAT
     case _: DoubleType => FieldSpec.DataType.DOUBLE
+    case _: DecimalType => FieldSpec.DataType.BIG_DECIMAL
     case _: BooleanType => FieldSpec.DataType.BOOLEAN
     case _: BinaryType => FieldSpec.DataType.BYTES
     case _: TimestampType => FieldSpec.DataType.LONG