Switch to Arrow Flight

[raymondoyondi]

No description provided.

[coderabbitai]

Warning

Review limit reached

@raymondoyondi, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 22 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: aaca2b12-f064-4bf2-a4ad-589c85edcf49

📥 Commits

Reviewing files that changed from the base of the PR and between 7e85cb3 and aae5b72.

⛔ Files ignored due to path filters (1)

• package-lock.json is excluded by !**/package-lock.json

📒 Files selected for processing (30)

• backend/app/config.py
• backend/app/middleware/__init__.py
• backend/app/middleware/guardrail_middleware.py
• backend/app/routers/gemini.py
• backend/app/routers/telemetry.py
• backend/app/services/arrow_service.py
• backend/app/services/gemini_service.py
• backend/app/utils/cache.py
• backend/main.py
• backend/tests/test_arrow_service.py
• backend/tests/test_cache_validated.py
• backend/tests/test_guardrail_middleware.py
• package.json
• src/App.tsx
• src/components/ArrowTelemetry.tsx
• src/components/DragDropGrid.tsx
• src/components/DuckDBAnalytics.tsx
• src/components/RagSearch.tsx
• src/components/charts/EChartsVisualization.tsx
• src/hooks/useDuckDB.ts
• src/index.css
• src/lib/arrowClient.test.ts
• src/lib/arrowClient.ts
• src/lib/duckdbClient.ts
• src/lib/duckdbQueries.test.ts
• src/lib/duckdbQueries.ts
• src/lib/ragClient.test.ts
• src/lib/ragClient.ts
• src/vite-env.d.ts
• vitest.config.ts

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch agent/lunar-coral-bjcy

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[greptile-apps]

Greptile Summary

This PR switches telemetry and analytics paths toward Arrow-based browser processing. The main changes are:

• Async guardrail middleware for Gemini requests.
• Arrow IPC telemetry endpoints and frontend decoding.
• DuckDB-WASM analytics in the browser.
• Hybrid local RAG matching with ONNX fallback.
• Tree-shaken ECharts and CSS grid layout updates.

Confidence Score: 4/5

The DuckDB analytics path can return duplicated rows after the analytics view remounts.

• The guardrail move and Arrow schema path look consistent with the inspected routes and consumers.
• The browser DuckDB singleton keeps its table across component remounts, while the hook reloads the same devices again.
• Arrow response metadata reports bytes where the header says rows.

src/lib/duckdbClient.ts, backend/app/routers/telemetry.py

Important Files Changed

Filename	Overview
backend/app/middleware/guardrail_middleware.py	Adds async prompt guardrails for the Gemini command route.
backend/app/routers/telemetry.py	Adds Arrow dataset routes, with incorrect row-count response metadata.
backend/app/services/arrow_service.py	Adds Arrow schema conversion, IPC serialization, and optional Flight server support.
src/lib/arrowClient.ts	Adds browser-side Arrow IPC fetch and table conversion helpers.
src/lib/duckdbClient.ts	Adds a DuckDB-WASM singleton, but repeated loads append duplicate device rows.
src/hooks/useDuckDB.ts	Loads device data into DuckDB using a per-mount fingerprint guard.
src/lib/ragClient.ts	Adds local semantic matching with optional ONNX embeddings and hash fallback.
src/components/charts/EChartsVisualization.tsx	Switches ECharts rendering to core imports with explicit module registration.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  Telemetry[FastAPI telemetry data] --> Arrow[Arrow IPC endpoints]
  Arrow --> ArrowUI[ArrowTelemetry UI]
  Arrow --> DuckUI[DuckDBAnalytics UI]
  DuckUI --> DuckDB[DuckDB-WASM singleton]
  DuckDB --> Queries[Local filters and aggregates]
  GeminiReq[Gemini command POST] --> Guardrails[Async guardrail middleware]
  Guardrails --> Gemini[Gemini router]

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart LR
  Telemetry[FastAPI telemetry data] --> Arrow[Arrow IPC endpoints]
  Arrow --> ArrowUI[ArrowTelemetry UI]
  Arrow --> DuckUI[DuckDBAnalytics UI]
  DuckUI --> DuckDB[DuckDB-WASM singleton]
  DuckDB --> Queries[Local filters and aggregates]
  GeminiReq[Gemini command POST] --> Guardrails[Async guardrail middleware]
  Guardrails --> Gemini[Gemini router]

