basin

cloudflare/skills/basin

Build and troubleshoot Cloudflare Basin analytics workflows with Basin Pipelines, Basin Catalog, and Basin SQL. Use for streaming data into R2 Iceberg tables, managing catalogs, or querying those tables; also use for requests using the former Data Platform, Pipelines, R2 Data Catalog, or R2 SQL names.

インストール · 854出典を見る

Installation

npx skills add https://github.com/cloudflare/skills --skill basin

スキルファイル

SKILL.md

最終同期 · 2026/10/23

references/catalog/api.md›

# Basin Catalog API Selection

Use the Iceberg REST catalog through an engine for table reads and writes; use the Cloudflare control-plane API for catalog administration. Copy the catalog connection values from the actual environment as described in [configuration](configuration.md).

| Task | Documentation |
|------|---------------|
| Enable or disable catalogs; inspect status, credentials, namespaces, tables, and maintenance configuration | [Basin Catalog control-plane API](https://developers.cloudflare.com/api/resources/r2_data_catalog/index.md) — select the affected operation for its schema, pagination, and namespace encoding |
| Connect and create tables through Python | [PyIceberg configuration](https://developers.cloudflare.com/basin-catalog/config-examples/pyiceberg/index.md) |
| Connect, create, write, and query through Spark | [PySpark configuration](https://developers.cloudflare.com/basin-catalog/config-examples/spark-python/index.md) |
| Plan automatic compaction and snapshot expiration | [Table maintenance](https://developers.cloudflare.com/basin-catalog/table-maintenance/index.md) |
| Delete rows, tables, or associated files | [Deleting data](https://developers.cloudflare.com/basin-catalog/deleting-data/index.md) |

For engine-specific operations beyond these Cloudflare examples, follow the upstream engine documentation linked from the relevant configuration guide and check the installed version. Do not infer engine method signatures from the control-plane API.

## Get Table (repository-specific metadata introspection note)

This existing repository note is retained because the published control-plane API reference does not document this operation or its snapshot-pruning response. Verify availability and response behavior against the target service or authoritative implementation before relying on it; it is not a documented API guarantee. Do not substitute the documented list-tables response for this metadata response.

`GET /namespaces/{ns}/tables/{table}` returns schema, partition spec, sort order, and snapshot info — like Iceberg "load table" but on the control plane, with snapshots pruned to the most recent 10.

```bash
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/live/tables/earthquakes" \
  -H "Authorization: Bearer $API_TOKEN"
```

```json
{"result": {
  "identifier": {"namespace": ["live"], "name": "earthquakes"},
  "table_uuid": "019edccf-3ac8-73e3-...",
  "metadata_location": "s3://live-data/__r2_data_catalog/.../metadata/01225-....metadata.json",
  "total_snapshots": 1225,
  "returned_snapshots": 10,
  "metadata": { /* standard Iceberg TableMetadata: schemas, partition-specs, sort-orders,
                   properties, current-snapshot-id, snapshots (≤10), snapshot-log, refs */ }
}, "success": true}
```

| Field | Description |
|-------|-------------|
| `identifier` | `{namespace: [...], name}` |
| `table_uuid` | Iceberg table UUID |
| `metadata_location` | R2 path to current metadata file |
| `total_snapshots` | Total before pruning |
| `returned_snapshots` | Count in `metadata.snapshots` (max 10) |
| `metadata` | Standard [Iceberg TableMetadata](https://iceberg.apache.org/spec/#table-metadata-fields), arrays pruned to 10 |

See [patterns](patterns.md) for engine selection and [troubleshooting](gotchas.md) for diagnosis.

references/catalog/configuration.md›

# Basin Catalog Configuration

Inspect the existing bucket, catalog, engine versions, and credential configuration before making changes. Use the project's installed tools and preserve its environment-variable or secret-management conventions.

| Task | Documentation |
|------|---------------|
| Enable a catalog and obtain connection details | [Enable Basin Catalog](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#enable-r2-data-catalog-on-a-bucket) |
| Select credentials for readers, writers, or maintenance | [Authenticate your Iceberg engine](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#authenticate-your-iceberg-engine) |
| Configure compaction and its service credential | [Enable compaction](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#enable-compaction) |
| Configure snapshot retention | [Enable snapshot expiration](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#enable-snapshot-expiration) |
| Choose file sizes, retention policy, and maintenance scope | [Table maintenance](https://developers.cloudflare.com/basin-catalog/table-maintenance/index.md) |
| Connect a Python client | [PyIceberg](https://developers.cloudflare.com/basin-catalog/config-examples/pyiceberg/index.md) |
| Connect Spark | [PySpark](https://developers.cloudflare.com/basin-catalog/config-examples/spark-python/index.md) |
| Connect another query engine | [Engine configuration guides](https://developers.cloudflare.com/basin-catalog/config-examples/index.md) |
| Disable catalog access | [Disable Basin Catalog](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#disable-r2-data-catalog-on-a-bucket) |

Copy the Catalog URI and Warehouse name exactly from the catalog detail page or Wrangler enable output. Scope both catalog and storage permissions to the operations the client needs; readers do not need a blanket write-enabled token. Treat maintenance credentials separately from reader credentials. Verify connectivity with a read operation before attempting writes, then check catalog and credential status through the [control-plane API](api.md).

references/catalog/gotchas.md›

# Basin Catalog Troubleshooting

Identify whether the failure occurs in catalog administration, engine metadata access, or underlying object access before changing permissions or client settings.

| Check | Documentation |
|-------|---------------|
| Catalog enablement, Catalog URI, or Warehouse mismatch | [Manage catalogs](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md) |
| Reader/writer token scope or file-access denial | [Engine authentication](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#authenticate-your-iceberg-engine) — inspect both catalog and storage permissions |
| Missing maintenance credentials or wrong table/catalog configuration | [Control-plane API](https://developers.cloudflare.com/api/resources/r2_data_catalog/index.md) and [enable compaction](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md#enable-compaction) |
| Compaction backlog, retention, or orphaned files | [Table maintenance](https://developers.cloudflare.com/basin-catalog/table-maintenance/index.md) |
| PyIceberg connection or table creation | [PyIceberg configuration](https://developers.cloudflare.com/basin-catalog/config-examples/pyiceberg/index.md) |
| Spark dependency, credential-vending, or signing configuration | [PySpark configuration](https://developers.cloudflare.com/basin-catalog/config-examples/spark-python/index.md) |
| Deleted data is still present | [Deleting data](https://developers.cloudflare.com/basin-catalog/deleting-data/index.md) |
| Catalog request or maintenance-job diagnosis | [Metrics and analytics](https://developers.cloudflare.com/basin-catalog/observability/metrics/index.md) |

Compare the client's configured URI and warehouse with the actual catalog values. Test a read operation first; do not grant write access merely to resolve a reader's failure. For schema or concurrency errors, inspect the installed engine's behavior and current table metadata before retrying. The [get-table note](api.md#get-table-repository-specific-metadata-introspection-note) is not a substitute for verifying the service's response contract.

See [configuration](configuration.md) and [patterns](patterns.md) for implementation choices.

references/catalog/patterns.md›

# Basin Catalog Patterns

Choose the engine based on the project's existing runtime and workload, then retrieve its current connection example.

| Need | Starting point |
|------|----------------|
| Python catalog operations and ingestion without a Spark deployment | [PyIceberg](https://developers.cloudflare.com/basin-catalog/config-examples/pyiceberg/index.md) |
| Existing Spark ETL and distributed table processing | [PySpark](https://developers.cloudflare.com/basin-catalog/config-examples/spark-python/index.md) |
| Connect an existing SQL engine | [Engine configuration guides](https://developers.cloudflare.com/basin-catalog/config-examples/index.md) |
| Query through Cloudflare's serverless SQL service | [Basin SQL](../sql/) |
| Stream events into tables | [Basin Pipelines patterns](../pipelines/patterns.md) |

Use the discovered Catalog URI and Warehouse name from [configuration](configuration.md). Match dependencies to the installed engine and the current guide instead of adopting a universal pinned Spark/Iceberg combination.

Plan ingestion, query, and maintenance responsibilities together. Prefer [automatic table maintenance](https://developers.cloudflare.com/basin-catalog/table-maintenance/index.md) when it meets the workload; align retention with time-travel needs before enabling expiration. For engine-specific partitioning, schema evolution, or manual procedures, consult that engine's linked upstream documentation and verify behavior on representative data.

When multiple writers share a table, design recovery around the actual failed operation and the engine's commit semantics. Reproduce conflicts and ensure retries do not duplicate application work. See [API selection](api.md) and [troubleshooting](gotchas.md).

references/catalog/README.md›

# Basin Catalog

Use Basin Catalog for Iceberg analytics and data pipelines on object storage. For transactional application queries, consider a database; for unstructured objects, use [R2](https://developers.cloudflare.com/r2/index.md).

Distinguish the Iceberg REST catalog used by query engines from Cloudflare's control-plane API for catalog administration. Start with the workflow you need:

| Task | Reference |
|------|-----------|
| Enable a catalog, discover connection values, and choose credentials | [Configuration](configuration.md) |
| Select administration or engine APIs | [API selection](api.md) |
| Choose a Python, Spark, or SQL workflow | [Patterns](patterns.md) |
| Diagnose authentication, maintenance, or client problems | [Troubleshooting](gotchas.md) |

Copy the actual **Catalog URI** and **Warehouse name** from the catalog detail page or Wrangler's enable output. Pass both to the selected engine; do not reconstruct them from an assumed bucket naming convention. Retrieve [Manage catalogs](https://developers.cloudflare.com/basin-catalog/manage-catalogs/index.md) before setup or permission changes.

Related workflows: [Basin Pipelines](../pipelines/) for ingest and [Basin SQL](../sql/) for querying tables.

references/pipelines/api.md›

# Basin Pipelines API Reference

Code templates and verified behavior. For the full SQL function set and HTTP status semantics, pull `https://developers.cloudflare.com/basin-pipelines/sql-reference/index.md` and the streams docs.

## Worker Binding Interface

```typescript
// from cloudflare:pipelines / @cloudflare/workers-types
interface Pipeline<T = any> { send(records: T[]): Promise<void>; }

interface Env { MY_STREAM: Pipeline; }

export default {
  async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    await env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]);
    return new Response("OK");
  }
} satisfies ExportedHandler<Env>;
```

- `send()` takes an **array**, returns `Promise<void>` (no confirmation payload).
- Throws on network errors — wrap in try/catch or use `ctx.waitUntil()` for fire-and-forget.
- Validation errors are **not** thrown here (deferred during processing — see [gotchas.md](gotchas.md)).
- Payload/rate limits apply — check `https://developers.cloudflare.com/basin-pipelines/platform/limits/index.md` before sizing batches.

## HTTP Ingest

```
https://{stream-id}.ingest.cloudflare.com
```

Get `{stream-id}` from `npx wrangler basin pipelines streams list`.

```bash
# Batch (preferred)
curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[{"event_id":"evt-1","amount":29.99},{"event_id":"evt-2","amount":14.99}]'

# Single event — auto-wrapped in an array
curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" -d '{"event_id":"evt-3","amount":9.99}'
```

If stream auth is enabled, add `-H "Authorization: Bearer $TOKEN"` (token needs **Workers Pipelines Send**). Standard HTTP status codes apply (400 invalid, 401 auth, 413 too large, 429 rate-limited, 5xx retry).

> **JSON only** — no Avro, Protobuf, or CSV input.

## REST Management API

Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`

```bash
# List
curl -s "$BASE_URL/streams"   -H "Authorization: Bearer $API_TOKEN"
curl -s "$BASE_URL/sinks"     -H "Authorization: Bearer $API_TOKEN"
curl -s "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN"

# Get one (pipeline GET includes status + failure_reason — useful for debugging)
curl -s "$BASE_URL/pipelines/{pipeline-id}" -H "Authorization: Bearer $API_TOKEN"

# Delete in reverse order: pipeline → sink → stream
curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/sinks/{id}"     -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/streams/{id}"   -H "Authorization: Bearer $API_TOKEN"
```

> `wrangler basin pipelines delete` defaults to "no" non-interactively — use the REST API for automated cleanup. Deleting a stream removes buffered events and dependent pipelines.

### Pipeline Lifecycle States

| Status | Meaning |
|--------|---------|
| `running` | Active, processing events |
| `initializing` | Starting up (minutes after creation or recovery) |
| `failed` | Stopped on error — check `failure_reason` (expired token, deleted bucket, disabled catalog) |

> A `GET` on a sink shows `schema.fields: []` — expected. The sink inherits schema from the stream via the pipeline SQL.

## Pipeline SQL (Transforms)

Row-level only — no GROUP BY/aggregation. CTEs (`WITH`) and `UNNEST` are supported. Full function list: `https://developers.cloudflare.com/basin-pipelines/sql-reference/index.md`.

```sql
-- Passthrough / filter / enrich
INSERT INTO my_sink SELECT * FROM my_stream;
INSERT INTO my_sink SELECT * FROM my_stream WHERE amount > 10;
INSERT INTO my_sink
SELECT event_id, UPPER(category) AS category, amount * 1.1 AS amount_with_tax
FROM my_stream;

-- CTE
WITH filtered AS (SELECT event_id, amount FROM my_stream WHERE amount > 50)
INSERT INTO my_sink SELECT * FROM filtered;

-- UNNEST arrays (one per SELECT)
SELECT UNNEST(tags) AS tag FROM my_stream;
```

Supported categories: string, regex, hashing (`sha256`), JSON extraction, timestamp conversion, conditional (`CASE`), `CAST`, `COALESCE`, math/comparison operators.

## Verifying End-to-End Data Flow

```bash
# 1. Pipeline running (not initializing/failed)?
curl -s "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"

# 2. Table created yet? (3–7 min on first flush)
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/my_ns/tables" \
  -H "Authorization: Bearer $API_TOKEN"

# 3. Data present? (Basin SQL)
curl -s -X POST \
  "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/basin-sql/query/$BUCKET" \
  -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
  -d '{"query": "SELECT COUNT(*) AS total FROM my_ns.my_table"}'
```

> Expect **3–7 minutes** from first send to first queryable data. Subsequent flushes are much faster.

## See Also

- [configuration.md](configuration.md) — creating resources · [patterns.md](patterns.md) — producers, Logpush, observability
- [sql/api.md](../sql/api.md) — querying results

references/pipelines/configuration.md›

# Basin Pipelines Configuration

Templates for creating streams, sinks, and pipelines via CLI, REST, or Terraform. For the full flag/field list and allowed values, pull `https://developers.cloudflare.com/basin-pipelines/reference/wrangler-commands/index.md` and the streams/sinks/pipelines docs.

## Naming Rules

- **Streams, sinks, pipelines** use underscores: `my_stream`, `my_sink`, `my_pipeline`.
- **Buckets** use hyphens: `my-bucket`.

## Schema (Structured Streams)

Schema is a JSON object with a `fields` array; each field has `name`, `type`, `required`.

```json
{
  "fields": [
    { "name": "event_id", "type": "string", "required": true },
    { "name": "amount", "type": "float64", "required": false }
  ]
}
```

Field types include `string`, `bool`, `int32/64`, `float32/64`, `timestamp`, `json`, `binary`, `list`, `struct` (with nested `items`/`fields`). For the authoritative type list, see `https://developers.cloudflare.com/basin-pipelines/sql-reference/sql-data-types/index.md`.

Unstructured streams (no schema) store everything in a single `value` column.

> Basin Pipelines auto-adds `__ingest_ts` (TIMESTAMP, day-partitioned). Do **not** include it in your schema.

## Option A: Interactive (Simplest)

```bash
npx wrangler basin pipelines setup   # creates stream + sink + pipeline, optionally bucket + catalog
```

## Option B: Wrangler CLI (Explicit)

```bash
# 1. Stream
npx wrangler basin pipelines streams create my_stream --schema-file schema.json

# 2. Sink — Basin Catalog (Iceberg). Creates the namespace + table.
npx wrangler basin pipelines sinks create my_sink \
  --type basin-catalog \
  --bucket my-bucket --namespace my_namespace --table my_table \
  --catalog-token $API_TOKEN \
  --compression zstd --roll-interval 300

# 2b. Sink — R2 raw Parquet (alternative)
npx wrangler basin pipelines sinks create my_sink \
  --type r2 --bucket my-bucket --format parquet \
  --path analytics/events --partitioning "year=%Y/month=%m/day=%d" \
  --access-key-id $KEY --secret-access-key $SECRET

# 3. Pipeline (SQL connects stream → sink)
npx wrangler basin pipelines create my_pipeline \
  --sql "INSERT INTO my_sink SELECT * FROM my_stream"
```

Tuning knobs (`--compression`, `--roll-interval`, `--roll-size`, etc.) and their allowed values/defaults change — pull the wrangler-commands and sinks docs rather than hardcoding. Rule of thumb: prod `--roll-interval 300+`, dev `10` (creates many small files).

> **⚠️ Basin Pipelines are immutable.** SQL, schema, and sink config can't be changed — delete and recreate.

## Option C: REST API (Programmatic)

Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`

```bash
# Stream
curl -X POST "$BASE_URL/streams" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" -d '{
    "name": "my_stream",
    "http": {"enabled": true, "authentication": false},
    "schema": {"fields": [{"name": "event_id", "type": "string", "required": true}]}
  }'

# Sink — NOTE REST field names differ from CLI flags (see table)
curl -X POST "$BASE_URL/sinks" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" -d '{
    "name": "my_sink", "type": "r2_data_catalog",
    "config": {"bucket": "my-bucket", "namespace": "my_namespace",
               "table_name": "my_table", "token": "'$API_TOKEN'",
               "rolling_policy": {"interval_seconds": 300}},
    "format": {"type": "parquet"}
  }'

# Pipeline
curl -X POST "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "my_pipeline", "sql": "INSERT INTO my_sink SELECT * FROM my_stream;"}'
```

**REST field names ≠ CLI flags** (common failure — not obvious from docs):

| REST (config body) | CLI flag | Gotcha |
|--------------------|----------|--------|
| `"type": "r2_data_catalog"` | `--type basin-catalog` | REST retains the legacy identifier; the new CLI value is `basin-catalog` |
| `"table_name"` | `--table` | different key |
| `"token"` | `--catalog-token` | different key |
| `"format": {"type": "parquet"}` | (implied) | required in REST, omitted in CLI |

## Worker Binding

```jsonc
// wrangler.jsonc
{ "pipelines": [ { "stream": "<STREAM_ID>", "binding": "MY_STREAM" } ] }
```

> Binding field is `"stream"` as of June 2026 (was `"pipeline"`, still accepted). Use the **stream ID** (`wrangler basin pipelines streams list`), not the pipeline ID. Redeploy after adding. Generate typed bindings with `npx wrangler types` → `Pipeline<Cloudflare.MyStreamRecord>` from `cloudflare:pipelines`.

## Terraform

Resources: `cloudflare_pipeline_stream`, `cloudflare_pipeline_sink`, `cloudflare_pipeline`. For current attribute schemas pull `https://developers.cloudflare.com/basin-pipelines/reference/terraform/index.md`.

```hcl
resource "cloudflare_pipeline_stream" "my_stream" {
  account_id     = var.cloudflare_account_id
  name           = "my_stream"
  format         = { type = "json" }
  schema         = { fields = [{ name = "value", type = "json", required = true }] }
  http           = { enabled = true, authentication = false, cors = {} }
  worker_binding = { enabled = false }
}

resource "cloudflare_pipeline_sink" "my_sink" {
  account_id = var.cloudflare_account_id
  name       = "my_sink"
  type       = "r2_data_catalog"
  format     = { type = "parquet" }
  schema     = { fields = [] }
  config     = {
    account_id = var.cloudflare_account_id
    bucket     = cloudflare_r2_bucket.pipeline_bucket.name
    table_name = "my_table"
    token      = var.catalog_token
  }
}

resource "cloudflare_pipeline" "my_pipeline" {
  account_id = var.cloudflare_account_id
  name       = "my_pipeline"
  sql        = "INSERT INTO ${cloudflare_pipeline_sink.my_sink.name} SELECT * FROM ${cloudflare_pipeline_stream.my_stream.name}"
}
```

## Credentials

| Type | Permission |
|------|------------|
| Catalog token (Iceberg sink) | R2 Storage Admin R&W + Basin Catalog R&W |
| R2 credentials (raw sink) | Object Read & Write |
| HTTP ingest token | Workers Pipelines Send (only if stream auth enabled) |

## See Also

- [api.md](api.md) — sending events, REST API, lifecycle · [gotchas.md](gotchas.md) — immutability, REST≠CLI

references/pipelines/gotchas.md›

# Basin Pipelines Gotchas

Non-obvious failure modes (not well covered by docs). For current limits and error semantics, pull `https://developers.cloudflare.com/basin-pipelines/platform/limits/index.md`.

## Events accepted but never appear (most common)

HTTP 200 / `send()` resolves, but no data in the sink. Causes:

1. **Schema validation failure** — structured streams accept then **silently drop** invalid events during processing. Validate client-side (Zod) and monitor `pipelinesUserErrorsAdaptiveGroups`.
2. **First-flush warm-up** — first data takes **3–7 minutes** (warm-up + namespace/table creation) even with `--roll-interval 10`. Poll ≥5 min in tests.
3. **Roll interval not elapsed** — default 300s.
4. **Silent sink failure** — deleted bucket or expired token. Check `recordsWritten > 0` but `filesWritten = 0`; inspect `failure_reason` via `GET /pipelines/{id}`.

## Everything is immutable

Cannot modify stream schema, pipeline SQL, or sink config — delete and recreate. Use version naming (`events_v1`) and keep SQL in version control.

```bash
curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/sinks/{id}"     -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/streams/{id}"   -H "Authorization: Bearer $API_TOKEN"
```

## Worker binding undefined (`env.MY_STREAM`)

1. Use the **stream ID**, not pipeline ID, in `wrangler.jsonc`.
2. Binding field is `"stream"` (June 2026); old `"pipeline"` still works.
3. Redeploy after adding the binding.

## REST API field names ≠ CLI flags

`r2_data_catalog` vs `--type basin-catalog`, `table_name` vs `--table`, `token` vs `--catalog-token`, and `format` is required in REST but implied in CLI. See [configuration.md](configuration.md#option-c-rest-api-programmatic).

## `wrangler basin pipelines delete` defaults to "no"

Non-interactive environments answer "no" automatically — use REST `DELETE` for CI/automation.

## Behavioral Notes

- **`__ingest_ts` auto-added** (TIMESTAMP, day-partitioned). Don't put it in your schema.
- **Sinks can't target existing tables** — the sink creates its own. Use PySpark to write to existing tables.
- **JSON-only input** — no Avro/Protobuf/CSV.
- **Naming:** streams/sinks/pipelines use underscores; buckets use hyphens.
- **Metrics lag 5–10 min** after creation.
- **Pipeline SQL is row-level only** — no GROUP BY/aggregation/window functions (do aggregation in [Basin SQL](../sql/) at query time). CTEs and `UNNEST` are supported.

## Debug Checklist

- [ ] Stream exists: `wrangler basin pipelines streams list`
- [ ] Pipeline `running` (not `initializing`/`failed`): `GET /pipelines/{id}`, check `failure_reason`
- [ ] SQL matches schema; sink token valid; bucket + catalog exist
- [ ] Worker redeployed; binding uses **stream ID** under `"stream"`
- [ ] Waited ≥5 min (first flush)
- [ ] Sink metrics: `filesWritten > 0`; error metrics show no drops

## See Also

- [configuration.md](configuration.md) · [api.md](api.md) · [patterns.md](patterns.md)

references/pipelines/patterns.md›

# Basin Pipelines Patterns

Code-first patterns. For observability dataset/field schemas and Logpush dataset lists, pull `https://developers.cloudflare.com/basin-pipelines/observability/metrics/index.md` and `https://developers.cloudflare.com/basin-pipelines/streams/logpush/index.md`.

## Fire-and-Forget Producer

```typescript
export default {
  async fetch(req, env, ctx) {
    const event = { event_id: crypto.randomUUID(), event_type: "page_view", timestamp: new Date().toISOString() };
    ctx.waitUntil(env.MY_STREAM.send([event]));  // don't block the response
    return new Response("OK");
  }
};
```

## Client-Side Validation with Zod

Structured streams drop invalid events silently during processing. Validate before sending for immediate feedback.

```typescript
import { z } from "zod";

const EventSchema = z.object({
  event_id: z.string(),
  category: z.enum(["purchase", "view"]),
  amount: z.number().positive().optional(),
});

const validated = EventSchema.parse(rawEvent);  // throws synchronously
await env.MY_STREAM.send([validated]);
```

## Scheduled Collector Worker

```jsonc
// wrangler.jsonc
{
  "name": "collector",
  "pipelines": [{ "stream": "<STREAM_ID>", "binding": "EVENT_STREAM" }],
  "triggers": { "crons": ["*/5 * * * *"] }
}
```

```typescript
export default {
  async scheduled(event, env, ctx) {
    const items = await (await fetch("https://api.example.com/data")).json();
    const events = items.map(i => ({
      event_id: crypto.randomUUID(),
      timestamp: new Date().toISOString(),
      category: i.type, amount: i.value,
    }));
    await env.EVENT_STREAM.send(events);
  },
};
```

## Logpush → Basin Pipelines

Basin Pipelines is a native Logpush destination — ingest Cloudflare logs, transform with SQL, store as Iceberg/Parquet. For the current supported dataset list and field names, pull the Logpush doc above.

```sql
INSERT INTO http_logs_sink
SELECT
  ClientIP,
  EdgeResponseStatus,
  to_timestamp_micros(EdgeStartTimestamp) AS event_time,
  upper(ClientRequestMethod) AS method,
  sha256(ClientIP) AS hashed_ip          -- redact PII at ingest
FROM http_logs_stream
WHERE EdgeResponseStatus >= 400;
```

Configure via Dashboard (**Logpush → Create a job → Basin Pipelines** destination) or API.

## Basin Pipelines + Queues Fan-out

```typescript
await Promise.all([
  env.ANALYTICS_STREAM.send([event]),  // long-term storage + SQL
  env.PROCESS_QUEUE.send(event),       // immediate processing + retries
]);
```

Use Basin Pipelines for long-term storage + SQL; Queues for immediate processing/retries/DLQ; both for fan-out.

## Observability (GraphQL Analytics)

Same R2 API token works. Endpoint: `https://api.cloudflare.com/client/v4/graphql`. Datasets cover ingestion, processing (incl. `decodeErrors`), delivery, sink writes (`filesWritten`), and user/validation errors — see the metrics doc for the full dataset/field catalog.

```bash
curl -X POST "https://api.cloudflare.com/client/v4/graphql" \
  -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
  -d '{"query": "query { viewer { accounts(filter: {accountTag: \"'$ACCOUNT_ID'\"}) { pipelinesIngestionAdaptiveGroups(filter: {pipelineId: \"PIPELINE-UUID-WITH-DASHES\", datetime_geq: \"2026-03-01T00:00:00Z\"}, limit: 10) { sum { ingestedRecords ingestedBytes } dimensions { datetimeHour } } } } }"}'
```

> **Sink/pipeline IDs need dashes for GraphQL** but wrangler may show them without: `b909fe6e544844abbd63f6dcbc81d602` → `b909fe6e-5448-44ab-bd63-f6dcbc81d602`. Metrics take 5–10 min to populate.

### Detecting Silent Data Loss

If a sink's bucket is deleted or its token expires, events are accepted but lost. Tell-tale: `recordsWritten > 0` but `filesWritten = 0`. Always verify data lands in R2 within the roll interval and Basin SQL returns expected counts.

## Schema Evolution (Immutable Basin Pipelines)

Basin Pipelines can't change. Version + dual-write:

```bash
npx wrangler basin pipelines streams create events_v2 --schema-file v2.json
```
```typescript
await Promise.all([env.EVENTS_V1.send([event]), env.EVENTS_V2.send([event])]);
// query across versions with UNION ALL in Basin SQL
```

## End-to-End: Streaming Analytics Dashboard

```
External APIs → Collector Worker (cron) → Pipeline → R2 (Iceberg) → Dashboard Worker → Basin SQL
```

1. Create bucket + enable catalog ([Basin Catalog](../catalog/configuration.md))
2. Create stream + sink + pipeline (here)
3. Collector Worker with cron + stream binding (above)
4. Dashboard Worker querying Basin SQL ([sql/patterns.md](../sql/patterns.md))
5. Enable automatic compaction

## See Also

- [configuration.md](configuration.md) · [api.md](api.md) · [gotchas.md](gotchas.md) · [Basin SQL](../sql/)

references/pipelines/README.md›

# Basin Pipelines

Streaming ingest: receive events over HTTP/Workers/Logpush, transform with SQL, write to R2 as Iceberg tables or Parquet/JSON files.

## Documentation

This reference is a fast-start with verified code and gotchas. For limits, settings, full SQL syntax, and pricing, **retrieve the live docs** — use the Cloudflare MCP `docs` tool if available, otherwise `webfetch` the URL. Docs are source of truth over this file.

| Topic | URL |
|-------|-----|
| Overview / getting started | `https://developers.cloudflare.com/basin-pipelines/getting-started/index.md` |
| Streams (write, manage, Logpush) | `https://developers.cloudflare.com/basin-pipelines/streams/index.md` |
| Sinks | `https://developers.cloudflare.com/basin-pipelines/sinks/index.md` |
| Basin Pipelines & SQL transforms | `https://developers.cloudflare.com/basin-pipelines/pipelines/index.md` |
| SQL reference (statements, types) | `https://developers.cloudflare.com/basin-pipelines/sql-reference/index.md` |
| Wrangler commands | `https://developers.cloudflare.com/basin-pipelines/reference/wrangler-commands/index.md` |
| Terraform | `https://developers.cloudflare.com/basin-pipelines/reference/terraform/index.md` |
| Limits | `https://developers.cloudflare.com/basin-pipelines/platform/limits/index.md` |
| Pricing | `https://developers.cloudflare.com/basin-pipelines/platform/pricing/index.md` |
| Metrics (GraphQL) | `https://developers.cloudflare.com/basin-pipelines/observability/metrics/index.md` |

## Three Components

```
Sources → Stream → Pipeline (SQL) → Sink → R2
          ↑          ↓                 ↓
   HTTP / Workers / Transform     Iceberg (Basin Catalog)
   Logpush          (row-level)   or Parquet/JSON files
```

| Component | Purpose |
|-----------|---------|
| **Stream** | Receives events (HTTP endpoint, Worker binding, or Logpush). Structured (schema-validated) or unstructured. |
| **Pipeline** | SQL connecting a stream to a sink. Row-level transforms only — no GROUP BY/aggregation. |
| **Sink** | Writes to R2 — Iceberg via Basin Catalog, or raw Parquet/JSON. |

**Status:** The linked docs PR proposes Basin general availability. Verify current availability, limits, and pricing in the live docs before making claims or sizing workloads.

## Quick Start

```bash
# Interactive — creates stream + sink + pipeline, optionally bucket + catalog
npx wrangler basin pipelines setup
```

Minimal Worker producer:
```typescript
interface Env { MY_STREAM: Pipeline; }

export default {
  async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    ctx.waitUntil(env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]));
    return new Response("OK");
  }
} satisfies ExportedHandler<Env>;
```

## Which Sink Type?

```
Need SQL queries / ACID / time-travel on the data?
  → Basin Catalog (Iceberg)   ✅ Basin SQL, schema evolution   ❌ more setup

Just archival / external tools (Spark, Athena)?
  → R2 raw files (Parquet/JSON) ✅ simple, partitioned files  ❌ no built-in SQL
```

## Critical Behaviors (read before building)

These are non-obvious and prevent most failures — see [gotchas.md](gotchas.md) for detail.

- **Everything is immutable after creation** — stream schema, pipeline SQL, sink config. To change, delete and recreate.
- **Sinks create their own table** — they cannot target an existing Iceberg table.
- **`__ingest_ts` is added automatically** (TIMESTAMP, partitioned by day). Don't define it in your schema.
- **Data isn't queryable immediately** — first flush takes **3–7 minutes** (warm-up + table creation) even with a short roll interval.
- **Schema validation is deferred** — invalid events are accepted then silently dropped. Monitor via GraphQL error metrics.
- **Binding field renamed `pipeline` → `stream`** (June 2026); old field still accepted.

## Reading Order

1. [configuration.md](configuration.md) — schema, streams, sinks, pipelines (CLI + REST + Terraform), bindings
2. [api.md](api.md) — `send()`, HTTP ingest, REST API, pipeline SQL, lifecycle states
3. [patterns.md](patterns.md) — fire-and-forget, validation, Logpush, observability, end-to-end
4. [gotchas.md](gotchas.md) — silent drops, immutability, REST≠CLI field names

## See Also

- [Basin Catalog](../catalog/) — Iceberg sink destination
- [Basin SQL](../sql/) — query the ingested data
- [r2](https://developers.cloudflare.com/r2/index.md) · [queues](https://developers.cloudflare.com/queues/index.md) · [workers](https://developers.cloudflare.com/workers/index.md)

references/sql/api.md›

# Basin SQL API Reference

Read-only SQL over Iceberg (Apache DataFusion). Query templates only. For the authoritative list of supported syntax, functions, data types, and limitations, pull the SQL reference (`sql-reference/`, `.../aggregate-functions/`, `.../scalar-functions/`, `.../complex-types/`) and `reference/limitations-best-practices/`.

## Query Endpoint

```
POST https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/basin-sql/query/{BUCKET}
Authorization: Bearer <token>
Content-Type: application/json
Body: {"query": "<SQL>"}
```

CLI: `npx wrangler basin sql query "{WAREHOUSE}" "<SQL>"` (with `WRANGLER_BASIN_SQL_AUTH_TOKEN`).

## Response Format

```json
{
  "result": {
    "request_id": "dqe-prod-01...",
    "schema": [{"name": "cnt", "descriptor": {"type": {"name": "int64"}, "nullable": false}}],
    "rows": [{"category": "Electronics", "cnt": 12345}],
    "metrics": {"r2_requests_count": 5, "files_scanned": 29, "bytes_scanned": 12345678, "cache_hits": 0}
  },
  "success": true, "errors": []
}
```

Error: `{"result": null, "success": false, "errors": [{"code": 40003, "message": "..."}]}`. `bytes_scanned` ≈ billable data.

## Query Structure

```sql
SELECT [DISTINCT] columns | expressions | aggregations
FROM namespace.table [alias]
[ [INNER|LEFT|RIGHT|FULL OUTER|CROSS] JOIN namespace.table2 alias2 ON ... ]
[WHERE ...] [GROUP BY ...] [HAVING ...]
[QUALIFY window_predicate]
[ORDER BY expr [ASC|DESC]]
[LIMIT n]                          -- default 500, max 10,000
```

## Schema Discovery

```sql
SHOW DATABASES;            -- list namespaces (aliases: SHOW NAMESPACES / SHOW SCHEMAS)
SHOW TABLES IN namespace;
DESCRIBE namespace.table;  -- columns, types, partition keys
EXPLAIN [FORMAT JSON] SELECT ...;   -- execution plan (free; no data scanned)
```

## JOINs / Subqueries / CTEs / Set Ops

```sql
-- JOINs: all types + multi-way
SELECT z.domain, COUNT(*) AS cnt
FROM ns.zones z
INNER JOIN ns.http_requests h ON z.zone_id = h.zone_id
LEFT  JOIN ns.firewall_events f ON z.zone_id = f.zone_id
GROUP BY z.domain ORDER BY cnt DESC LIMIT 20;

-- Subqueries: IN / EXISTS / scalar / derived
SELECT * FROM ns.t1 WHERE id IN (SELECT id FROM ns.t2 WHERE x > 0);
SELECT col, (SELECT COUNT(*) FROM ns.t2 s WHERE s.id = t.id) AS cnt FROM ns.t1 t;

-- Multi-table CTE with JOIN
WITH top AS (SELECT zone_id, COUNT(*) AS req FROM ns.http_requests GROUP BY zone_id ORDER BY req DESC LIMIT 50)
SELECT t.zone_id, t.req FROM top t LEFT JOIN ns.zones z ON t.zone_id = z.zone_id;

-- Set ops: UNION / UNION ALL / INTERSECT / EXCEPT
SELECT zone_id FROM ns.firewall_events WHERE action = 'block'
UNION SELECT zone_id FROM ns.http_requests WHERE risk_score > 0.8;
```

## Window Functions

Use inline `OVER (...)`. See the SQL reference for the full list of supported window functions and frame syntax.

```sql
SELECT event_id,
       ROW_NUMBER() OVER (PARTITION BY mag_type ORDER BY magnitude DESC) AS rn,
       LAG(magnitude, 2, 0.0) OVER (ORDER BY occurred_at) AS prev2,   -- offset + default
       NTH_VALUE(magnitude, 2) OVER (ORDER BY magnitude DESC) AS n2,
       SUM(magnitude) OVER (ORDER BY occurred_at) AS running,
       AVG(magnitude) OVER (ORDER BY magnitude ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS moving_avg
FROM ns.earthquakes;

-- QUALIFY: filter on a window result (top row per partition)
SELECT event_id, mag_type, magnitude FROM ns.earthquakes
QUALIFY ROW_NUMBER() OVER (PARTITION BY mag_type ORDER BY magnitude DESC) = 1;
```

## Functions

Aggregate, scalar, JSON, and array/map function catalogs are in the docs — pull `sql-reference/aggregate-functions/` and `.../scalar-functions/`. JSON functions accept variadic paths, e.g. `json_get_int(doc, 'user', 'profile', 'level')`.

## Data Types

`integer`, `float`, `string` (single quotes), `boolean`, `timestamp` (RFC3339 **with timezone**), `date` (ISO 8601), `struct`, `array` (1-indexed), `map`. No implicit conversions — quote strings, include timezone on timestamps, don't quote integers. Full type docs: `sql-reference/`.

```sql
WHERE status = 200 AND method = 'GET'              -- not '200', not GET
  AND ts >= '2026-01-01T00:00:00Z'                 -- not '2026-01-01'
```

## Complex Types (quick examples; full ref in docs)

```sql
SELECT pricing['price'] AS price, get_field(pricing, 'discount') AS disc FROM ns.t;  -- struct
SELECT tags[1] AS first_tag, array_length(tags) AS n FROM ns.t;                       -- array (1-indexed)
SELECT map_keys(meta), map_extract(meta, 'source') FROM ns.t;                         -- map
```

## Errors

Failed queries return `{"success": false, "errors": [{"code": ..., "message": ...}]}`. For error codes and troubleshooting, see `https://developers.cloudflare.com/basin-sql/troubleshooting/index.md`.

## See Also

- [patterns.md](patterns.md) — query examples · [gotchas.md](gotchas.md) — limits & workarounds · [configuration.md](configuration.md)

references/sql/configuration.md›

# Basin SQL Configuration

Auth and setup. For the current permission matrix and wrangler flags, pull `https://developers.cloudflare.com/basin-sql/reference/wrangler-commands/index.md` and the Basin Catalog manage-catalogs doc.

## Prerequisites

- R2 bucket with Basin Catalog enabled ([catalog/configuration.md](../catalog/configuration.md))
- R2 API token: **R2 Storage Admin Read & Write** (includes Basin SQL Read), or add **Basin SQL Read** explicitly
- Wrangler CLI (for CLI queries)

> Open-beta limitation: R2 Storage **Admin Read & Write is required even for read-only Basin SQL queries**.

## Enable Catalog + Get Warehouse

```bash
npx wrangler basin catalog enable my-bucket
```

You query by **warehouse** name (`{ACCOUNT_ID}_{BUCKET}`), shown in the output alongside the Catalog URI.

## Configure Auth

### Wrangler CLI

```bash
export WRANGLER_BASIN_SQL_AUTH_TOKEN=<your-token>
# or a .env file in the project dir (auto-loaded): WRANGLER_BASIN_SQL_AUTH_TOKEN=<your-token>
```

> Wrangler does **not** use the `wrangler login` OAuth session for Basin SQL — the env var is required.

### REST API

```bash
curl -X POST \
  "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/basin-sql/query/$BUCKET" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"query": "SELECT * FROM default.my_table LIMIT 10"}'
```

## Verify Setup

```bash
npx wrangler basin sql query "${ACCOUNT_ID}_my-bucket" "SHOW DATABASES"
npx wrangler basin sql query "${ACCOUNT_ID}_my-bucket" "SHOW TABLES IN default"
```

## See Also

- [api.md](api.md) — SQL syntax · [patterns.md](patterns.md) — query examples · [gotchas.md](gotchas.md) — troubleshooting

references/sql/gotchas.md›

# Basin SQL Gotchas

Operational pitfalls. For the authoritative list of supported features, unsupported features, and recommended workarounds, pull `https://developers.cloudflare.com/basin-sql/reference/limitations-best-practices/index.md` and `https://developers.cloudflare.com/basin-sql/troubleshooting/index.md`.

## Access

- **No Workers binding.** There is no `env.R2_SQL`. Query the REST endpoint via `fetch()` from a Worker ([patterns.md](patterns.md#dashboard-worker)), or use D1 / an external DB for OLTP.
- Wrangler needs `WRANGLER_BASIN_SQL_AUTH_TOKEN` — it does **not** reuse the `wrangler login` OAuth session.
- R2 Storage **Admin Read & Write is currently required even for read-only** queries. Verify this in the live authentication docs.

## Type Safety

```sql
-- ❌ wrong                          -- ✅ right
WHERE status = '200'                 WHERE status = 200
WHERE ts > '2026-01-01'              WHERE ts > '2026-01-01T00:00:00Z'   -- need time + tz
WHERE method = GET                   WHERE method = 'GET'
```

No implicit conversions. Timestamps must be RFC3339 with timezone; dates ISO 8601.

## Performance

- **File count dominates latency** — enable automatic compaction.
- **Partition-filter + narrow time windows + always `LIMIT`.**
- **Multi-way JOINs on large tables** can exceed resource limits — filter heavily, join through dimension tables.
- Per-query `metrics` (`files_scanned`, `bytes_scanned`, `cache_hits`) are the primary observability signal; `bytes_scanned` ≈ billable data. For LIMIT bounds, pagination, and other guidance, see the limitations-best-practices doc.

## Debug Checklist

1. `wrangler basin catalog enable <bucket>` — catalog on?
2. `echo $WRANGLER_BASIN_SQL_AUTH_TOKEN` — token set?
3. `SHOW DATABASES` → `SHOW TABLES IN ns` → `DESCRIBE ns.table`
4. `SELECT COUNT(*) FROM ns.table` — data present?
5. Add filters incrementally; read `metrics` to tune.

## See Also

- [api.md](api.md) · [patterns.md](patterns.md) · [configuration.md](configuration.md)

references/sql/patterns.md›

# Basin SQL Patterns

Code templates for CLI, REST, and Worker access. For performance/partitioning best practices, pull `https://developers.cloudflare.com/basin-sql/reference/limitations-best-practices/index.md`.

## Wrangler CLI

```bash
export WRANGLER_BASIN_SQL_AUTH_TOKEN=$API_TOKEN

npx wrangler basin sql query "${ACCOUNT_ID}_my-bucket" "
  SELECT category, COUNT(*) AS cnt, round(AVG(amount), 2) AS avg_amount
  FROM analytics.events
  WHERE __ingest_ts >= '2026-01-01T00:00:00Z'
  GROUP BY category ORDER BY cnt DESC LIMIT 100"
```

## REST API (Python)

```python
import requests

API = f"https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/basin-sql/query/{BUCKET}"
HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}

def r2sql(query):
    body = requests.post(API, headers=HEADERS, json={"query": query}, timeout=180).json()
    if body["success"]:
        return body["result"]["rows"], body["result"]["metrics"]
    raise RuntimeError(body["errors"])

rows, metrics = r2sql("SELECT category, COUNT(*) AS cnt FROM analytics.events GROUP BY category LIMIT 10")
```

## REST API (curl)

```bash
curl -X POST \
  "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/basin-sql/query/$BUCKET" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"query": "SELECT COUNT(*) AS total FROM analytics.events"}'
```

## Dashboard Worker

No Basin SQL binding exists — query the REST endpoint via `fetch()`.

```typescript
interface Env { ACCOUNT_ID: string; BUCKET: string; R2_SQL_TOKEN: string; }

async function queryR2SQL(env: Env, query: string) {
  const url = `https://api.sql.cloudflarestorage.com/api/v1/accounts/${env.ACCOUNT_ID}/basin-sql/query/${env.BUCKET}`;
  const resp = await fetch(url, {
    method: "POST",
    headers: { Authorization: `Bearer ${env.R2_SQL_TOKEN}`, "Content-Type": "application/json" },
    body: JSON.stringify({ query }),
  });
  if (!resp.ok) throw new Error(`Basin SQL ${resp.status}: ${await resp.text()}`);
  return (await resp.json() as any).result;
}

export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (new URL(req.url).pathname === "/api/analytics") {
      const result = await queryR2SQL(env, `
        SELECT category, COUNT(*) AS cnt FROM analytics.events
        GROUP BY category ORDER BY cnt DESC LIMIT 10`);
      return Response.json(result.rows);
    }
    return new Response("Not found", { status: 404 });
  },
};
```

```bash
npx wrangler secret put R2_SQL_TOKEN
```

## Example Queries

```sql
-- Error rate by endpoint
SELECT path, COUNT(*) AS total, SUM(CASE WHEN status >= 400 THEN 1 ELSE 0 END) AS errors
FROM logs.http_requests WHERE __ingest_ts >= '2026-01-01T00:00:00Z'
GROUP BY path ORDER BY errors DESC LIMIT 20;

-- Top-3 slowest requests per method (window + QUALIFY)
SELECT method, path, response_time_ms FROM logs.http_requests
QUALIFY ROW_NUMBER() OVER (PARTITION BY method ORDER BY response_time_ms DESC) <= 3;

-- Cross-table analytics with approx distinct
SELECT z.domain, COUNT(*) AS requests, approx_distinct(h.client_ip) AS uniques
FROM ns.zones z INNER JOIN ns.http_requests h ON z.zone_id = h.zone_id
WHERE h.__ingest_ts >= '2026-06-01T00:00:00Z'
GROUP BY z.domain ORDER BY requests DESC LIMIT 25;
```

## Cursor-Based Pagination

Paginate on a sortable (ideally partition) column rather than `OFFSET`:

```sql
SELECT * FROM logs.requests ORDER BY __ingest_ts DESC LIMIT 500;                       -- page 1
SELECT * FROM logs.requests WHERE __ingest_ts < '<last_ts>' ORDER BY __ingest_ts DESC LIMIT 500;  -- page 2
```

## Performance (essentials)

- **Always `LIMIT`** (early termination); **filter on partition keys first** (`__ingest_ts` range), then add predicates.
- **Narrow time ranges**; **compact tables** (file count dominates latency — enable automatic compaction in [Basin Catalog](../catalog/configuration.md)).
- Read response `metrics` (`files_scanned`, `bytes_scanned`) to tune. Full guidance: limitations-best-practices doc.

## Basin Pipelines → Basin SQL

After `npx wrangler basin pipelines setup` (Basin Catalog destination), wait for first flush (3–7 min), then query the table. See [pipelines/patterns.md](../pipelines/patterns.md).

## See Also

- [api.md](api.md) · [gotchas.md](gotchas.md) · [catalog/patterns.md](../catalog/patterns.md)

references/sql/README.md›

# Basin SQL

Serverless, distributed, **read-only** query engine (Apache DataFusion) for Apache Iceberg tables in Basin Catalog.

## Documentation

For full function lists, data types, and pricing, **retrieve the live docs** — use the Cloudflare MCP `docs` tool if available, otherwise `webfetch`.

| Topic | URL |
|-------|-----|
| Overview / get started | `https://developers.cloudflare.com/basin-sql/get-started/index.md` |
| Query data | `https://developers.cloudflare.com/basin-sql/query-data/index.md` |
| SQL reference | `https://developers.cloudflare.com/basin-sql/sql-reference/index.md` |
| Aggregate functions | `https://developers.cloudflare.com/basin-sql/sql-reference/aggregate-functions/index.md` |
| Scalar functions | `https://developers.cloudflare.com/basin-sql/sql-reference/scalar-functions/index.md` |
| Complex types | `https://developers.cloudflare.com/basin-sql/sql-reference/complex-types/index.md` |
| Limitations & best practices | `https://developers.cloudflare.com/basin-sql/reference/limitations-best-practices/index.md` |
| Wrangler commands | `https://developers.cloudflare.com/basin-sql/reference/wrangler-commands/index.md` |
| Pricing | `https://developers.cloudflare.com/basin-sql/platform/pricing/index.md` |

## Connection Values

| Value | Format |
|-------|--------|
| REST endpoint | `https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/basin-sql/query/{BUCKET}` |
| Wrangler | `npx wrangler basin sql query "{WAREHOUSE}" "<SQL>"` with `WRANGLER_BASIN_SQL_AUTH_TOKEN` set |
| Warehouse | `{ACCOUNT_ID}_{BUCKET}` |

> The REST endpoint is `api.sql.cloudflarestorage.com` — **not** the account API host. The legacy `/r2-sql/query/` path may still work, but use `/basin-sql/query/` for new integrations.

## Quick Start

```bash
npx wrangler basin catalog enable my-bucket           # 1. enable catalog
export WRANGLER_BASIN_SQL_AUTH_TOKEN=<r2-token>              # 2. auth (Admin R&W + Basin SQL Read)
npx wrangler basin sql query "$ACCOUNT_ID"_my-bucket \
  "SELECT * FROM default.my_table LIMIT 10"                # 3. query
```

## SQL Surface

Basin SQL is read-only and supports a broad analytical SQL surface (SELECT, JOINs, subqueries, CTEs, set operations, window functions, and aggregate/scalar/JSON functions over complex types). For the authoritative, current list of supported syntax, functions, and limitations, see the SQL reference and limitations docs linked above. [api.md](api.md) has query templates.

## When to Use

**Use for:** SQL analytics over Iceberg (logs, BI, fraud, ad-hoc), multi-cloud queries without egress, dashboards (query from a Worker via HTTP).

**Don't use for:** writes (use PySpark/PyIceberg) or real-time OLTP (<100 ms).

## No Workers Binding

There is no `env.R2_SQL` binding. Query from a Worker via `fetch()` to the REST endpoint with the token as a secret (see [patterns.md](patterns.md#dashboard-worker)).

## Reading Order

1. [configuration.md](configuration.md) — enable catalog, tokens, env setup
2. [api.md](api.md) — SQL syntax templates, JOIN/window examples, response format, data types
3. [patterns.md](patterns.md) — CLI/REST/Worker queries, use cases, pagination, performance
4. [gotchas.md](gotchas.md) — what works vs. not, performance, troubleshooting

## See Also

- [Basin Catalog](../catalog/) — PyIceberg/PySpark, table management
- [Basin Pipelines](../pipelines/) — streaming ingest into queryable tables

SKILL.md›

---
name: basin
description: Build and troubleshoot Cloudflare Basin analytics workflows with Basin Pipelines, Basin Catalog, and Basin SQL. Use for streaming data into R2 Iceberg tables, managing catalogs, or querying those tables; also use for requests using the former Data Platform, Pipelines, R2 Data Catalog, or R2 SQL names.
---

# Cloudflare Basin

Basin ingests and transforms events, manages Apache Iceberg tables in R2, and queries them with distributed SQL. Start with the [Basin overview](https://developers.cloudflare.com/basin/index.md) and retrieve the relevant current docs before implementing changes, especially for limits, pricing, permissions, and CLI syntax. While PR #33850 remains open, use the proposed pages in that PR if the new URLs have not been published.

## Rebrand

Cloudflare Data Platform is now **Basin**. Cloudflare Pipelines, R2 Data Catalog, and R2 SQL are now **Basin Pipelines**, **Basin Catalog**, and **Basin SQL**. Existing resources and configurations continue to work. The prior Wrangler command families also continue to work: `wrangler pipelines`, `wrangler r2 bucket catalog`, and `wrangler r2 sql`. Use the new Basin commands and documentation URLs in new guidance. Legacy API identifiers and metrics names may still use the old terms; check the current docs for deprecation details. The migration is documented in [cloudflare-docs PR #33850](https://github.com/cloudflare/cloudflare-docs/pull/33850).

## Choose the workflow

| Need | Product | Reference | Current docs |
|------|---------|-----------|--------------|
| Receive events, transform rows, and deliver to R2 | Basin Pipelines | [Pipelines guide](references/pipelines/README.md) | [Basin Pipelines](https://developers.cloudflare.com/basin-pipelines/index.md) |
| Manage Iceberg metadata, table maintenance, and engine access | Basin Catalog | [Catalog guide](references/catalog/README.md) | [Basin Catalog](https://developers.cloudflare.com/basin-catalog/index.md) |
| Query Iceberg tables with analytical SQL | Basin SQL | [SQL guide](references/sql/README.md) | [Basin SQL](https://developers.cloudflare.com/basin-sql/index.md) |

Typical flow: Basin Pipelines → Basin Catalog tables in R2 → Basin SQL or a compatible external engine. Begin with the [Basin getting started guide](https://developers.cloudflare.com/basin/get-started/guide/index.md) for an end-to-end setup.

The proposed Wrangler command families are `wrangler basin pipelines`, `wrangler basin catalog`, and `wrangler basin sql`. Basin SQL uses `WRANGLER_BASIN_SQL_AUTH_TOKEN`; its REST query path is `/basin-sql/query/{BUCKET}`. Check the installed Wrangler version and current reference pages before running commands because the migration is ongoing.