JDBC/ODBC vs Arrow Flight SQL:

Traditional JDBC/ODBC:
  Server -> Columnar Store -> Row Serialize -> Network -> Row Deserialize -> Client
  Bottleneck: Serialization/deserialization = 70-90% of CPU

Arrow Flight SQL:
  Server -> Arrow Columnar Format -> Network -> Client (zero-copy)
  Advantage: Data already in columnar format, no serialization needed

Performance (10M rows, 20 columns):

+-----------------+----------+----------+----------+
| Protocol         | Throughput| Latency  | CPU      |
+-----------------+----------+----------+----------+
| JDBC (MySQL)     | 80 MB/s  | High     | High     |
| JDBC (Doris)     | 120 MB/s | Medium   | Medium   |
| Arrow Flight     | 2 GB/s   | Low      | Low      |
| Arrow Flight SQL | 1.5 GB/s | Low      | Low      |
+-----------------+----------+----------+----------+

Arrow Flight SQL Protocol Stack:

+--------------------------------------+
|          Application Layer            |
|  (OQL Query / SQL Query)              |
+--------------------------------------+
|          Flight SQL Layer             |
|  (GetFlightInfo / DoGet / DoPut)      |
+--------------------------------------+
|          Arrow Flight Layer           |
|  (Arrow IPC Format / Streaming)       |
+--------------------------------------+
|          gRPC Transport               |
|  (HTTP/2, multiplexing, flow control) |
+--------------------------------------+
|          TLS / mTLS                   |
|  (Encrypted transport, mutual auth)   |
+--------------------------------------+

Flight SQL Query Execution Flow:

1. Client connects to FE
   FlightClient.connect("grpc://doris-fe:8040")
   FlightClient.authenticate(user, password)

2. Client sends query
   FlightInfo info = client.getFlightInfo(
     FlightDescriptor.command(
       "SELECT * FROM entity_common LIMIT 1000"
     )
   )

3. FE returns FlightInfo
   FlightInfo {
     schema: Arrow Schema,
     endpoints: [
       {ticket: "q-123-be1", locations: ["grpc://be1:8060"]},
       {ticket: "q-123-be2", locations: ["grpc://be2:8060"]},
       {ticket: "q-123-be3", locations: ["grpc://be3:8060"]},
     ]
   }

4. Client fetches data from BEs in parallel
   for endpoint in info.endpoints:
       be_client = FlightClient.connect(endpoint.locations[0])
       stream = be_client.doGet(endpoint.ticket)
       for batch in stream:
           process(batch)  # Arrow RecordBatch, zero-copy

import pyarrow.flight as flight
import pyarrow as pa

class OntoFlightSQLClient:
    """coomia-dip Flight SQL client"""

    def __init__(self, host: str, port: int = 8040):
        self._location = flight.Location.for_grpc_tcp(host, port)
        self._client = flight.FlightClient(self._location)
        self._token: bytes | None = None

    def authenticate(self, username: str, password: str):
        self._token = self._client.authenticate_basic_token(
            username, password
        )

    def execute_query(self, sql: str) -> pa.Table:
        options = flight.FlightCallOptions(
            headers=[(b"authorization", self._token)]
        )

        info = self._client.get_flight_info(
            flight.FlightDescriptor.for_command(sql.encode()),
            options=options
        )

        batches = []
        for endpoint in info.endpoints:
            reader = self._client.do_get(
                endpoint.ticket, options=options
            )
            for batch in reader:
                batches.append(batch.data)

        if not batches:
            return pa.table({})

        return pa.Table.from_batches(batches, schema=info.schema)

    async def execute_streaming(
        self, sql: str
    ) -> AsyncIterator[pa.RecordBatch]:
        options = flight.FlightCallOptions(
            headers=[(b"authorization", self._token)]
        )

        info = self._client.get_flight_info(
            flight.FlightDescriptor.for_command(sql.encode()),
            options=options
        )

        for endpoint in info.endpoints:
            reader = self._client.do_get(
                endpoint.ticket, options=options
            )
            for batch in reader:
                yield batch.data

