Skip to content

FermiHDI SDK — Go

BETA. The most exercised of the wrappers: four production services query through it. Still beta, because its API may change with the SDK's.

cgo bindings over the C++ SDK. Read the SDK overview first — particularly §1, the control plane / data plane split, because it explains why this API looks the way it does.

Modes

This binding is built on the C ABI, which today carries no registration call and no ingest client. A Go program therefore works in proxy mode (submit through the Query Proxy, records over the data plane), and can be handed SCC targets with SetSccTarget, but it cannot register as an HD_CLIENT node or ingest. Direct mode and custom ingesters are C++ and Python for now; the SDK overview has the two modes side by side.

Install

The package has no go.mod in this repository: it is vendored by consumers, and the packaged SDK writes the module line when it assembles the payload. From the package image:

COPY --from=cr.fermihdi.io/hd/sdk:<version> /opt/fermihdi-sdk /opt/fermihdi-sdk
RUN . /opt/fermihdi-sdk/cgo-env.sh && go build ./...

cgo-env.sh sets CGO_ENABLED, CGO_CXXFLAGS, CGO_LDFLAGS and LD_LIBRARY_PATH — including the full library list that -lfermihdi_client alone does not pull in. Source it rather than hand-maintaining those flags.

API

func NewQueryClient(proxyURL string) *QueryClient
func NewQueryClientWithInterfaces(proxyURL, dataPlaneIface, ctrlPlaneIface string) *QueryClient

func (q *QueryClient) Close()
func (q *QueryClient) LoadDatasetSchema(schemaJSON string)
func (q *QueryClient) SetSccTarget(ip string, port uint16)
func (q *QueryClient) GenerateHDQuery(query HDQuery) string
func (q *QueryClient) PrepareQuery(queryID string)
func (q *QueryClient) SetExpectedBatches(batchIDs []string)
func (q *QueryClient) SetFilterVerdict(wouldExclude []string, bloomrfMode string)
func (q *QueryClient) FilterFalseNegatives() (mode string, falseNegatives []string)
func (q *QueryClient) ExecuteToIPC(fermihdiJSON, duckdbSQL string) []byte
func (q *QueryClient) ExecuteEmptyToIPC(duckdbSQL string) []byte   // the SQL over zero records
func (q *QueryClient) ExecuteDatasetToIPC(fermihdiJSON string) []byte      // the raw record table, untouched
func (q *QueryClient) ApplySQLToIPC(ipc []byte, duckdbSQL string) []byte   // SQL over a table you hold
func (q *QueryClient) SetQueryTimeoutMs(ms uint64)
func (q *QueryClient) CancelQuery(queryID uint32)

Types

type HDQuery struct {
    QueryID                    uint32
    BatchIDs                   []string
    SubQueries                 []SubQuery
    Targets                    []SccAddress
    QueryQoS                   uint32   // 1–255, priority at the storage node
    AcknowledgeFullDatasetScan bool     // see below
}

type SubQuery struct  { Filters []Filter }
type Filter struct    { FieldName, Value string; Op FilterOp }
type SccAddress struct{ IP string; Port uint16 }

FilterOp: Equality, NotEqual, Greater, Less.

The ordering that matters

PrepareQuery before you dispatch. Storage nodes can start streaming the moment the controllers fan out, which is before your dispatch call returns, and records arriving at an unarmed listener are discarded.

client := fermihdi.NewQueryClient("https://proxy.internal:443")
defer client.Close()

client.LoadDatasetSchema(schemaJSON)

hq := fermihdi.HDQuery{
    QueryID:  nextQueryID(),          // distinct per query — see below
    BatchIDs: batchIDs,
    Targets:  []fermihdi.SccAddress{{IP: advertisedIP, Port: 6987}},
    SubQueries: []fermihdi.SubQuery{{
        Filters: []fermihdi.Filter{
            // An address is its integer: 10.0.0.1 is 167772161. Convert before the SDK sees it.
            {FieldName: "ip_src", Value: "167772161", Op: fermihdi.Equality},
        },
    }},
}
queryJSON := client.GenerateHDQuery(hq)
if queryJSON == "" {
    return errors.New("query refused: no batch_ids and no full-scan acknowledgement")
}

// 1. arm the listener
client.PrepareQuery(strconv.FormatUint(uint64(hq.QueryID), 10))

// 2. control plane: POST to the proxy yourself, read found_batches
found, err := dispatchToProxy(proxyURL, queryJSON)
if err != nil {
    reportQueryEnd(proxyURL, hq.QueryID, "cancel")   // you own the ending
    return err
}

// 3. wait for exactly what exists, and say what the filter would have skipped
client.SetExpectedBatches(found)
client.SetFilterVerdict(wouldExclude, bloomrfMode)   // from the same result event

// 4. data plane: blocks until every expected batch has arrived
ipc := client.ExecuteToIPC(queryJSON, "SELECT ip_src, bytes FROM ipflows LIMIT 10")

// 5. tell the cluster it is over
reportQueryEnd(proxyURL, hq.QueryID, "complete")

Query IDs must be distinct

The ID ties together the listener, the returning end-of-batch markers, and the ending you report. Two queries sharing an ID are indistinguishable at every one of those points — a completion for one untracks the other.

Seed a counter at random, not from the clock: services that start in the same second otherwise issue identical first IDs.

var queryIDSeq atomic.Uint32

func init() {
    var b [4]byte
    if _, err := rand.Read(b[:]); err == nil {
        queryIDSeq.Store(binary.BigEndian.Uint32(b[:]))
    } else {
        queryIDSeq.Store(uint32(time.Now().UnixNano()) ^ uint32(os.Getpid())<<16)
    }
}

func nextQueryID() uint32 { return queryIDSeq.Add(1) }

You own the ending

This binding does not submit the query, so it cannot report the ending either. Whatever route you dispatched by, relay complete or cancel back the same way — on success, on error, on timeout, and on your own shutdown.

Omitting cancel is the costly one: the storage nodes keep reading for a query nobody is waiting for.

Reading results

ExecuteToIPC returns Arrow IPC bytes, or an empty slice when there is nothing. When the dispatch itself came back with no batches to read -- an empty window, or a filter service that excluded every candidate -- there is nothing to wait for and ExecuteToIPC is never called; ExecuteEmptyToIPC(sql) then gives the row an aggregate produces over nothing (count 0, sum NULL), so the answer keeps its shape.

if len(ipc) == 0 {
    return nil, nil          // no rows; not an error
}
reader, err := ipc.NewReader(bytes.NewReader(ipcBytes))

Use cases

A query service. Hold one QueryClient for the process lifetime, generate a distinct ID per request, and relay endings. This is what hd_sql, hd_mcp, hd_graphql and hd_ipflow_graphql do.

A batch job over a wide window. Set AcknowledgeFullDatasetScan in code — you have decided a full scan is what you mean — and raise SetQueryTimeoutMs.

Behind NAT. Bind locally, advertise the routable address, and make sure the advertised host and port appear in Targets. Records sent where nothing is listening are lost with no error on either side.

Gotchas

  • GenerateHDQuery returning "" is the full-dataset-scan guard.
  • Waiting on batches that were never stored is how a query hangs to its timeout. Always SetExpectedBatches(found).
  • The SDK only learns what the filter service would have excluded when it places the query itself. A Go service reads the proxy's stream, so it must pass would_exclude_batches and bloomrf_mode on with SetFilterVerdict, or the false-negative check never runs and passive mode measures nothing.
  • Close() releases the C++ client. It does not cancel a running query — send the cancel first.