Add StarRocks Arrow Flight SQL plugin with direct backend fetch (#120)

* Add StarRocks Arrow Flight SQL plugin with direct backend fetch

Connects to StarRocks via Arrow Flight SQL and, when the FE returns
distinct per-backend endpoint locations, redeems each ticket by
dialing the BE directly and fanning the fetches out in parallel
instead of funneling every row back through the FE.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Fix Wiz findings: bump vulnerable grpc dependency, pin minimum TLS version

- google.golang.org/grpc was pulled in below v1.79.3, affected by CVE-2026-33186
- dialFlightClient's TLS config lacked an explicit MinVersion (CWE-327)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Bump transitive deps flagged by Wiz (otel, x/crypto, x/net)

go.opentelemetry.io/otel < 1.41.0 (CVE-2026-29181), golang.org/x/crypto
< 0.52.0 (13 CVEs), and golang.org/x/net < 0.55.0 (CVE-2026-25680,
CVE-2026-39821) were pulled in transitively by the grpc bump. The
crypto fix version requires go >= 1.25, hence the go.mod bump;
GOTOOLCHAIN=auto (unpinned in this repo) lets CI's Go 1.24.5 fetch
1.25 automatically rather than needing a workflow change.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Version bump

* Split starrocks.go into one file per struct for readability

command_context.go (commandContext + plugin.Handler methods),
cluster_context.go (clusterContext, no methods), job_context.go
(jobContext + collectResults), backend_client_pool.go
(backendClientPool + fetchEndpoint/locationHost). starrocks.go keeps
just the shared consts/telemetry vars and dialFlightClient, used by
both command_context.go and backend_client_pool.go.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Fold commandContext and clusterContext back into starrocks.go

Keep job_context.go and backend_client_pool.go split out, but
commandContext/clusterContext (plugin.Handler entry points +
cluster config) stay in the main file.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: nitinbhakar

.github/workflows/build.yml

@@ -12,7 +12,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v5
         with:
-          go-version: 1.24.5
+          go-version: 1.25.0
           # setup-go caches the module + build cache keyed on go.sum by default.
       - name: Cache pnpm store
         uses: actions/cache@v4

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.24.6 AS go-builder
+FROM golang:1.25.0 AS go-builder
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
       build-essential \

go.mod

@@ -1,10 +1,11 @@
 module github.com/patterninc/heimdall
 
-go 1.24.6
+go 1.25.0
 
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.40.3
 	github.com/antlr4-go/antlr/v4 v4.13.1
+	github.com/apache/arrow-go/v18 v18.4.0
 	github.com/aws/aws-sdk-go-v2 v1.42.0
 	github.com/aws/aws-sdk-go-v2/config v1.30.3
 	github.com/aws/aws-sdk-go-v2/credentials v1.18.3
@@ -30,6 +31,8 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/snowflakedb/gosnowflake v1.15.0
 	github.com/stretchr/testify v1.11.1
+	golang.org/x/sync v0.20.0
+	google.golang.org/grpc v1.79.3
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/api v0.33.4
 	k8s.io/apimachinery v0.33.4
@@ -46,7 +49,6 @@ require (
 	github.com/BurntSushi/toml v1.5.0 // indirect
 	github.com/ClickHouse/ch-go v0.68.0 // indirect
 	github.com/andybalholm/brotli v1.2.0 // indirect
-	github.com/apache/arrow-go/v18 v18.4.0 // indirect
 	github.com/apache/thrift v0.22.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.2 // indirect
@@ -111,23 +113,24 @@ require (
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
-	go.opentelemetry.io/otel v1.38.0 // indirect
-	go.opentelemetry.io/otel/trace v1.38.0 // indirect
+	go.opentelemetry.io/otel v1.41.0 // indirect
+	go.opentelemetry.io/otel/trace v1.41.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/crypto v0.42.0 // indirect
+	golang.org/x/crypto v0.52.0 // indirect
 	golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 // indirect
-	golang.org/x/mod v0.27.0 // indirect
-	golang.org/x/net v0.44.0 // indirect
-	golang.org/x/oauth2 v0.30.0 // indirect
-	golang.org/x/sync v0.17.0 // indirect
-	golang.org/x/sys v0.36.0 // indirect
-	golang.org/x/term v0.35.0 // indirect
-	golang.org/x/text v0.29.0 // indirect
+	golang.org/x/mod v0.35.0 // indirect
+	golang.org/x/net v0.55.0 // indirect
+	golang.org/x/oauth2 v0.34.0 // indirect
+	golang.org/x/sys v0.45.0 // indirect
+	golang.org/x/telemetry v0.0.0-20260409153401-be6f6cb8b1fa // indirect
+	golang.org/x/term v0.43.0 // indirect
+	golang.org/x/text v0.37.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
-	golang.org/x/tools v0.36.0 // indirect
+	golang.org/x/tools v0.44.0 // indirect
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/linkedin/goavro.v1 v1.0.5 // indirect

go.sum

@@ -0,0 +1,142 @@
+package starrocks
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
+
+	"github.com/patterninc/heimdall/pkg/result/column"
+)
+
+var arrowTypeToResultTypeName = map[arrow.Type]string{
+	arrow.INT8:         "int",
+	arrow.INT16:        "int",
+	arrow.INT32:        "int",
+	arrow.INT64:        "long",
+	arrow.UINT8:        "int",
+	arrow.UINT16:       "int",
+	arrow.UINT32:       "long",
+	arrow.UINT64:       "long",
+	arrow.FLOAT32:      "float",
+	arrow.FLOAT64:      "double",
+	arrow.DECIMAL128:   "double",
+	arrow.DECIMAL256:   "double",
+	arrow.STRING:       "string",
+	arrow.LARGE_STRING: "string",
+	arrow.BINARY:       "string",
+	arrow.LARGE_BINARY: "string",
+	arrow.BOOL:         "boolean",
+	arrow.DATE32:       "string",
+	arrow.DATE64:       "string",
+	arrow.TIMESTAMP:    "string",
+	arrow.TIME32:       "string",
+	arrow.TIME64:       "string",
+	arrow.LIST:         "string",
+	arrow.LARGE_LIST:   "string",
+	arrow.MAP:          "string",
+	arrow.STRUCT:       "string",
+}
+
+func columnsFromSchema(schema *arrow.Schema) []*column.Column {
+
+	fields := schema.Fields()
+	columns := make([]*column.Column, len(fields))
+
+	for i, f := range fields {
+		typeName, ok := arrowTypeToResultTypeName[f.Type.ID()]
+		if !ok {
+			typeName = "string"
+		}
+		columns[i] = &column.Column{
+			Name: f.Name,
+			Type: column.Type(typeName),
+		}
+	}
+
+	return columns
+
+}
+
+func recordToRows(rec arrow.Record) [][]any {
+
+	numRows := int(rec.NumRows())
+	numCols := int(rec.NumCols())
+
+	rows := make([][]any, numRows)
+	for r := 0; r < numRows; r++ {
+		row := make([]any, numCols)
+		for c := 0; c < numCols; c++ {
+			row[c] = arrowValue(rec.Column(c), r)
+		}
+		rows[r] = row
+	}
+
+	return rows
+
+}
+
+func arrowValue(col arrow.Array, i int) any {
+
+	if col.IsNull(i) {
+		return nil
+	}
+
+	switch v := col.(type) {
+	case *array.Int8:
+		return v.Value(i)
+	case *array.Int16:
+		return v.Value(i)
+	case *array.Int32:
+		return v.Value(i)
+	case *array.Int64:
+		return v.Value(i)
+	case *array.Uint8:
+		return v.Value(i)
+	case *array.Uint16:
+		return v.Value(i)
+	case *array.Uint32:
+		return v.Value(i)
+	case *array.Uint64:
+		return v.Value(i)
+	case *array.Float32:
+		return v.Value(i)
+	case *array.Float64:
+		return v.Value(i)
+	case *array.Decimal128:
+		return v.Value(i).ToFloat64(int32(v.DataType().(*arrow.Decimal128Type).Scale))
+	case *array.Decimal256:
+		return v.Value(i).ToFloat64(int32(v.DataType().(*arrow.Decimal256Type).Scale))
+	case *array.String:
+		return v.Value(i)
+	case *array.LargeString:
+		return v.Value(i)
+	case *array.Binary:
+		return string(v.Value(i))
+	case *array.LargeBinary:
+		return string(v.Value(i))
+	case *array.Boolean:
+		return v.Value(i)
+	case *array.Date32:
+		return v.Value(i).ToTime().Format("2006-01-02")
+	case *array.Date64:
+		return v.Value(i).ToTime().Format("2006-01-02")
+	case *array.Timestamp:
+		unit := v.DataType().(*arrow.TimestampType).Unit
+		return v.Value(i).ToTime(unit).Format("2006-01-02T15:04:05.999999999")
+	case *array.Time32:
+		return v.Value(i).ToTime(v.DataType().(*arrow.Time32Type).Unit).Format("15:04:05")
+	case *array.Time64:
+		return v.Value(i).ToTime(v.DataType().(*arrow.Time64Type).Unit).Format("15:04:05.999999999")
+	case *array.List, *array.LargeList, *array.Map, *array.Struct:
+		b, err := json.Marshal(col.GetOneForMarshal(i))
+		if err != nil {
+			return fmt.Sprintf("%v", col.GetOneForMarshal(i))
+		}
+		return string(b)
+	default:
+		return fmt.Sprintf("%v", col.GetOneForMarshal(i))
+	}
+
+}

internal/pkg/object/command/starrocks/arrow_types.go

@@ -0,0 +1,114 @@
+package starrocks
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"sync"
+
+	"github.com/apache/arrow-go/v18/arrow/flight"
+	"github.com/apache/arrow-go/v18/arrow/flight/flightsql"
+)
+
+// backendClientPool lazily dials one Flight SQL client per unique BE
+// location, reusing the FE client when an endpoint has no distinct BE address.
+type backendClientPool struct {
+	feClient       *flightsql.Client
+	dialedEndpoint string
+	useTLS         bool
+
+	mu      sync.Mutex
+	clients map[string]*flightsql.Client
+}
+
+func newBackendClientPool(feClient *flightsql.Client, dialedEndpoint string, useTLS bool) *backendClientPool {
+	return &backendClientPool{
+		feClient:       feClient,
+		dialedEndpoint: dialedEndpoint,
+		useTLS:         useTLS,
+		clients:        make(map[string]*flightsql.Client),
+	}
+}
+
+// clientFor returns the FE client, or a cached/dialed direct BE connection if the endpoint has a distinct location.
+func (p *backendClientPool) clientFor(ctx context.Context, endpoint *flight.FlightEndpoint) (*flightsql.Client, error) {
+
+	for _, loc := range endpoint.Location {
+		if loc.Uri == flight.LocationReuseConnection {
+			continue
+		}
+		host := locationHost(loc.Uri)
+		if host == `` {
+			return nil, fmt.Errorf("failed to parse flight endpoint location %q", loc.Uri)
+		}
+		if host == p.dialedEndpoint {
+			continue
+		}
+		return p.get(ctx, host)
+	}
+
+	return p.feClient, nil
+
+}
+
+func (p *backendClientPool) get(ctx context.Context, host string) (*flightsql.Client, error) {
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if c, ok := p.clients[host]; ok {
+		return c, nil
+	}
+
+	c, err := dialFlightClient(ctx, host, p.useTLS)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to StarRocks backend %q: %v", host, err)
+	}
+
+	p.clients[host] = c
+	return c, nil
+
+}
+
+func (p *backendClientPool) closeAll() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for _, c := range p.clients {
+		c.Client.Close()
+	}
+}
+
+func fetchEndpoint(ctx context.Context, beClients *backendClientPool, endpoint *flight.FlightEndpoint) ([][]any, error) {
+
+	client, err := beClients.clientFor(ctx, endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	reader, err := client.DoGet(ctx, endpoint.Ticket)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch result stream: %v", err)
+	}
+	defer reader.Release()
+
+	rows := make([][]any, 0, 128)
+	for reader.Next() {
+		rows = append(rows, recordToRows(reader.Record())...)
+	}
+
+	if err := reader.Err(); err != nil {
+		return nil, fmt.Errorf("error reading result stream: %v", err)
+	}
+
+	return rows, nil
+
+}
+
+// locationHost extracts host:port from a Flight location URI (e.g. "grpc+tcp://host:9408").
+func locationHost(uri string) string {
+	u, err := url.Parse(uri)
+	if err != nil {
+		return ``
+	}
+	return u.Host
+}