Queries and runs

import { useComputation } from "@bijectionhq/compute/react";
import { usePaginatedQueryState } from "bijection/react";

const report = useComputation(api.customerHealth, { interactionKey });
const run = report.run;
const customers = usePaginatedQueryState({
  query: api.customerHealth.listCustomers,
  args:
    run && report.observation?.status === "success" &&
    report.observation.data.outputs.customers.kind === "available"
      ? { run }
      : "skip",
  initialNumItems: 50,
});
// In an explicit event handler:
await report.start({ tenantId, ownerId, windowEndDay });

export const { start, get, listCustomers, getSummary } =
  healthRuns.api(internal.customerSuccess.customerHealth, {
    access: reportAccess,
    readers: { listCustomers: "customers", getSummary: "summary" },
  });

const compute = new Compute(components.compute);

export const calculate = internalAction(
  compute.sqlHandler({
    args: { region: v.string() },
    returns: v.object({ rows: dataset(regionTotalsRow) }),
    outputs: {
      rows: `SELECT region, COUNT(*) AS "orderCount"
             FROM orders WHERE region = :region GROUP BY region`,
    },
    limits: { maxRows: 100_000, maxBytes: 64 * 1024 * 1024 },
    retainForMs: 86_400_000,
  }),
);

export const { start, get, listRows } = compute.api(internal.reports.calculate, {
  access: reportAccess,
  readers: { listRows: "rows" },
});

import { q, sqlRelation } from "bijection/server";

const ordersByRegion = sqlRelation({
  sql: `
    SELECT c.region AS region, COUNT(*) AS orders
    FROM customers AS c
    INNER JOIN orders AS o ON c.customer = o.customer
    WHERE c.tenant = :tenant
    GROUP BY c.region
    ORDER BY orders DESC, region ASC
    LIMIT 100
  `,
  inputs: {
    customers: q.from(schema.tables.customers),
    orders: q.from(schema.tables.orders),
  },
  parameters: { tenant: "acme" },
});

const localOrders = sqlRelation({
  context: { timeZone: "Europe/Paris", collation: "binary" },
  inputs: { orders },
  sql: "SELECT TO_DATE(created_at) AS local_date FROM orders",
});

/// <reference types="bijection/sql-files" />
import query from "./queries/revenue.sql";

const revenue = sqlRelation({
  sql: query,
  sourceFile: "queries/revenue.sql",
  inputs: { orders },
  parameters: { minimum: 100n },
});

import { inspectSQL } from "bijection/computations/inspection";

const coverage = inspectSQL({
  query: {
    sql: query,
    sourceFile: "queries/revenue.sql",
    inputs: { orders },
    parameters: { minimum: 100n },
  },
  schemas: { orders: orderColumns },
  targets: [
    { kind: "portable_rows" },
    { kind: "sql", dialect: "spark" },
    { kind: "sql", dialect: "snowflake" },
  ],
});
console.log(JSON.stringify(coverage));

SELECT source.id, item.pos, item.col
FROM source
CROSS JOIN POSEXPLODE(source.items) AS item;

const summary = q.from(schema.tables.customers).select((customer) => ({
  tags: q.arrayDistinct(customer.tags),
  matchingTags: q.arrayIntersect(customer.tags, customer.preferredTags),
  emailDigest: q.sha2(customer.email, 256),
}));

bijection sql client powerbi --endpoint https://sql.example.com --output ./powerbi-client
bijection sql client tableau --endpoint https://sql.example.com --output ./tableau-client

CREATE TABLE totals (id BIGINT, at TIMESTAMP)
USING iceberg
TBLPROPERTIES ('comment' = 'Daily totals', 'user.owner' = 'analytics')
PARTITION BY (day(at), bucket(16, id));

BEGIN;
ALTER TABLE totals ADD COLUMN note STRING;
INSERT INTO totals (id, note) VALUES (1, :note);
COMMIT;

ALTER TABLE totals CREATE TAG release;
ALTER TABLE totals CREATE BRANCH draft AS OF VERSION 'tag:release';
ALTER TABLE totals REPLACE BRANCH draft AS OF VERSION 'accepted-version-id';
SELECT * FROM totals VERSION AS OF 'tag:release';
ALTER TABLE totals DROP TAG IF EXISTS release;

SELECT id FROM totals VERSION AS OF 'retained-version-id';

import { SQLWarehouse } from "@bijectionhq/pipelines/sql";
import { internal } from "./_generated/api";
import { datasets, compute, totals } from "./resources";