Loading

Prompt To Fix All With AI

Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
src/lib/duckdbClient.ts:65
**DuckDB Rows Append On Remount**

`DuckDBAnalytics` is mounted only while the analytics view is active, but `duckDBAnalytics` is a singleton and `loadDevices()` only inserts into the existing `devices` table. Switching away and back resets the hook's `loadedKey` and inserts the same Arrow rows again, so filters and aggregates return duplicated devices and inflated counts.

```suggestion
    await conn.query(buildCreateDevicesTableSQL());
    await conn.insertArrowTable(table as never, { name: "devices" });
```

### Issue 2 of 2
backend/app/routers/telemetry.py:127-133
**Row Header Reports Bytes**

`payload` is the serialized Arrow IPC byte stream, so `len(payload)` reports byte length instead of table rows. Any client or gateway that uses `X-Arrow-Rows` for dataset size will receive incorrect metadata for every Arrow dataset.

```suggestion
    telemetry = get_default_telemetry()
    table = arrow.telemetry_to_table(telemetry, dataset)
    payload = arrow.serialize_ipc(table)
    return StreamingResponse(
        iter([payload]),
        media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
        headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(table.num_rows)},
    )
```

_{Reviews (1): Last reviewed commit: "feat(frontend): native Tailwind 4 CSS gr..." | Re-trigger Greptile}

[greptile-apps]

DuckDB Rows Append On Remount

DuckDBAnalytics is mounted only while the analytics view is active, but duckDBAnalytics is a singleton and loadDevices() only inserts into the existing devices table. Switching away and back resets the hook's loadedKey and inserts the same Arrow rows again, so filters and aggregates return duplicated devices and inflated counts.

Suggested change

	await conn.insertArrowTable(table as never, { name: "devices" });
	await conn.query(buildCreateDevicesTableSQL());
	await conn.insertArrowTable(table as never, { name: "devices" });

Prompt To Fix With AI

This is a comment left during a code review.
Path: src/lib/duckdbClient.ts
Line: 65

Comment:
**DuckDB Rows Append On Remount**

`DuckDBAnalytics` is mounted only while the analytics view is active, but `duckDBAnalytics` is a singleton and `loadDevices()` only inserts into the existing `devices` table. Switching away and back resets the hook's `loadedKey` and inserts the same Arrow rows again, so filters and aggregates return duplicated devices and inflated counts.

```suggestion
    await conn.query(buildCreateDevicesTableSQL());
    await conn.insertArrowTable(table as never, { name: "devices" });
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

Row Header Reports Bytes

payload is the serialized Arrow IPC byte stream, so len(payload) reports byte length instead of table rows. Any client or gateway that uses X-Arrow-Rows for dataset size will receive incorrect metadata for every Arrow dataset.

Suggested change

	telemetry = get_default_telemetry()
	payload = arrow.serialize_dataset(telemetry, dataset)
	return StreamingResponse(
	iter([payload]),
	media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
	headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(len(payload))},
	)
	telemetry = get_default_telemetry()
	table = arrow.telemetry_to_table(telemetry, dataset)
	payload = arrow.serialize_ipc(table)
	return StreamingResponse(
	iter([payload]),
	media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
	headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(table.num_rows)},
	)

Prompt To Fix With AI

This is a comment left during a code review.
Path: backend/app/routers/telemetry.py
Line: 127-133

Comment:
**Row Header Reports Bytes**

`payload` is the serialized Arrow IPC byte stream, so `len(payload)` reports byte length instead of table rows. Any client or gateway that uses `X-Arrow-Rows` for dataset size will receive incorrect metadata for every Arrow dataset.

```suggestion
    telemetry = get_default_telemetry()
    table = arrow.telemetry_to_table(telemetry, dataset)
    payload = arrow.serialize_ipc(table)
    return StreamingResponse(
        iter([payload]),
        media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
        headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(table.num_rows)},
    )