class OQLFlightExecutor:
    """Execute OQL queries via Flight SQL"""

    def __init__(self, flight_client: OntoFlightSQLClient):
        self._client = flight_client
        self._compiler = OQLToSQLCompiler()

    def execute_oql(self, oql: str) -> pa.Table:
        sql = self._compiler.compile(oql)
        result = self._client.execute_query(sql)
        result = self._post_process(result, oql)
        return result

    def execute_oql_to_pandas(self, oql: str) -> 'pd.DataFrame':
        return self.execute_oql(oql).to_pandas()

    def execute_oql_to_polars(self, oql: str) -> 'pl.DataFrame':
        return pl.from_arrow(self.execute_oql(oql))

class FlightConnectionPool:
    """Flight SQL connection pool"""

    def __init__(
        self, host: str, port: int,
        min_connections: int = 5,
        max_connections: int = 20,
    ):
        self._host = host
        self._port = port
        self._min = min_connections
        self._max = max_connections
        self._pool: asyncio.Queue = asyncio.Queue()
        self._active_count = 0

    async def acquire(self) -> OntoFlightSQLClient:
        try:
            client = self._pool.get_nowait()
            if client.is_healthy():
                return client
            self._active_count -= 1
        except asyncio.QueueEmpty:
            pass

        if self._active_count < self._max:
            client = OntoFlightSQLClient(self._host, self._port)
            client.authenticate(self._username, self._password)
            self._active_count += 1
            return client

        return await self._pool.get()

    async def release(self, client: OntoFlightSQLClient):
        if self._pool.qsize() < self._min:
            await self._pool.put(client)
        else:
            client.close()
            self._active_count -= 1

Flight SQL Security Authentication:

Option 1: Basic Auth + Token
  1. Client sends username/password
  2. Server returns Bearer Token
  3. Subsequent requests carry Token
  For: Internal service communication

Option 2: mTLS (Mutual TLS)
  1. Both client and server have certificates
  2. Mutual verification at connection
  3. Transport layer encryption
  For: Cross-network, high-security scenarios

Option 3: JWT Token
  1. Obtain JWT via OAuth2/OIDC
  2. Flight SQL requests carry JWT
  3. Server validates JWT signature and permissions
  For: Enterprise SSO integration

Flight SQL Performance Tuning Parameters:

+----------------------+----------+------------------+
| Parameter             | Default  | Recommended       |
+----------------------+----------+------------------+
| batch_size             | 4096     | 8192-16384       |
| max_message_size       | 4 MB     | 16 MB            |
| grpc_keepalive_time    | 120s     | 30s              |
| flight_parallelism     | 1        | # of BE nodes    |
| compression            | none     | lz4 / zstd       |
| connection_pool_size   | 5        | 10-20            |
| timeout_seconds        | 30       | 60-300           |
+----------------------+----------+------------------+

Batch size tuning:
  Too small -> high gRPC overhead
  Too large -> high per-batch latency
  Optimal: 8192-16384 (adjust by column count/width)

Compression tuning:
  None: lowest CPU, highest bandwidth
  LZ4: low CPU, moderate compression (recommended)
  ZSTD: moderate CPU, high compression (bandwidth-limited)

class TestFlightSQL:

    def test_basic_query(self):
        client = OntoFlightSQLClient("localhost", 8040)
        client.authenticate("admin", "password")
        result = client.execute_query("SELECT 1 AS n")
        assert result.num_rows == 1
        assert result.column('n')[0].as_py() == 1

    def test_large_result_set(self):
        client = OntoFlightSQLClient("localhost", 8040)
        client.authenticate("admin", "password")
        result = client.execute_query(
            "SELECT * FROM entity_common LIMIT 1000000"
        )
        assert result.num_rows == 1000000

    def test_parallel_endpoint_fetch(self):
        info = client.get_flight_info(
            "SELECT * FROM entity_common"
        )
        assert len(info.endpoints) >= 1

    def test_oql_via_flight_sql(self):
        executor = OQLFlightExecutor(client)
        result = executor.execute_oql(
            "FETCH Person WHERE age > 30 LIMIT 100"
        )
        assert result.num_rows <= 100

    async def test_connection_pool(self):
        pool = FlightConnectionPool("localhost", 8040)
        clients = await asyncio.gather(*[
            pool.acquire() for _ in range(10)
        ])
        assert len(clients) == 10
        for c in clients:
            await pool.release(c)

    def test_throughput_benchmark(self):
        start = time.time()
        result = client.execute_query(
            "SELECT * FROM entity_common LIMIT 10000000"
        )
        elapsed = time.time() - start
        throughput_mbps = result.nbytes / elapsed / 1024 / 1024
        assert throughput_mbps > 500  # > 500 MB/s