const warehouse = new SQLWarehouse({
  owner: "reporting",
  datasets,
  compute,
  targets: {
    totals: {
      dataset: totals,
      selection: "totals",
      writable: true,
      candidate: internal.materialize.totals,
    },
  },
  authorize: internal.policy.warehouse,
  executionAccess: internal.warehouse.executorAccess,
});
export const { catalog, prepare, compile, begin, end, execute } =
  warehouse.functions();
export const executorAccess = warehouse.executorPolicy();
export const candidateAccess = warehouse.resourcePolicy("totals", {
  release: internal.policy.releaseSQLResult,
});

bijection sql serve --warehouse warehouse

bijection sql catalog
bijection sql catalog --features
bijection sql explain --file orders.sql --params @parameters.json
bijection run sql --file orders.sql --params @parameters.json \
  --max-rows 1000 --max-bytes 4194304

bijection sql explain --file orders.sql --execution '{"kind":"binding","binding":"analytics"}'
bijection run sql --file orders.sql --execution '{"kind":"binding","binding":"analytics"}'

bijection sql serve --listen 127.0.0.1:31337
# A network listener requires a PEM certificate and key.
bijection sql serve --listen 0.0.0.0:31337 \
  --tls-cert server.pem --tls-key server.key

bijection sql serve --preflight \
  --execution '{"kind":"binding","binding":"analytics"}' \
  --max-result-rows 10000000 --max-result-bytes 10737418240

bijection auth credential create reports --name bi
bijection access grant <type>:<role> --to service:reports --on <type>/<id>

bijection sql serve --listen 0.0.0.0:31337 \
  --tls-cert server.pem --tls-key server.key

bijection+flightsql://sql.example.com:31337/bijection

{
  "connect_args": { "username": "reports", "password": "<service credential>" },
  "pool_pre_ping": true,
  "hide_parameters": true
}

import { applicationApi } from "bijection/apps";
import { bijectionToJson } from "bijection/values";

const api = applicationApi();
const prepared = await client.query(api.sqlCompile, {
  sql: "SELECT c.region FROM customers c WHERE c.denominator > :minimum",
  parameters: JSON.stringify(bijectionToJson({ minimum: 9007199254740993n })),
});
const input = {
  program_json: prepared.program_json,
  expected_inputs_digest: prepared.expected_inputs_digest,
};
const plan = await client.action(api.sqlExplain, input);

const sql = `SELECT c.region FROM customers c
  WHERE c.denominator > CAST(:minimum AS BIGINT)`;
const template = await client.query(api.sqlPrepare, { sql });
const bound = await client.query(api.sqlBind, {
  sql,
  expected_template_json: template.program_json,
  expected_inputs_digest: template.expected_inputs_digest,
  parameters: JSON.stringify(bijectionToJson({ minimum: 9007199254740993n })),
});
// bound.program_json and bound.expected_inputs_digest identify the exact program.

import { sqlTableProvider } from "bijection/apps";

const catalog = await client.query(api.sqlCatalog, {});
const selected = sqlTableProvider(catalog, {
  kind: "where",
  input: { kind: "collection", collection: "customers" },
  filters: [{ field: "region", operator: "eq", value: "EU" }],
});
const sql =
  "SELECT region, COUNT(*) AS customers FROM selected GROUP BY region";
const tableProviders = { selected };
const template = await client.query(api.sqlPrepare, {
  sql,
  tableProviders,
});
const bound = await client.query(api.sqlBind, {
  sql,
  tableProviders,
  expected_template_json: template.program_json,
  expected_inputs_digest: template.expected_inputs_digest,
  parameters: "{}",
});

DECLARE @selected `customers`;
SELECT region FROM @selected;

DECLARE @minimum DOUBLE = 75000.0, @department STRING = 'Engineering';
SELECT name FROM employees
WHERE salary > @minimum AND department = @department;

import {
  sqlInterfaceTableProvider,
  sqlSeriesTableProvider,
} from "bijection/apps";

const selected = sqlInterfaceTableProvider(catalog, {
  interface: "Party",
  implementations: ["customers"],
});
const interfaceTemplate = await client.query(api.sqlPrepare, {
  sql: "SELECT * FROM selected",
  tableProviders: { selected },
});
const samples = sqlSeriesTableProvider(catalog, {
  collection: "readings",
  series_ids: [seriesId],
  from,
  to,
  fields: ["temperature"],
});
const template = await client.query(api.sqlPrepare, {
  sql: "SELECT s.series, s.time, s.values.temperature FROM samples AS s ORDER BY time",
  tableProviders: { samples },
});