```

How can I resolve this? If you propose a fix, please make it concise.

[gemini-code-assist]

Code Review

This pull request introduces an asynchronous prompt-injection guardrail middleware, zero-copy columnar telemetry streaming using Apache Arrow IPC, and in-browser edge analytics powered by DuckDB-WASM and ONNX-based hybrid RAG. Feedback on these changes highlights several issues: the X-Arrow-Rows header incorrectly reports payload bytes instead of row count; the ONNXEmbedder fails to initialize properly and uses incorrect tensor types; httpx.AsyncClient is inefficiently instantiated on every request; the Arrow Flight server hardcodes an unroutable 0.0.0.0 address; and column lookups in the frontend Arrow client are performed inefficiently inside a loop.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The X-Arrow-Rows header is currently set to str(len(payload)), which represents the serialized payload size in bytes rather than the actual number of rows in the dataset. This is highly misleading for clients. Retrieve the row count from the PyArrow table before serialization.

Suggested change

	telemetry = get_default_telemetry()
	payload = arrow.serialize_dataset(telemetry, dataset)
	return StreamingResponse(
	iter([payload]),
	media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
	headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(len(payload))},
	)
	telemetry = get_default_telemetry()
	table = arrow.telemetry_to_table(telemetry, dataset)
	payload = arrow.serialize_ipc(table)
	return StreamingResponse(
	iter([payload]),
	media_type=arrow.ARROW_STREAM_MEDIA_TYPE,
	headers={"X-Arrow-Dataset": dataset, "X-Arrow-Rows": str(table.num_rows)},
	)

[gemini-code-assist]

There are two critical bugs in ONNXEmbedder:

1. The dynamic import of onnxruntime-web and session creation is only triggered inside ready(), which is never called. Consequently, this.session remains null and the embedder silently falls back to hashEmbedding forever.
2. The makeTensor helper returns a plain object, but onnxruntime-web strictly requires real ort.Tensor instances, which would cause a runtime crash once loaded.

We can fix both by lazily calling ready() on the first embed() call and using the real ort.Tensor constructor.

export class ONNXEmbedder {
  private session: any | null = null;
  private ort: any | null = null;
  private loading: Promise<void> | null = null;
  private readonly modelUrl?: string;
  private readonly dim: number;

  constructor(opts: { modelUrl?: string; dim?: number } = {}) {
    this.modelUrl = opts.modelUrl;
    this.dim = opts.dim ?? DEFAULT_DIM;
  }

  async ready(): Promise<boolean> {
    if (!this.modelUrl) return false;
    if (this.session) return true;
    if (!this.loading) {
      this.loading = this._load();
    }
    try {
      await this.loading;
    } catch {
      this.loading = null;
      return false;
    }
    return this.session !== null;
  }

  private async _load(): Promise<void> {
    // Dynamic import keeps onnxruntime-web out of the main bundle until used.
    const ort = await import("onnxruntime-web");
    const session = await ort.InferenceSession.create(this.modelUrl!, {
      executionProviders: ["wasm"],
    });
    this.session = session;
    this.ort = ort;
  }

  /** Embed one or more texts. Uses ONNX when available, else hashing. */
  async embed(texts: string[]): Promise<number[][]> {
    if (this.modelUrl && !this.session) {
      await this.ready();
    }
    if (!this.session || !this.ort) {
      return texts.map((t) => hashEmbedding(t, this.dim));
    }

    const inputName = this.session.inputNames[0];
    const results: number[][] = [];
    for (const text of texts) {
      const vec = hashEmbedding(text, this.dim);
      const tensor = new this.ort.Tensor("float32", Float32Array.from(vec), [1, vec.length]);
      const out = await this.session.run({ [inputName]: tensor });
      const data = out[this.session.outputNames[0]].data;
      results.push(Array.from(data as Float32Array));
    }
    return results;
  }
}

[gemini-code-assist]

Instantiating a new httpx.AsyncClient on every request is highly inefficient because it does not reuse connection pools, leading to socket exhaustion and increased latency under load. Consider lazily initializing a single client instance on the middleware and reusing it across requests.

Suggested change

	async def _check_edge(self, prompt: str) -> bool: # pragma: no cover - needs live edge svc
	try:
	import httpx
	async with httpx.AsyncClient(timeout=1.0) as client:
	resp = await client.post(self.edge_url, json={"prompt": prompt})
	if resp.status_code == 200:
	data = resp.json()
	return bool(data.get("allowed", True))
	# Non-200 from the edge => treat as blocked for safety.
	return False
	except Exception as exc:
	logger.warning(
	"Edge guardrail unreachable (%s); degrading to heuristic", exc
	)
	result = await asyncio.to_thread(self.guardrails.check_input, prompt)
	return result.allowed
	async def _check_edge(self, prompt: str) -> bool: # pragma: no cover - needs live edge svc
	try:
	import httpx
	if not hasattr(self, "_client"):
	self._client = httpx.AsyncClient(timeout=1.0)
	resp = await self._client.post(self.edge_url, json={"prompt": prompt})
	if resp.status_code == 200:
	data = resp.json()
	return bool(data.get("allowed", True))
	# Non-200 from the edge => treat as blocked for safety.
	return False
	except Exception as exc:
	logger.warning(
	"Edge guardrail unreachable (%s); degrading to heuristic", exc
	)
	result = await asyncio.to_thread(self.guardrails.check_input, prompt)
	return result.allowed

[gemini-code-assist]

Hardcoding 0.0.0.0 as the location in FlightEndpoint is problematic because 0.0.0.0 is not a routable address for clients (especially on non-Unix platforms like Windows). Parse the host and port from the configured uri and fallback to localhost if the host is 0.0.0.0.

    def __init__(
        self,
        uri: str = "grpc://0.0.0.0:8815",
        telemetry_provider: Optional[Callable[[], Dict[str, Any]]] = None,
    ) -> None:
        if not FLIGHT_AVAILABLE:
            raise RuntimeError("pyarrow.flight is not available in this build")
        self._uri = uri
        self._telemetry_provider = telemetry_provider or (lambda: {})

        from urllib.parse import urlparse
        parsed = urlparse(uri)
        self._host = parsed.hostname or "localhost"
        self._port = parsed.port or 8815
        if self._host == "0.0.0.0":
            self._host = "localhost"

[gemini-code-assist]

Update the Flight server builder to use the parsed host and port instead of hardcoded "0.0.0.0" and 8815 so that clients receive a valid, routable endpoint location.

    def _build_server(self) -> "_flight.FlightServerBase":
        provider = self._telemetry_provider
        host = self._host
        port = self._port

        class _Server(_flight.FlightServerBase):
            def list_flights(self, context, criteria):  # noqa: D401, ANN001
                for ds in DATASETS:
                    yield _flight.FlightInfo(
                        telemetry_to_table(provider(), ds).schema,
                        _flight.FlightDescriptor.for_command(ds.encode()),
                        [],
                        -1,
                        -1,
                    )

            def get_flight_info(self, context, descriptor):  # noqa: ANN001
                ds = descriptor.command.decode() if descriptor.command else "devices"
                table = telemetry_to_table(provider(), ds)
                endpoint = _flight.FlightEndpoint(
                    ds.encode(),
                    [_flight.Location.for_grpc_tcp(host, port)],
                )
                return _flight.FlightInfo(
                    table.schema, descriptor, [endpoint], table.num_rows, -1
                )

            def do_get(self, context, ticket):  # noqa: ANN001
                ds = ticket.ticket.decode()
                table = telemetry_to_table(provider(), ds)
                return _flight.RecordBatchStream(table)

            def do_put(self, context, descriptor, reader):  # noqa: ANN001
                # Read-and-discard: this server is a read-only publisher.
                for _ in reader:
                    pass

        return _Server(self._uri)

[gemini-code-assist]

Looking up columns by name inside the loop via col(name) is highly inefficient as it performs a lookup O(N * M) times. Extract the column references once outside the loop to improve performance significantly.

/** Convert a `devices` Arrow table into structured device records. */
export function tableToDevices(table: Table): ArrowDevice[] {
  const idCol = table.getChild("id");
  const scoreCol = table.getChild("score");
  const uptimeCol = table.getChild("uptime");
  const loadCol = table.getChild("load");
  const statusCol = table.getChild("status");
  const typeCol = table.getChild("type");
  const activeCol = table.getChild("active");
  const n = table.numRows;
  const out: ArrowDevice[] = [];

  for (let i = 0; i < n; i++) {
    out.push({
      id: String(idCol?.get(i) ?? ""),
      score: Number(scoreCol?.get(i) ?? 0),
      uptime: Number(uptimeCol?.get(i) ?? 0),
      load: String(loadCol?.get(i) ?? ""),
      status: String(statusCol?.get(i) ?? ""),
      type: String(typeCol?.get(i) ?? ""),
      active: Boolean(activeCol?.get(i) ?? false),
    });
  }
  return out;
}