const args = {
  sql: "SELECT region FROM customers WHERE denominator > :minimum",
  parameters: { minimum: 100n },
  max_rows: 1000,
  max_bytes: 1024 * 1024,
};
const branchClient = await client.withContext(branchRef);
try {
  const branchResult = await branchClient.query(api.sqlRead, args);
} finally {
  await branchClient.close();
}
const historical = await client
  .at(snapshotRef)
  .queryWithBasis(api.sqlRead, args);
// historical.value is the complete SQL result; historical.basis identifies
// the retained snapshot and installed query program that produced it.

import { defineVirtualSeries, q } from "bijection/server";

export const lossHistory = defineVirtualSeries({ source: observations, temporal: { key: ["tank_id"], observations: { kind: "revisable" as const, key: ["reading_id"] as const, withdrawal: null, order: "source_order" }, time: "time_us", measures: { loss: { field: "loss_ppm", unit: "ppm" } } } });

const ranking = q
  .from(population)
  .summarizeSeries("loss", {
    from: lossHistory.measure("loss"),
    interval: { from: 0n, to: 86400000000n },
    on: (tank, sample) => q.eq(tank.tank_id, sample.subject.tank_id),
    measures: (m) => ({
      samples: m.count(),
      peak: m.max((r) => r.value),
    }),
  })
  .orderBy((tank) => [q.desc(tank.loss.peak), tank.tank_id])
  .limit(100);

const selected = q
  .from(Facility, { implementations: [warehouses, factories] })
  .where((row) => q.eq(row.region, "eu"))
  .select((row) => ({ code: row.code, name: row.name }))
  .orderBy((row) => [row.name, row.code])
  .limit(50);

const facilities = q.program({ rows: selected });
// In an authorized action, explicitly enter the installed analytical owner.
import { runs } from "bijection/analytics";
const { request, submission } = await runs(ctx).start(facilities);
// Preserve request.ref and inspect submission before continuing the request.
// Readiness, execution recovery and work limits belong to its owner.

const basisRequirements = {
  fixed: ["/tables/orders"],
  snapshots: [{ inputs: ["/tables/orders", "/tables/customers"] }],
  freshness: [
    {
      input: "/tables/orders",
      scope: "sources" as const,
      clock: "observation" as const,
      max_age_ms: 60_000,
    },
  ],
  retention: [],
};

limits: {
  maxRows: 128,
  maxBytes: 1_048_576,
  maxScannedBytesPerStatement: 1_000_000_000n,
  maxProviderBilledBytesPerRun: 10_000_000_000n,
},

const completeInputs = defineReadRequirements({
  scope: "allSources",
  default: {
    population: "complete",
    consistency: "sourceSnapshot",
    maxObservationAge: "5m",
  },
});

const totals = defineView({
  relation: q.from(invoices),
  requirements: completeInputs,
});

const invoices = defineSyncedTable({source: erp.invoices, ...{
    read: {
    population: "complete",
    consistency: "sourceSnapshot",
    maxObservationAge: "15m",
  },
  }})
  .index("by_vendor_status", ["vendorKey", "status"]);

const source = await ctx.sources.resolve({
  table: "invoices",
  scope: { companyId: vendor.companyId },
});

const openInvoice = await ctx.db
  .query(source.table)
  .withIndex("by_vendor_status", (q) =>
    q.eq("vendorKey", vendor.erpKey).eq("status", "open"),
  )
  .require({ maxObservationAge: "1m" })
  .first();

if (openInvoice !== null) {
  throw new Error("The vendor has open invoices");
}

const assessment = await ctx.db.checkRead({
  sources: {
    invoices: {
      population: "complete",
      consistency: "sourceSnapshot",
      maxObservationAge: "1m",
    },
  },
});

if (assessment.status === "ready") {
  // Exact local revision and evidence selected for this assessment.
  console.log(assessment.basis.revision, assessment.validUntilMs);
} else {
  // Closed native categories: missingEvidence or unsupported.
  console.log(assessment.issues);
}

const invoices = defineVirtualTable({
  executor: analytics,
  schema: { invoiceId: v.string(), amount: v.int64() },
})
  .key("invoiceId")
  .access(invoiceAccess);

{
  "invoices": {
    "catalog": "snowflake",
    "schema": "billing",
    "table": "INVOICES",
    "columns": { "invoiceId": "INVOICE_ID", "amount": "AMOUNT" }
  }
}

bijection connection executor setup analytics \
  --integration analytics:analytics --engine trino \
  --endpoint https://trino.example:8443 --user reader \
  --token-file /private/trino.token --catalog-binding lake=iceberg \
  --tables executor-tables.json
bijection connection executor bindings analytics
bijection connection executor discover analytics
bijection connection executor probe analytics
bijection connection executor preflight reports:sales

// bijection/analytics.ts
import { defineExecutor } from "bijection/connectors";
export const lake = defineExecutor({ engine: "trino" });
export const legacy = defineExecutor({ engine: "presto" });
export const finance = defineExecutor({ engine: "bigquery" });

bijection check --remote=offline --candidate-out executor-candidate.json
bijection connection executor setup analytics --integration analytics:lake \
  --candidate executor-candidate.json --engine trino \
  --endpoint https://trino.example.com --credential-ref ANALYTICS_TOKEN

{
  "projects": {
    "lake": {
      "adapter": "sql.catalog",
      "source": { "file": "remote/lake.json" }
    },
    "legacy": {
      "adapter": "sql.catalog",
      "source": { "file": "remote/legacy.json" }
    },
    "finance": {
      "adapter": "sql.catalog",
      "source": { "file": "remote/finance.json" }
    }
  }
}

{
  "executor": "analytics:lake",
  "tables": {
    "orders": {
      "source": {
        "catalog": "iceberg",
        "schema": "commerce",
        "table": "orders"
      },
      "columns": {
        "tenantId": "tenant_id",
        "orderId": "order_id",
        "amountMinor": "amount_minor"
      }
    }
  }
}

# Print a candidate capture without changing source expectations.
bijection remote capture lake --connection analytics

# Write captured contracts for source review.
bijection remote capture lake --connection analytics --write

# Fully offline generation and freshness comparison for remote bindings.
bijection remote codegen
bijection remote codegen --check
bijection remote check --offline

# Review the physical placements and install them with the executor connection.
bijection remote placements lake > lake.placements.json
bijection connection executor setup analytics --integration analytics:lake \
  --engine trino --endpoint https://trino.example.com \
  --credential-ref ANALYTICS_TOKEN --tables lake.placements.json

# Compare with the selected deployment's executor.
bijection remote check lake --connection analytics

# Discover proposals and drift without changing approved declarations.
bijection remote scan lake --connection analytics --catalog iceberg \
  --schema commerce --limit 16
bijection remote scan lake --connection analytics --catalog iceberg \
  --schema commerce --limit 16 --resume

# Ordinary codegen also generates remote bindings before module discovery.
bijection codegen

import { defineVirtualTable } from "bijection/server";
import { remote } from "./_generated/imports";
import { lake } from "./analytics";
import { remoteAccess } from "./policies";

export const orders = defineVirtualTable({
  executor: lake,
  source: remote.lake.orders,
})
  .key("tenantId", "orderId")
  .access(remoteAccess);

{
  "executor": "analytics:finance",
  "tables": {
    "invoices": {
      "targetSources": {
        "development": {
          "project": "acme-development",
          "dataset": "billing",
          "table": "invoices"
        },
        "production": {
          "project": "acme-production",
          "dataset": "billing",
          "table": "invoices"
        }
      },
      "columns": { "invoiceId": "invoice_id", "amountMinor": "amount_minor" }
    }
  }
}

bijection remote capture finance --deployment development --connection finance --write
# Select production using the ordinary deployment selection options.
bijection remote capture finance --deployment production --connection finance --write
bijection remote codegen

source: remote.finance.invoices.forTarget("production");

bijection check --remote=required
bijection check --remote=offline
bijection deploy
bijection connection executor preflight reports:tenantOrders \
  --args '{"tenantId":"example"}' --provider-plan

const rows = await requestCatalog.output(ctx, reportRequest, "rows");
if (rows === null) throw new Error("Report output is not accepted yet");
const pipelineRun = await inventoryPipeline.start(
  ctx,
  { day: "2026-10-19", raw_inventory: rows },
  { requestKey },
);
await requestCatalog.close(ctx, reportRequest);

// bijection/outputs.ts
export const inventoryOutput = defineOutput({ row: inventoryRow, key: ["sku"] });

// bijection/schema.ts
export default defineSchema(
  {
    inventory: defineSyncedTable({source: inventoryOutput})
      .access(readAccess),
  },
  { access },
);