diff --git a/skills/cloudflare/SKILL.md b/skills/cloudflare/SKILL.md index 6d913f2..2c7da21 100644 --- a/skills/cloudflare/SKILL.md +++ b/skills/cloudflare/SKILL.md @@ -69,6 +69,8 @@ Need storage? ├─ Strongly-consistent per-entity state → durable-objects/ (DO storage) ├─ Secrets management → secrets-store/ ├─ Streaming ETL to R2 → pipelines/ +├─ Managed Apache Iceberg catalog on R2 → r2-data-catalog/ +├─ Serverless SQL analytics over Iceberg tables → r2-sql/ └─ Persistent cache (long-term retention) → cache-reserve/ ``` @@ -126,6 +128,7 @@ Need analytics? ├─ Custom high-cardinality metrics from Workers → analytics-engine/ ├─ Client-side (RUM) performance data → web-analytics/ ├─ Workers Logs and real-time debugging → observability/ +├─ SQL over Iceberg data lake (logs, events) → r2-sql/ (+ pipelines/, r2-data-catalog/) └─ Raw logs (Logpush to external tools) → Cloudflare docs ``` diff --git a/skills/cloudflare/references/pipelines/README.md b/skills/cloudflare/references/pipelines/README.md index 2724485..e87de9c 100644 --- a/skills/cloudflare/references/pipelines/README.md +++ b/skills/cloudflare/references/pipelines/README.md @@ -1,52 +1,56 @@ # Cloudflare Pipelines -ETL streaming platform for ingesting, transforming, and loading data into R2 with SQL transformations. +Streaming ingest: receive events over HTTP/Workers/Logpush, transform with SQL, write to R2 as Iceberg tables or Parquet/JSON files. -## Overview +## Documentation -Pipelines provides: -- **Streams**: Durable event buffers (HTTP/Workers ingestion) -- **Pipelines**: SQL-based transformations -- **Sinks**: R2 destinations (Iceberg tables or Parquet/JSON files) +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-docs` MCP/search tool if available, otherwise `webfetch` the URL. Docs are source of truth over this file. -**Status**: Open beta (Workers Paid plan) -**Pricing**: No charge beyond standard R2 storage/operations +| Topic | URL | +|-------|-----| +| Overview / getting started | `https://developers.cloudflare.com/pipelines/getting-started/` | +| Streams (write, manage, Logpush) | `https://developers.cloudflare.com/pipelines/streams/` | +| Sinks | `https://developers.cloudflare.com/pipelines/sinks/` | +| Pipelines & SQL transforms | `https://developers.cloudflare.com/pipelines/pipelines/` | +| SQL reference (statements, types) | `https://developers.cloudflare.com/pipelines/sql-reference/` | +| Wrangler commands | `https://developers.cloudflare.com/pipelines/reference/wrangler-commands/` | +| Terraform | `https://developers.cloudflare.com/pipelines/reference/terraform/` | +| Limits | `https://developers.cloudflare.com/pipelines/platform/limits/` | +| Pricing | `https://developers.cloudflare.com/pipelines/platform/pricing/` | +| Metrics (GraphQL) | `https://developers.cloudflare.com/pipelines/observability/metrics/` | -## Architecture +## Three Components ``` -Data Sources → Streams → Pipelines (SQL) → Sinks → R2 - ↑ ↓ ↓ - HTTP/Workers Transform Iceberg/Parquet +Sources → Stream → Pipeline (SQL) → Sink → R2 + ↑ ↓ ↓ + HTTP / Workers / Transform Iceberg (Data Catalog) + Logpush (row-level) or Parquet/JSON files ``` -| Component | Purpose | Key Feature | -|-----------|---------|-------------| -| Streams | Event ingestion | Structured (validated) or unstructured | -| Pipelines | Transform with SQL | Immutable after creation | -| Sinks | Write to R2 | Exactly-once delivery | +| 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 Data Catalog, or raw Parquet/JSON. | + +**Status:** Open beta (Workers Paid for production). Pricing announced; verify billing status in docs. ## Quick Start ```bash -# Interactive setup (recommended) +# Interactive — creates stream + sink + pipeline, optionally bucket + catalog npx wrangler pipelines setup ``` -**Minimal Worker example:** +Minimal Worker producer: ```typescript -interface Env { - STREAM: Pipeline; -} +interface Env { MY_STREAM: Pipeline; } export default { - async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { - const event = { user_id: "123", event_type: "purchase", amount: 29.99 }; - - // Fire-and-forget pattern - ctx.waitUntil(env.STREAM.send([event])); - - return new Response('OK'); + async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise { + ctx.waitUntil(env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }])); + return new Response("OK"); } } satisfies ExportedHandler; ``` @@ -54,52 +58,33 @@ export default { ## Which Sink Type? ``` -Need SQL queries on data? - → R2 Data Catalog (Iceberg) - ✅ ACID transactions, time-travel, schema evolution - ❌ More setup complexity (namespace, table, catalog token) - -Just file storage/archival? - → R2 Storage (Parquet) - ✅ Simple, direct file access - ❌ No built-in SQL queries - -Using external tools (Spark/Athena)? - → R2 Storage (Parquet with partitioning) - ✅ Standard format, partition pruning for performance - ❌ Must manage schema compatibility yourself -``` - -## Common Use Cases +Need SQL queries / ACID / time-travel on the data? + → R2 Data Catalog (Iceberg) ✅ R2 SQL, schema evolution ❌ more setup -- **Analytics pipelines**: Clickstream, telemetry, server logs -- **Data warehousing**: ETL into queryable Iceberg tables -- **Event processing**: Mobile/IoT with enrichment -- **Ecommerce analytics**: User events, purchases, views +Just archival / external tools (Spark, Athena)? + → R2 raw files (Parquet/JSON) ✅ simple, partitioned files ❌ no built-in SQL +``` -## Reading Order +## Critical Behaviors (read before building) -**New to Pipelines?** Start here: -1. [configuration.md](./configuration.md) - Setup streams, sinks, pipelines -2. [api.md](./api.md) - Send events, TypeScript types, SQL functions -3. [patterns.md](./patterns.md) - Best practices, integrations, complete example -4. [gotchas.md](./gotchas.md) - Critical warnings, troubleshooting +These are non-obvious and prevent most failures — see [gotchas.md](gotchas.md) for detail. -**Task-based routing:** -- Setup pipeline → [configuration.md](./configuration.md) -- Send/query data → [api.md](./api.md) -- Implement pattern → [patterns.md](./patterns.md) -- Debug issue → [gotchas.md](./gotchas.md) +- **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. -## In This Reference +## Reading Order -- [configuration.md](./configuration.md) - wrangler.jsonc bindings, schema definition, sink options, CLI commands -- [api.md](./api.md) - Pipeline binding interface, send() method, HTTP ingest, SQL function reference -- [patterns.md](./patterns.md) - Fire-and-forget, schema validation with Zod, integrations, performance tuning -- [gotchas.md](./gotchas.md) - Silent validation failures, immutable pipelines, latency expectations, limits +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 -- [r2](../r2/) - R2 storage backend for sinks -- [queues](../queues/) - Compare with Queues for async processing -- [workers](../workers/) - Worker runtime for event ingestion +- [r2-data-catalog](../r2-data-catalog/) — Iceberg sink destination +- [r2-sql](../r2-sql/) — query the ingested data +- [r2](../r2/) · [queues](../queues/) · [workers](../workers/) diff --git a/skills/cloudflare/references/pipelines/api.md b/skills/cloudflare/references/pipelines/api.md index ff302c7..98cb953 100644 --- a/skills/cloudflare/references/pipelines/api.md +++ b/skills/cloudflare/references/pipelines/api.md @@ -1,208 +1,124 @@ # Pipelines API Reference -## Pipeline Binding Interface +Code templates and verified behavior. For the full SQL function set and HTTP status semantics, pull `https://developers.cloudflare.com/pipelines/sql-reference/` and the streams docs. -```typescript -// From @cloudflare/workers-types -interface Pipeline { - send(data: object | object[]): Promise; -} - -interface Env { - STREAM: Pipeline; -} - -export default { - async fetch(request: Request, env: Env): Promise { - // send() returns Promise - no result data - await env.STREAM.send([event]); - return new Response('OK'); - } -} satisfies ExportedHandler; -``` - -**Key points:** -- `send()` accepts single object or array -- Always returns `Promise` (no confirmation data) -- Throws on network/validation errors (wrap in try/catch) -- Use `ctx.waitUntil()` for fire-and-forget pattern - -## Writing Events - -### Single Event +## Worker Binding Interface ```typescript -await env.STREAM.send([{ - user_id: "12345", - event_type: "purchase", - product_id: "widget-001", - amount: 29.99 -}]); -``` +// from cloudflare:pipelines / @cloudflare/workers-types +interface Pipeline { send(records: T[]): Promise; } -### Batch Events +interface Env { MY_STREAM: Pipeline; } -```typescript -const events = [ - { user_id: "user1", event_type: "view" }, - { user_id: "user2", event_type: "purchase", amount: 50 } -]; -await env.STREAM.send(events); -``` - -**Limits:** -- Max 1 MB per request -- 5 MB/s per stream - -### Fire-and-Forget Pattern - -```typescript export default { - async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { - const event = { /* ... */ }; - - // Don't block response on send - ctx.waitUntil(env.STREAM.send([event])); - - return new Response('OK'); + async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise { + await env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]); + return new Response("OK"); } -}; -``` - -### Error Handling - -```typescript -try { - await env.STREAM.send([event]); -} catch (error) { - console.error('Pipeline send failed:', error); - // Log to another system, retry, or return error response - return new Response('Failed to track event', { status: 500 }); -} +} satisfies ExportedHandler; ``` -## HTTP Ingest API +- `send()` takes an **array**, returns `Promise` (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/pipelines/platform/limits/` before sizing batches. -### Endpoint Format +## HTTP Ingest ``` https://{stream-id}.ingest.cloudflare.com ``` -Get `{stream-id}` from: `npx wrangler pipelines streams list` - -### Request Format - -**CRITICAL:** Must send array, not single object +Get `{stream-id}` from `npx wrangler pipelines streams list`. ```bash -# ✅ Correct +# Batch (preferred) curl -X POST https://{stream-id}.ingest.cloudflare.com \ -H "Content-Type: application/json" \ - -d '[{"user_id": "123", "event_type": "purchase"}]' + -d '[{"event_id":"evt-1","amount":29.99},{"event_id":"evt-2","amount":14.99}]' -# ❌ Wrong - will fail +# Single event — auto-wrapped in an array curl -X POST https://{stream-id}.ingest.cloudflare.com \ - -H "Content-Type: application/json" \ - -d '{"user_id": "123", "event_type": "purchase"}' + -H "Content-Type: application/json" -d '{"event_id":"evt-3","amount":9.99}' ``` -### Authentication +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). -```bash -curl -X POST https://{stream-id}.ingest.cloudflare.com \ - -H "Content-Type: application/json" \ - -H "Authorization: Bearer YOUR_API_TOKEN" \ - -d '[{"event": "data"}]' -``` - -**Required permission:** Workers Pipeline Send +> **JSON only** — no Avro, Protobuf, or CSV input. -Create token: Dashboard → Workers → API tokens → Create with Pipeline Send permission +## REST Management API -### Response Codes +Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1` -| Code | Meaning | Action | -|------|---------|--------| -| 200 | Accepted | Success | -| 400 | Invalid format | Check JSON array, schema match | -| 401 | Auth failed | Verify token valid | -| 413 | Payload too large | Split into smaller batches (<1 MB) | -| 429 | Rate limited | Back off, retry with delay | -| 5xx | Server error | Retry with exponential backoff | - -## SQL Functions Quick Reference +```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" +``` -Available in `INSERT INTO sink SELECT ... FROM stream` transformations: +> `wrangler pipelines delete` defaults to "no" non-interactively — use the REST API for automated cleanup. Deleting a stream removes buffered events and dependent pipelines. -| Function | Example | Use Case | -|----------|---------|----------| -| `UPPER(s)` | `UPPER(event_type)` | Normalize strings | -| `LOWER(s)` | `LOWER(email)` | Case-insensitive matching | -| `CONCAT(...)` | `CONCAT(user_id, '_', product_id)` | Generate composite keys | -| `CASE WHEN ... THEN ... END` | `CASE WHEN amount > 100 THEN 'high' ELSE 'low' END` | Conditional enrichment | -| `CAST(x AS type)` | `CAST(timestamp AS string)` | Type conversion | -| `COALESCE(x, y)` | `COALESCE(amount, 0.0)` | Default values | -| Math operators | `amount * 1.1`, `price / quantity` | Calculations | -| Comparison | `amount > 100`, `status IN ('active', 'pending')` | Filtering | +### Pipeline Lifecycle States -**String types for CAST:** `string`, `int32`, `int64`, `float32`, `float64`, `bool`, `timestamp` +| 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) | -Full reference: [Pipelines SQL Reference](https://developers.cloudflare.com/pipelines/sql-reference/) +> A `GET` on a sink shows `schema.fields: []` — expected. The sink inherits schema from the stream via the pipeline SQL. -## SQL Transform Examples +## Pipeline SQL (Transforms) -### Filter Events +Row-level only — no GROUP BY/aggregation. CTEs (`WITH`) and `UNNEST` are supported. Full function list: `https://developers.cloudflare.com/pipelines/sql-reference/`. ```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 * FROM my_stream -WHERE event_type = 'purchase' AND amount > 100 -``` +SELECT event_id, UPPER(category) AS category, amount * 1.1 AS amount_with_tax +FROM my_stream; -### Select Specific Fields +-- CTE +WITH filtered AS (SELECT event_id, amount FROM my_stream WHERE amount > 50) +INSERT INTO my_sink SELECT * FROM filtered; -```sql -INSERT INTO my_sink -SELECT user_id, event_type, timestamp, amount -FROM my_stream +-- UNNEST arrays (one per SELECT) +SELECT UNNEST(tags) AS tag FROM my_stream; ``` -### Transform and Enrich - -```sql -INSERT INTO my_sink -SELECT - user_id, - UPPER(event_type) as event_type, - timestamp, - amount * 1.1 as amount_with_tax, - CONCAT(user_id, '_', product_id) as unique_key, - CASE - WHEN amount > 1000 THEN 'high_value' - WHEN amount > 100 THEN 'medium_value' - ELSE 'low_value' - END as customer_tier -FROM my_stream -WHERE event_type IN ('purchase', 'refund') -``` +Supported categories: string, regex, hashing (`sha256`), JSON extraction, timestamp conversion, conditional (`CASE`), `CAST`, `COALESCE`, math/comparison operators. -## Querying Results (R2 Data Catalog) +## Verifying End-to-End Data Flow ```bash -export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_CATALOG_TOKEN - -npx wrangler r2 sql query "warehouse_name" " -SELECT - event_type, - COUNT(*) as event_count, - SUM(amount) as total_revenue -FROM default.my_table -WHERE event_type = 'purchase' - AND timestamp >= '2025-01-01' -GROUP BY event_type -ORDER BY total_revenue DESC -LIMIT 100" +# 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? (R2 SQL) +curl -s -X POST \ + "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/r2-sql/query/$BUCKET" \ + -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \ + -d '{"query": "SELECT COUNT(*) AS total FROM my_ns.my_table"}' ``` -**Note:** Iceberg tables support standard SQL queries with GROUP BY, JOINs, WHERE, ORDER BY, etc. +> 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 +- [r2-sql/api.md](../r2-sql/api.md) — querying results diff --git a/skills/cloudflare/references/pipelines/configuration.md b/skills/cloudflare/references/pipelines/configuration.md index 75e65f5..464b589 100644 --- a/skills/cloudflare/references/pipelines/configuration.md +++ b/skills/cloudflare/references/pipelines/configuration.md @@ -1,98 +1,155 @@ # Pipelines Configuration -## Worker Binding +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/pipelines/reference/wrangler-commands/` and the streams/sinks/pipelines docs. -```jsonc -// wrangler.jsonc -{ - "pipelines": [ - { "pipeline": "", "binding": "STREAM" } - ] -} -``` +## Naming Rules -Get stream ID: `npx wrangler pipelines streams list` +- **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": "user_id", "type": "string", "required": true }, - { "name": "event_type", "type": "string", "required": true }, - { "name": "amount", "type": "float64", "required": false }, - { "name": "timestamp", "type": "timestamp", "required": true } + { "name": "event_id", "type": "string", "required": true }, + { "name": "amount", "type": "float64", "required": false } ] } ``` -**Types:** `string`, `int32`, `int64`, `float32`, `float64`, `bool`, `timestamp`, `json`, `binary`, `list`, `struct` +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/pipelines/sql-reference/sql-data-types/`. -## Stream Setup +Unstructured streams (no schema) store everything in a single `value` column. -```bash -# With schema -npx wrangler pipelines streams create my-stream --schema-file schema.json +> Pipelines auto-adds `__ingest_ts` (TIMESTAMP, day-partitioned). Do **not** include it in your schema. -# Unstructured (no validation) -npx wrangler pipelines streams create my-stream +## Option A: Interactive (Simplest) -# List/get/delete -npx wrangler pipelines streams list -npx wrangler pipelines streams get -npx wrangler pipelines streams delete +```bash +npx wrangler pipelines setup # creates stream + sink + pipeline, optionally bucket + catalog ``` -## Sink Configuration +## Option B: Wrangler CLI (Explicit) -**R2 Data Catalog (Iceberg):** ```bash -npx wrangler pipelines sinks create my-sink \ +# 1. Stream +npx wrangler pipelines streams create my_stream --schema-file schema.json + +# 2. Sink — R2 Data Catalog (Iceberg). Creates the namespace + table. +npx wrangler pipelines sinks create my_sink \ --type r2-data-catalog \ - --bucket my-bucket --namespace default --table events \ - --catalog-token $TOKEN \ - --compression zstd --roll-interval 60 -``` + --bucket my-bucket --namespace my_namespace --table my_table \ + --catalog-token $API_TOKEN \ + --compression zstd --roll-interval 300 -**R2 Raw (Parquet):** -```bash -npx wrangler pipelines sinks create my-sink \ +# 2b. Sink — R2 raw Parquet (alternative) +npx wrangler pipelines sinks create my_sink \ --type r2 --bucket my-bucket --format parquet \ - --path analytics/events \ - --partitioning "year=%Y/month=%m/day=%d" \ + --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 pipelines create my_pipeline \ + --sql "INSERT INTO my_sink SELECT * FROM my_stream" ``` -| Option | Values | Guidance | -|--------|--------|----------| -| `--compression` | `zstd`, `snappy`, `gzip` | `zstd` best ratio, `snappy` fastest | -| `--roll-interval` | Seconds | Low latency: 10-60, Query perf: 300 | -| `--roll-size` | MB | Larger = better compression | +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). -## Pipeline Creation +> **⚠️ 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 -npx wrangler pipelines create my-pipeline \ - --sql "INSERT INTO my_sink SELECT * FROM my_stream WHERE event_type = 'purchase'" +# 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;"}' ``` -**⚠️ Pipelines are immutable** - cannot modify SQL. Must delete/recreate. +**REST field names ≠ CLI flags** (common failure — not obvious from docs): -## Credentials +| REST (config body) | CLI flag | Gotcha | +|--------------------|----------|--------| +| `"type": "r2_data_catalog"` | `--type r2-data-catalog` | underscores vs hyphens | +| `"table_name"` | `--table` | different key | +| `"token"` | `--catalog-token` | different key | +| `"format": {"type": "parquet"}` | (implied) | required in REST, omitted in CLI | -| Type | Permission | Get From | -|------|------------|----------| -| Catalog token | R2 Admin Read & Write | Dashboard → R2 → API tokens | -| R2 credentials | Object Read & Write | `wrangler r2 bucket create` output | -| HTTP ingest token | Workers Pipeline Send | Dashboard → Workers → API tokens | +## Worker Binding -## Complete Example +```jsonc +// wrangler.jsonc +{ "pipelines": [ { "stream": "", "binding": "MY_STREAM" } ] } +``` -```bash -npx wrangler r2 bucket create my-bucket -npx wrangler r2 bucket catalog enable my-bucket -npx wrangler pipelines streams create my-stream --schema-file schema.json -npx wrangler pipelines sinks create my-sink --type r2-data-catalog --bucket my-bucket ... -npx wrangler pipelines create my-pipeline --sql "INSERT INTO my_sink SELECT * FROM my_stream" -npx wrangler deploy +> Binding field is `"stream"` as of June 2026 (was `"pipeline"`, still accepted). Use the **stream ID** (`wrangler pipelines streams list`), not the pipeline ID. Redeploy after adding. Generate typed bindings with `npx wrangler types` → `Pipeline` from `cloudflare:pipelines`. + +## Terraform + +Resources: `cloudflare_pipeline_stream`, `cloudflare_pipeline_sink`, `cloudflare_pipeline`. For current attribute schemas pull `https://developers.cloudflare.com/pipelines/reference/terraform/`. + +```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 + R2 Data 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 diff --git a/skills/cloudflare/references/pipelines/gotchas.md b/skills/cloudflare/references/pipelines/gotchas.md index 2a2a75f..9770c63 100644 --- a/skills/cloudflare/references/pipelines/gotchas.md +++ b/skills/cloudflare/references/pipelines/gotchas.md @@ -1,80 +1,58 @@ # Pipelines Gotchas -## Critical Issues +Non-obvious failure modes (not well covered by docs). For current limits and error semantics, pull `https://developers.cloudflare.com/pipelines/platform/limits/`. -### Events Silently Dropped +## Events accepted but never appear (most common) -**Most common issue.** Events accepted (HTTP 200) but never appear in sink. +HTTP 200 / `send()` resolves, but no data in the sink. Causes: -**Causes:** -1. Schema validation fails - structured streams drop invalid events silently -2. Waiting for roll interval (10-300s) - expected behavior +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}`. -**Solution:** Validate client-side with Zod: -```typescript -const EventSchema = z.object({ user_id: z.string(), amount: z.number() }); -try { - const validated = EventSchema.parse(rawEvent); - await env.STREAM.send([validated]); -} catch (e) { /* get immediate feedback */ } -``` - -### Pipelines Are Immutable +## Everything is immutable -Cannot modify SQL after creation. Must delete and recreate. +Cannot modify stream schema, pipeline SQL, or sink config — delete and recreate. Use version naming (`events_v1`) and keep SQL in version control. ```bash -npx wrangler pipelines delete old-pipeline -npx wrangler pipelines create new-pipeline --sql "..." +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" ``` -**Tip:** Use version naming (`events-pipeline-v1`) and keep SQL in version control. +## Worker binding undefined (`env.MY_STREAM`) -### Worker Binding Not Found +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. -**`env.STREAM is undefined`** +## REST API field names ≠ CLI flags -1. Use **stream ID** (not pipeline ID) in `wrangler.jsonc` -2. Redeploy after adding binding +`r2_data_catalog` vs `--type r2-data-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). -```bash -npx wrangler pipelines streams list # Get stream ID -npx wrangler deploy -``` +## `wrangler pipelines delete` defaults to "no" -## Common Errors +Non-interactive environments answer "no" automatically — use REST `DELETE` for CI/automation. -| Error | Cause | Fix | -|-------|-------|-----| -| Events not in R2 | Roll interval not elapsed | Wait 10-300s, check `roll_interval` | -| Schema validation failures | Type mismatch, missing fields | Validate client-side | -| Rate limit (429) | >5 MB/s per stream | Batch events, request increase | -| Payload too large (413) | >1 MB request | Split into smaller batches | -| Cannot delete stream | Pipeline references it | Delete pipelines first | -| Sink credential errors | Token expired | Recreate sink with new credentials | +## Behavioral Notes -## Limits (Open Beta) +- **`__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 [R2 SQL](../r2-sql/) at query time). CTEs and `UNNEST` are supported. -| Resource | Limit | -|----------|-------| -| Streams/Sinks/Pipelines per account | 20 each | -| Payload size | 1 MB | -| Ingest rate per stream | 5 MB/s | -| Event retention | 24 hours | -| Recommended batch size | 100 events | - -## SQL Limitations +## Debug Checklist -- **No JOINs** - single stream per pipeline -- **No window functions** - basic SQL only -- **No subqueries** - must use `INSERT INTO ... SELECT ... FROM` -- **No schema evolution** - cannot modify after creation +- [ ] Stream exists: `wrangler 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 -## Debug Checklist +## See Also -- [ ] Stream exists: `npx wrangler pipelines streams list` -- [ ] Pipeline healthy: `npx wrangler pipelines get ` -- [ ] SQL syntax matches schema -- [ ] Worker redeployed after binding added -- [ ] Waited for roll interval -- [ ] Accepted vs processed count matches (no validation drops) +- [configuration.md](configuration.md) · [api.md](api.md) · [patterns.md](patterns.md) diff --git a/skills/cloudflare/references/pipelines/patterns.md b/skills/cloudflare/references/pipelines/patterns.md index 186b6a2..42ed20a 100644 --- a/skills/cloudflare/references/pipelines/patterns.md +++ b/skills/cloudflare/references/pipelines/patterns.md @@ -1,87 +1,130 @@ # Pipelines Patterns -## Fire-and-Forget +Code-first patterns. For observability dataset/field schemas and Logpush dataset lists, pull `https://developers.cloudflare.com/pipelines/observability/metrics/` and `https://developers.cloudflare.com/pipelines/streams/logpush/`. + +## Fire-and-Forget Producer ```typescript export default { - async fetch(request, env, ctx) { - const event = { user_id: '...', event_type: 'page_view', timestamp: new Date().toISOString() }; - ctx.waitUntil(env.STREAM.send([event])); // Don't block response - return new Response('OK'); + 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"); } }; ``` -## Schema Validation with Zod +## Client-Side Validation with Zod + +Structured streams drop invalid events silently during processing. Validate before sending for immediate feedback. ```typescript -import { z } from 'zod'; +import { z } from "zod"; const EventSchema = z.object({ - user_id: z.string(), - event_type: z.enum(['purchase', 'view']), - amount: z.number().positive().optional() + event_id: z.string(), + category: z.enum(["purchase", "view"]), + amount: z.number().positive().optional(), }); -const validated = EventSchema.parse(rawEvent); // Throws on invalid -await env.STREAM.send([validated]); +const validated = EventSchema.parse(rawEvent); // throws synchronously +await env.MY_STREAM.send([validated]); ``` -**Why:** Structured streams drop invalid events silently. Client validation gives immediate feedback. +## Scheduled Collector Worker -## SQL Transform Patterns +```jsonc +// wrangler.jsonc +{ + "name": "collector", + "pipelines": [{ "stream": "", "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 → Pipelines + +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 --- Filter early (reduce storage) -INSERT INTO my_sink -SELECT user_id, event_type, amount -FROM my_stream -WHERE event_type = 'purchase' AND amount > 10 - --- Select only needed fields -INSERT INTO my_sink -SELECT user_id, event_type, timestamp FROM my_stream - --- Enrich with CASE -INSERT INTO my_sink -SELECT user_id, amount, - CASE WHEN amount > 1000 THEN 'vip' ELSE 'standard' END as tier -FROM my_stream +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 → Pipelines** destination) or API. + ## Pipelines + Queues Fan-out ```typescript await Promise.all([ - env.ANALYTICS_STREAM.send([event]), // Long-term storage - env.PROCESS_QUEUE.send(event) // Immediate processing + env.ANALYTICS_STREAM.send([event]), // long-term storage + SQL + env.PROCESS_QUEUE.send(event), // immediate processing + retries ]); ``` -| Need | Use | -|------|-----| -| Long-term storage, SQL queries | Pipelines | -| Immediate processing, retries | Queues | -| Both | Fan-out pattern | +Use 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. -## Performance Tuning +### Detecting Silent Data Loss -| Goal | Config | -|------|--------| -| Low latency | `--roll-interval 10` | -| Query performance | `--roll-interval 300 --roll-size 100` | -| Cost optimal | `--compression zstd --roll-interval 300` | +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 R2 SQL returns expected counts. -## Schema Evolution +## Schema Evolution (Immutable Pipelines) -Pipelines are immutable. Use versioning: +Pipelines can't change. Version + dual-write: ```bash -# Create v2 stream/sink/pipeline -npx wrangler pipelines streams create events-v2 --schema-file v2.json - -# Dual-write during transition +npx wrangler 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 R2 SQL +``` + +## End-to-End: Streaming Analytics Dashboard -# Query across versions with UNION ALL ``` +External APIs → Collector Worker (cron) → Pipeline → R2 (Iceberg) → Dashboard Worker → R2 SQL +``` + +1. Create bucket + enable catalog ([r2-data-catalog](../r2-data-catalog/configuration.md)) +2. Create stream + sink + pipeline (here) +3. Collector Worker with cron + stream binding (above) +4. Dashboard Worker querying R2 SQL ([r2-sql/patterns.md](../r2-sql/patterns.md)) +5. Enable automatic compaction + +## See Also + +- [configuration.md](configuration.md) · [api.md](api.md) · [gotchas.md](gotchas.md) · [r2-sql](../r2-sql/) diff --git a/skills/cloudflare/references/r2-data-catalog/README.md b/skills/cloudflare/references/r2-data-catalog/README.md index 88702fa..2df7162 100644 --- a/skills/cloudflare/references/r2-data-catalog/README.md +++ b/skills/cloudflare/references/r2-data-catalog/README.md @@ -1,149 +1,75 @@ -# Cloudflare R2 Data Catalog Skill Reference +# Cloudflare R2 Data Catalog -Expert guidance for Cloudflare R2 Data Catalog - Apache Iceberg catalog built into R2 buckets. +Managed Apache Iceberg REST catalog built into R2 buckets. No catalog servers to run. -## Reading Order - -**New to R2 Data Catalog?** Start here: -1. Read "What is R2 Data Catalog?" and "When to Use" below -2. [configuration.md](configuration.md) - Enable catalog, create tokens -3. [patterns.md](patterns.md) - PyIceberg setup and common patterns -4. [api.md](api.md) - REST API reference as needed -5. [gotchas.md](gotchas.md) - Troubleshooting when issues arise - -**Quick reference?** Jump to: -- [Enable catalog on bucket](configuration.md#enable-catalog-on-bucket) -- [PyIceberg connection pattern](patterns.md#pyiceberg-connection-pattern) -- [Permission errors](gotchas.md#permission-errors) +## Documentation -## What is R2 Data Catalog? +This reference is a fast-start with verified connection details and code. For limits, maintenance settings, engine config examples, and pricing, **retrieve the live docs** — use the `cloudflare-docs` MCP/search tool if available, otherwise `webfetch` the URL. Docs are source of truth over this file. -R2 Data Catalog is a **managed Apache Iceberg REST catalog** built directly into R2 buckets. It provides: +| Topic | URL | +|-------|-----| +| Overview / get started | `https://developers.cloudflare.com/r2/data-catalog/get-started/` | +| Manage catalogs (enable, tokens) | `https://developers.cloudflare.com/r2/data-catalog/manage-catalogs/` | +| Engine config examples | `https://developers.cloudflare.com/r2/data-catalog/config-examples/` (`pyiceberg/`, `spark-python/`, `spark-scala/`, `duckdb/`, `snowflake/`, `trino/`, `starrocks/`) | +| Table maintenance (compaction, snapshots) | `https://developers.cloudflare.com/r2/data-catalog/table-maintenance/` | +| Deleting data | `https://developers.cloudflare.com/r2/data-catalog/deleting-data/` | +| Metrics (GraphQL) | `https://developers.cloudflare.com/r2/data-catalog/observability/metrics/` | +| Pricing | `https://developers.cloudflare.com/r2/data-catalog/platform/pricing/` | +| Iceberg spec | `https://iceberg.apache.org/spec/` | -- **Apache Iceberg tables** - ACID transactions, schema evolution, time-travel queries -- **Zero-egress costs** - Query from any cloud/region without data transfer fees -- **Standard REST API** - Works with Spark, PyIceberg, Snowflake, Trino, DuckDB -- **No infrastructure** - Fully managed, no catalog servers to run -- **Public beta** - Available to all R2 subscribers, no extra cost beyond R2 storage +## Connection Values -### What is Apache Iceberg? - -Open table format for analytics datasets in object storage. Features: -- **ACID transactions** - Safe concurrent reads/writes -- **Metadata optimization** - Fast queries without full scans -- **Schema evolution** - Add/rename/delete columns without rewrites -- **Time-travel** - Query historical snapshots -- **Partitioning** - Organize data for efficient queries - -## When to Use +Use the exact **Catalog URI** and **Warehouse** printed by `npx wrangler r2 bucket catalog enable ` (also shown in the dashboard). They follow these formats: -**Use R2 Data Catalog for:** -- **Log analytics** - Store and query application/system logs -- **Data lakes/warehouses** - Analytical datasets queried by multiple engines -- **BI pipelines** - Aggregate data for dashboards and reports -- **Multi-cloud analytics** - Share data across clouds without egress fees -- **Time-series data** - Event streams, metrics, sensor data +| Value | Format | Example | +|-------|--------|---------| +| Catalog URI | `https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}` | `https://catalog.cloudflarestorage.com/4482a1.../live-data` | +| Warehouse | `{ACCOUNT_ID}_{BUCKET}` (hyphens preserved) | `4482a1..._live-data` | +| Token | R2 API token (Admin R&W on Storage + R&W on Data Catalog) | `cfut_...` | -**Don't use for:** -- **Transactional workloads** - Use D1 or external database instead -- **Sub-second latency** - Iceberg optimized for batch/analytical queries -- **Small datasets (<1GB)** - Setup overhead not worth it -- **Unstructured data** - Store files directly in R2, not as Iceberg tables +The Iceberg `/config` route needs `?warehouse={WAREHOUSE}`. ## Architecture ``` -┌─────────────────────────────────────────────────┐ -│ Query Engines │ -│ (PyIceberg, Spark, Trino, Snowflake, DuckDB) │ -└────────────────┬────────────────────────────────┘ - │ - │ REST API (OAuth2 token) - ▼ -┌─────────────────────────────────────────────────┐ -│ R2 Data Catalog (Managed Iceberg REST Catalog)│ -│ • Namespace/table metadata │ -│ • Transaction coordination │ -│ • Snapshot management │ -└────────────────┬────────────────────────────────┘ - │ - │ Vended credentials - ▼ -┌─────────────────────────────────────────────────┐ -│ R2 Bucket Storage │ -│ • Parquet data files │ -│ • Metadata files │ -│ • Manifest files │ -└─────────────────────────────────────────────────┘ +Engines (PyIceberg, PySpark, Trino, Snowflake, DuckDB, R2 SQL) + │ Iceberg REST API (Bearer token) + ▼ +R2 Data Catalog ── namespace/table metadata, snapshots, txn coordination + │ vended S3 credentials + ▼ +R2 Bucket ── Parquet data files + Iceberg metadata ``` -**Key concepts:** -- **Catalog URI** - REST endpoint for catalog operations (e.g., `https://.r2.cloudflarestorage.com/iceberg/`) -- **Warehouse** - Logical grouping of tables (typically same as bucket name) -- **Namespace** - Schema/database containing tables (e.g., `logs`, `analytics`) -- **Table** - Iceberg table with schema, data files, snapshots -- **Vended credentials** - Temporary S3 credentials catalog provides for data access +- **Warehouse** — top-level catalog grouping (`{ACCOUNT_ID}_{BUCKET}`) +- **Namespace** — schema/database; nested namespaces supported +- **Table** — Iceberg table (schema, partition spec, snapshots) +- **Vended credentials** — temp S3 creds the catalog hands engines (`X-Iceberg-Access-Delegation: vended-credentials`) -## Limits - -| Resource | Limit | Notes | -|----------|-------|-------| -| Namespaces per catalog | No hard limit | Organize tables logically | -| Tables per namespace | <10,000 recommended | Performance degrades beyond this | -| Files per table | <100,000 recommended | Run compaction regularly | -| Snapshots per table | Configurable retention | Expire >7 days old | -| Partitions per table | 100-1,000 optimal | Too many = slow metadata ops | -| Table size | Same as R2 bucket | 10GB-10TB+ common | -| API rate limits | Standard R2 API limits | Shared with R2 storage operations | -| Target file size | 128-512 MB | After compaction | - -## Current Status +## When to Use -**Public Beta** (as of Jan 2026) -- Available to all R2 subscribers -- No extra cost beyond standard R2 storage/operations -- Production-ready, but breaking changes possible -- Supports: namespaces, tables, snapshots, compaction, time-travel, table maintenance +**Use for:** log/analytics data lakes, BI pipelines, time-series/event data, multi-cloud or multi-engine analytics needing ACID + schema evolution on object storage. -## Decision Tree: Is R2 Data Catalog Right For You? +**Don't use for:** OLTP (use D1/a database), sub-second point lookups, tiny datasets (<1 GB), or unstructured blobs (store directly in R2). -``` -Start → Need analytics on object storage data? - │ - ├─ No → Use R2 directly for object storage - │ - └─ Yes → Dataset >1GB with structured schema? - │ - ├─ No → Too small, use R2 + ad-hoc queries - │ - └─ Yes → Need ACID transactions or schema evolution? - │ - ├─ No → Consider simpler solutions (Parquet on R2) - │ - └─ Yes → Need multi-cloud/multi-tool access? - │ - ├─ No → D1 or external DB may be simpler - │ - └─ Yes → ✅ Use R2 Data Catalog -``` +## Two APIs — Don't Confuse Them -**Quick check:** If you answer "yes" to all: -- Dataset >1GB and growing -- Structured/tabular data (logs, events, metrics) -- Multiple query tools or cloud environments -- Need versioning, schema changes, or concurrent access +| API | Base | Use for | +|-----|------|---------| +| **Iceberg REST catalog** | `https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}` | Table reads/writes via PyIceberg, PySpark, Trino, etc. | +| **Control-plane REST API** | `https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/r2-catalog/{BUCKET}` | Enable/disable, maintenance config, list namespaces/tables, get-table | -→ R2 Data Catalog is a good fit. +**Status:** Open beta. Available to all R2 subscribers; verify pricing/billing status in docs. -## In This Reference +## Reading Order -- **[configuration.md](configuration.md)** - Enable catalog, create API tokens, connect clients -- **[api.md](api.md)** - REST endpoints, operations, maintenance -- **[patterns.md](patterns.md)** - PyIceberg examples, common use cases -- **[gotchas.md](gotchas.md)** - Troubleshooting, best practices, limitations +1. [configuration.md](configuration.md) — enable catalog, tokens, maintenance, client connection +2. [api.md](api.md) — control-plane REST (incl. get-table), PyIceberg client, maintenance +3. [patterns.md](patterns.md) — PyIceberg + PySpark templates, partitioning, external engines +4. [gotchas.md](gotchas.md) — auth errors, maintenance behavior, troubleshooting ## See Also -- [Cloudflare R2 Data Catalog Docs](https://developers.cloudflare.com/r2/data-catalog/) -- [Apache Iceberg Docs](https://iceberg.apache.org/) -- [PyIceberg Docs](https://py.iceberg.apache.org/) +- [pipelines](../pipelines/) — stream events into Iceberg tables +- [r2-sql](../r2-sql/) — serverless SQL over these tables +- [r2](../r2/) — underlying object storage diff --git a/skills/cloudflare/references/r2-data-catalog/api.md b/skills/cloudflare/references/r2-data-catalog/api.md index 3d57d4f..d1e12ce 100644 --- a/skills/cloudflare/references/r2-data-catalog/api.md +++ b/skills/cloudflare/references/r2-data-catalog/api.md @@ -1,199 +1,122 @@ -# API Reference +# R2 Data Catalog API Reference -R2 Data Catalog exposes standard [Apache Iceberg REST Catalog API](https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml). +Two APIs: the **control-plane REST API** (Cloudflare-specific) and the **Iceberg REST catalog API** (standard, used via PyIceberg/PySpark). For PyIceberg method details pull `https://py.iceberg.apache.org/`; for engine configs see `https://developers.cloudflare.com/r2/data-catalog/config-examples/`. -## Quick Reference +## Control-Plane REST API -**Most common operations:** - -| Task | PyIceberg Code | -|------|----------------| -| Connect | `RestCatalog(name="r2", warehouse=bucket, uri=uri, token=token)` | -| List namespaces | `catalog.list_namespaces()` | -| Create namespace | `catalog.create_namespace("logs")` | -| Create table | `catalog.create_table(("ns", "table"), schema=schema)` | -| Load table | `catalog.load_table(("ns", "table"))` | -| Append data | `table.append(pyarrow_table)` | -| Query data | `table.scan().to_pandas()` | -| Compact files | `table.rewrite_data_files(target_file_size_bytes=128*1024*1024)` | -| Expire snapshots | `table.expire_snapshots(older_than=timestamp_ms, retain_last=10)` | - -## REST Endpoints - -Base: `https://.r2.cloudflarestorage.com/iceberg/` +Base: `https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/r2-catalog/{BUCKET}` +Auth: `Authorization: Bearer $API_TOKEN` | Operation | Method | Path | |-----------|--------|------| -| Catalog config | GET | `/v1/config` | -| List namespaces | GET | `/v1/namespaces` | -| Create namespace | POST | `/v1/namespaces` | -| Delete namespace | DELETE | `/v1/namespaces/{ns}` | -| List tables | GET | `/v1/namespaces/{ns}/tables` | -| Create table | POST | `/v1/namespaces/{ns}/tables` | -| Load table | GET | `/v1/namespaces/{ns}/tables/{table}` | -| Update table | POST | `/v1/namespaces/{ns}/tables/{table}` | -| Delete table | DELETE | `/v1/namespaces/{ns}/tables/{table}` | -| Rename table | POST | `/v1/tables/rename` | - -**Authentication:** Bearer token in header: `Authorization: Bearer ` - -## PyIceberg Client API - -Most users use PyIceberg, not raw REST. - -### Connection - -```python -from pyiceberg.catalog.rest import RestCatalog - -catalog = RestCatalog( - name="my_catalog", - warehouse="", - uri="", - token="", -) +| Get catalog details | GET | `/r2-catalog/{bucket}` | +| Enable / disable | POST | `/r2-catalog/{bucket}/enable` · `/disable` | +| Store compaction credential | POST | `/r2-catalog/{bucket}/credential` | +| List namespaces | GET | `/namespaces` | +| List tables | GET | `/namespaces/{ns}/tables` | +| **Get table metadata** | GET | `/namespaces/{ns}/tables/{table}` | +| Get/update maintenance config | GET/POST | `/maintenance-configs` and `/namespaces/{ns}/tables/{table}/maintenance-configs` | + +List endpoints accept `?return_uuids=true`, `?return_details=true`, `?parent={ns}`, and pagination. **Nested namespaces use `%1F` (Unit Separator)**, not `/` or `.`: `/namespaces/parent%1Fchild/tables`. + +```bash +# Catalog details (status, maintenance_config, credential_status) +curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET" \ + -H "Authorization: Bearer $API_TOKEN" + +# Store token for compaction (pure-API setups) +curl -s -X POST "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/credential" \ + -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \ + -d '{"token": "'$API_TOKEN'"}' + +# Update maintenance config (all fields optional; table-level overrides catalog-level) +curl -s -X POST "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/maintenance-configs" \ + -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \ + -d '{"compaction": {"state": "enabled", "target_size_mb": "256"}, + "snapshot_expiration": {"state": "enabled", "min_snapshots_to_keep": 10, "max_snapshot_age": "7d"}}' ``` -### Namespace Operations +### Get Table (metadata introspection) -```python -from pyiceberg.exceptions import NamespaceAlreadyExistsError +`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. (Newer than the published API docs.) -namespaces = catalog.list_namespaces() # [('default',), ('logs',)] -catalog.create_namespace("logs", properties={"owner": "team"}) -catalog.drop_namespace("logs") # Must be empty +```bash +curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/live/tables/earthquakes" \ + -H "Authorization: Bearer $API_TOKEN" ``` -### Table Operations - -```python -from pyiceberg.schema import Schema -from pyiceberg.types import NestedField, StringType, IntegerType - -schema = Schema( - NestedField(1, "id", IntegerType(), required=True), - NestedField(2, "name", StringType(), required=False), -) -table = catalog.create_table(("logs", "app_logs"), schema=schema) -tables = catalog.list_tables("logs") -table = catalog.load_table(("logs", "app_logs")) -catalog.rename_table(("logs", "old"), ("logs", "new")) +```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} ``` -### Data Operations +| 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 | -```python -import pyarrow as pa +### Error Format -data = pa.table({"id": [1, 2], "name": ["Alice", "Bob"]}) -table.append(data) -table.overwrite(data) - -# Read with filters -scan = table.scan(row_filter="id > 100", selected_fields=["id", "name"]) -df = scan.to_pandas() +```json +{"success": false, "errors": [{"code": 10000, "message": "Authentication error"}]} ``` -### Schema Evolution - -```python -from pyiceberg.types import IntegerType, LongType +Standard HTTP codes (401 auth, 403 perms, 404 not enabled/found, 409 conflict). -with table.update_schema() as update: - update.add_column("user_id", IntegerType(), doc="User ID") - update.rename_column("msg", "message") - update.delete_column("old_field") - update.update_column("id", field_type=LongType()) # int→long only -``` +## Iceberg REST Catalog API (via PyIceberg) -### Time-Travel +Standard [Iceberg REST Catalog](https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml). Base: `https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}`. The `/config` route needs `?warehouse={WAREHOUSE}`. ```python -from datetime import datetime, timedelta - -# Query specific snapshot or timestamp -scan = table.scan(snapshot_id=table.snapshots()[-2].snapshot_id) -yesterday_ms = int((datetime.now() - timedelta(days=1)).timestamp() * 1000) -scan = table.scan(as_of_timestamp=yesterday_ms) -``` - -### Partitioning - -```python -from pyiceberg.partitioning import PartitionSpec, PartitionField -from pyiceberg.transforms import DayTransform -from pyiceberg.types import TimestampType - -partition_spec = PartitionSpec( - PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day") -) -table = catalog.create_table(("events", "actions"), schema=schema, partition_spec=partition_spec) -scan = table.scan(row_filter="day = '2026-01-27'") # Prunes partitions +from pyiceberg.catalog.rest import RestCatalog +catalog = RestCatalog(name="r2", warehouse=WAREHOUSE, uri=CATALOG_URI, token=TOKEN) ``` -## Table Maintenance - -### Compaction +Common operations (see PyIceberg docs for full signatures): ```python -files = table.scan().plan_files() -avg_mb = sum(f.file_size_in_bytes for f in files) / len(files) / (1024**2) -print(f"Files: {len(files)}, Avg: {avg_mb:.1f} MB") - -table.rewrite_data_files(target_file_size_bytes=128 * 1024 * 1024) +catalog.create_namespace_if_not_exists("logs") +catalog.list_tables("logs") +table = catalog.create_table(("logs", "events"), schema=schema) # pyiceberg.schema.Schema +table = catalog.load_table(("logs", "events")) +table.append(pyarrow_table) # also .overwrite(...) +table.scan(row_filter="id > 100").to_pandas() ``` -**When:** Avg <10MB or >1000 files. **Frequency:** High-write daily, medium weekly. - -### Snapshot Expiration - +Schema evolution (add nullable columns; widen types only): ```python -from datetime import datetime, timedelta - -seven_days_ms = int((datetime.now() - timedelta(days=7)).timestamp() * 1000) -table.expire_snapshots(older_than=seven_days_ms, retain_last=10) +with table.update_schema() as u: + u.add_column("user_id", LongType(), doc="User ID") + u.rename_column("msg", "message") ``` -**Retention:** Production 7-30d, dev 1-7d, audit 90+d. - -### Orphan Cleanup - +Time-travel: ```python -three_days_ms = int((datetime.now() - timedelta(days=3)).timestamp() * 1000) -table.delete_orphan_files(older_than=three_days_ms) +table.scan(snapshot_id=table.snapshots()[-2].snapshot_id) +table.scan(as_of_timestamp=ms_epoch) ``` -⚠️ Always expire snapshots first, use 3+ day threshold, run during low traffic. +## Manual Maintenance (PySpark) -### Full Maintenance +Prefer automatic maintenance (control-plane API/wrangler). For manual control or very large tables, use Spark procedures (`rewrite_data_files`, `rewrite_manifests`, `expire_snapshots`, `remove_orphan_files`). See `https://developers.cloudflare.com/r2/data-catalog/table-maintenance/`. ```python -# Compact → Expire → Cleanup (in order) -if len(table.scan().plan_files()) > 1000: - table.rewrite_data_files(target_file_size_bytes=128 * 1024 * 1024) -seven_days_ms = int((datetime.now() - timedelta(days=7)).timestamp() * 1000) -table.expire_snapshots(older_than=seven_days_ms, retain_last=10) -three_days_ms = int((datetime.now() - timedelta(days=3)).timestamp() * 1000) -table.delete_orphan_files(older_than=three_days_ms) +spark.sql("CALL r2dc.system.rewrite_data_files(table => 'ns.tbl')") +# Orphan removal REQUIRES S3 credentials (vended creds fail with NoAuthWithAWSException) +spark.sql("CALL r2dc.system.remove_orphan_files(table => 'ns.tbl', older_than => TIMESTAMP '2026-02-28 00:00:00')") ``` -## Metadata Inspection - -```python -table = catalog.load_table(("logs", "app_logs")) -print(table.schema()) -print(table.current_snapshot()) -print(table.properties) -print(f"Files: {len(table.scan().plan_files())}") -``` - -## Error Codes - -| Code | Meaning | Common Causes | -|------|---------|---------------| -| 401 | Unauthorized | Invalid/missing token | -| 404 | Not Found | Catalog not enabled, namespace/table missing | -| 409 | Conflict | Already exists, concurrent update | -| 422 | Validation | Invalid schema, incompatible type | +## See Also -See [gotchas.md](gotchas.md) for detailed troubleshooting. +- [configuration.md](configuration.md) · [patterns.md](patterns.md) · [gotchas.md](gotchas.md) diff --git a/skills/cloudflare/references/r2-data-catalog/configuration.md b/skills/cloudflare/references/r2-data-catalog/configuration.md index 15915da..b8905dc 100644 --- a/skills/cloudflare/references/r2-data-catalog/configuration.md +++ b/skills/cloudflare/references/r2-data-catalog/configuration.md @@ -1,198 +1,98 @@ -# Configuration +# R2 Data Catalog Configuration -How to enable R2 Data Catalog and configure authentication. +Enable the catalog, create tokens, turn on automatic maintenance, connect clients. For exhaustive token/permission options and maintenance settings, pull `https://developers.cloudflare.com/r2/data-catalog/manage-catalogs/` and `.../table-maintenance/`. -## Prerequisites - -- Cloudflare account with [R2 subscription](https://developers.cloudflare.com/r2/pricing/) -- R2 bucket created -- Access to Cloudflare dashboard or Wrangler CLI - -## Enable Catalog on Bucket - -Choose one method: - -### Via Wrangler (Recommended) +## Step 1: Create Bucket + Enable Catalog ```bash -npx wrangler r2 bucket catalog enable +npx wrangler r2 bucket create my-bucket +npx wrangler r2 bucket catalog enable my-bucket ``` -**Output:** +`enable` outputs the two values used everywhere: + ``` -✅ Data Catalog enabled for bucket 'my-bucket' - Catalog URI: https://.r2.cloudflarestorage.com/iceberg/my-bucket - Warehouse: my-bucket +Warehouse: 4482a1cd43bf5197657ae1d8636c414a_my-bucket # {ACCOUNT_ID}_{BUCKET} +Catalog URI: https://catalog.cloudflarestorage.com/4482a1cd43bf5197657ae1d8636c414a/my-bucket ``` -### Via Dashboard +Enabling creates `__r2_data_catalog/` metadata in the bucket; existing objects are untouched. -1. Navigate to **R2** → Select your bucket → **Settings** tab -2. Scroll to "R2 Data Catalog" section → Click **Enable** -3. Note the **Catalog URI** and **Warehouse name** shown +## Step 2: Create an API Token -**Result:** -- Catalog URI: `https://.r2.cloudflarestorage.com/iceberg/` -- Warehouse: `` (same as bucket name) +Dashboard → **R2** → **Manage R2 API tokens** → **Create API token**. -### Via API (Programmatic) +**Simplest:** one token with **R2 Storage Admin Read & Write** + **R2 Data Catalog Read & Write**, scoped to your bucket(s). Add **R2 SQL Read** if you also query. This token works for the Iceberg REST API, control-plane API, R2 SQL, and GraphQL Analytics. Token creation also yields S3 Access Key ID / Secret (needed only for Spark orphan-file removal). -```bash -curl -X POST \ - "https://api.cloudflare.com/client/v4/accounts//r2/buckets//catalog" \ - -H "Authorization: Bearer " \ - -H "Content-Type: application/json" -``` +> Open-beta limitation: R2 Storage **Admin Write is required even for read-only data access**. See the manage-catalogs doc for the current permission matrix. -**Response:** -```json -{ - "result": { - "catalog_uri": "https://.r2.cloudflarestorage.com/iceberg/", - "warehouse": "" - }, - "success": true -} -``` +## Step 3: Enable Automatic Maintenance (Recommended) -## Check Catalog Status +R2 Data Catalog runs compaction and snapshot expiration for you. ```bash -npx wrangler r2 bucket catalog status -``` - -**Output:** -``` -Catalog Status: enabled -Catalog URI: https://.r2.cloudflarestorage.com/iceberg/my-bucket -Warehouse: my-bucket -``` - -## Disable Catalog (If Needed) +# Compaction — merges small files (target size MB; default 128) +npx wrangler r2 bucket catalog compaction enable my-bucket \ + --target-size 128 --token $API_TOKEN -```bash -npx wrangler r2 bucket catalog disable +# Snapshot expiration — removes old snapshots AND their unreferenced data files +npx wrangler r2 bucket catalog snapshot-expiration enable my-bucket \ + --token $API_TOKEN --older-than-days 7 --retain-last 10 ``` -⚠️ **Warning:** Disabling does NOT delete tables/data. Files remain in bucket. Metadata becomes inaccessible until re-enabled. - -## API Token Creation - -R2 Data Catalog requires API token with **both** R2 Storage + R2 Data Catalog permissions. - -### Dashboard Method (Recommended) +Compaction needs a **stored credential** to access files. `compaction enable` (and the dashboard wizard) stores it automatically; pure-API setups must call `/credential` (see [api.md](api.md)). -1. Go to **R2** → **Manage R2 API Tokens** → **Create API Token** -2. Select permission level: - - **Admin Read & Write** - Full catalog + storage access (read/write) - - **Admin Read only** - Read-only access (for query engines) -3. Copy token value immediately (shown only once) +> Compaction triggers **hourly** with **no hard throughput cap** (the former 2 GB/hour limit was lifted). Snapshot expiration deletes unreferenced data files automatically (since April 2026) — manual orphan cleanup is rarely needed. For target-size guidance per workload, see the table-maintenance doc. -**Permission groups included:** -- `Workers R2 Data Catalog Write` (or Read) -- `Workers R2 Storage Bucket Item Write` (or Read) +## Step 4: Verify -### API Method (Programmatic) +```bash +npx wrangler r2 bucket catalog status my-bucket +# or control-plane API: +curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET" \ + -H "Authorization: Bearer $API_TOKEN" +``` -Use Cloudflare API to create tokens programmatically. Required permissions: -- `Workers R2 Data Catalog Write` (or Read) -- `Workers R2 Storage Bucket Item Write` (or Read) +Expect `"status": "active"`, `compaction.state: "enabled"`, `credential_status: "present"`. -## Client Configuration +## Client Connection ### PyIceberg -```python -from pyiceberg.catalog.rest import RestCatalog - -catalog = RestCatalog( - name="my_catalog", - warehouse="", # Same as bucket name - uri="", # From enable command - token="", # From token creation -) -``` - -**Full example with credentials:** ```python import os from pyiceberg.catalog.rest import RestCatalog -# Store credentials in environment variables -WAREHOUSE = os.getenv("R2_WAREHOUSE") # e.g., "my-bucket" -CATALOG_URI = os.getenv("R2_CATALOG_URI") # e.g., "https://abc123.r2.cloudflarestorage.com/iceberg/my-bucket" -TOKEN = os.getenv("R2_TOKEN") # API token - catalog = RestCatalog( name="r2_catalog", - warehouse=WAREHOUSE, - uri=CATALOG_URI, - token=TOKEN, + warehouse=os.environ["R2_WAREHOUSE"], # {ACCOUNT_ID}_{BUCKET} + uri=os.environ["R2_CATALOG_URI"], # https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET} + token=os.environ["R2_TOKEN"], ) - -# Test connection -print(catalog.list_namespaces()) +print(catalog.list_namespaces()) # connection test ``` -### Spark / Trino / DuckDB +### PySpark / DuckDB / Trino / Snowflake -See [patterns.md](patterns.md) for integration examples with other query engines. - -## Connection String Format - -For quick reference: - -``` -Catalog URI: https://.r2.cloudflarestorage.com/iceberg/ -Warehouse: -Token: -``` - -**Where to find values:** - -| Value | Source | -|-------|--------| -| `` | Dashboard URL or `wrangler whoami` | -| `` | R2 bucket name | -| Catalog URI | Output from `wrangler r2 bucket catalog enable` | -| Token | R2 API Token creation page | - -## Security Best Practices - -1. **Store tokens securely** - Use environment variables or secret managers, never hardcode -2. **Use least privilege** - Read-only tokens for query engines, write tokens only where needed -3. **Rotate tokens regularly** - Create new tokens, test, then revoke old ones -4. **One token per application** - Easier to track and revoke if compromised -5. **Monitor token usage** - Check R2 analytics for unexpected patterns -6. **Bucket-scoped tokens** - Create tokens per bucket, not account-wide +Full, current engine configs live at `https://developers.cloudflare.com/r2/data-catalog/config-examples/`. A verified PySpark session template is in [patterns.md](patterns.md#pyspark-session) (needs Iceberg 1.6.1 and `X-Iceberg-Access-Delegation: vended-credentials`). ## Environment Variables Pattern ```bash # .env (never commit) -R2_CATALOG_URI=https://.r2.cloudflarestorage.com/iceberg/ -R2_WAREHOUSE= +R2_CATALOG_URI=https://catalog.cloudflarestorage.com// +R2_WAREHOUSE=_ R2_TOKEN= ``` -```python -import os -from pyiceberg.catalog.rest import RestCatalog +## Disable Catalog -catalog = RestCatalog( - name="r2", - uri=os.getenv("R2_CATALOG_URI"), - warehouse=os.getenv("R2_WAREHOUSE"), - token=os.getenv("R2_TOKEN"), -) +```bash +npx wrangler r2 bucket catalog disable my-bucket ``` -## Troubleshooting +Preserves data and metadata; tables become inaccessible via the catalog until re-enabled. -| Problem | Solution | -|---------|----------| -| 404 "catalog not found" | Run `wrangler r2 bucket catalog enable ` | -| 401 "unauthorized" | Check token has both Catalog + Storage permissions | -| 403 on data files | Token needs both permission groups | +## See Also -See [gotchas.md](gotchas.md) for detailed troubleshooting. +- [api.md](api.md) — control-plane + PyIceberg API · [gotchas.md](gotchas.md) — auth & maintenance troubleshooting diff --git a/skills/cloudflare/references/r2-data-catalog/gotchas.md b/skills/cloudflare/references/r2-data-catalog/gotchas.md index 6bfad9e..4761a12 100644 --- a/skills/cloudflare/references/r2-data-catalog/gotchas.md +++ b/skills/cloudflare/references/r2-data-catalog/gotchas.md @@ -1,170 +1,55 @@ -# Gotchas & Troubleshooting +# R2 Data Catalog Gotchas -Common problems → causes → solutions. +Common failure modes and operational behavior. For limits, recommendations, and supported settings, pull `https://developers.cloudflare.com/r2/data-catalog/` and `.../table-maintenance/`. -## Permission Errors +## Connection / Auth -### 401 Unauthorized +- **Catalog URI / warehouse mismatch (most common).** Copy both values exactly from `wrangler r2 bucket catalog enable` (Catalog URI `https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}`, warehouse `{ACCOUNT_ID}_{BUCKET}`). Mismatched values fail to connect. +- **401 Unauthorized** — token lacks Data Catalog R&W. Test with `catalog.list_namespaces()`. +- **403 on data files** — token lacks R2 Storage. Open beta requires **Admin Read & Write on R2 Storage even for read-only** data access. +- **`/config` "Warehouse name missing in query param"** — the Iceberg `/v1/config` route needs `?warehouse={ACCOUNT_ID}_{BUCKET}`. PyIceberg/PySpark add it automatically when you set `warehouse=`. -**Error:** `"401 Unauthorized"` -**Cause:** Token missing R2 Data Catalog permissions. -**Solution:** Use "Admin Read & Write" token (includes catalog + storage permissions). Test with `catalog.list_namespaces()`. +## Maintenance Behavior (updated) -### 403 Forbidden +- **No throughput cap on compaction.** The former 2 GB/hour/table limit is **lifted** — compaction triggers hourly and processes the backlog with no hard cap. Large small-file backlogs still take multiple hourly cycles. +- **Snapshot expiration deletes data files** (since April 2026), not just metadata. Manual `remove_orphan_files` is rarely needed. +- **Compaction requires a stored credential.** `wrangler ... compaction enable` and the dashboard wizard store it automatically; pure-API setups must POST `/credential`. +- Compaction is **Parquet-only**. -**Error:** `"403 Forbidden"` on data files -**Cause:** Token lacks storage permissions. -**Solution:** Token needs both R2 Data Catalog + R2 Storage Bucket Item permissions. +## Tables & Schema -### Token Rotation Issues +- `TableAlreadyExistsError` / `NamespaceAlreadyExistsError` → use `create_*_if_not_exists` / load existing. +- `422 Validation` on schema update → only add nullable columns and widen types (int→long, float→double). +- `TypeError: Cannot cast` on append → PyArrow type ≠ Iceberg schema; cast to int64 (Iceberg default); check `table.schema()`. -**Error:** New token fails after rotation. -**Solution:** Create new token → test in staging → update prod → monitor 24h → revoke old. +## Concurrency -## Catalog URI Issues +- `CommitFailedException` → optimistic-locking conflict; retry with backoff (see [patterns.md](patterns.md#concurrent-writes-with-retry-pyiceberg)). +- Stale metadata after external writes → reload: `table = catalog.load_table(("ns","tbl"))`. -### 404 Not Found +## PySpark / Iceberg -**Error:** `"404 Catalog not found"` -**Cause:** Catalog not enabled or wrong URI. -**Solution:** Run `wrangler r2 bucket catalog enable `. URI must be HTTPS with `/iceberg/` and case-sensitive bucket name. +| Issue | Fix | +|-------|-----| +| Catalog auth fails | Add header `X-Iceberg-Access-Delegation: vended-credentials` | +| `NoAuthWithAWSException` on orphan removal | Supply S3 access/secret keys (vended creds don't work here) | +| Version mismatch | Use Iceberg `1.6.1` | +| Slow first run (~30–60s) | JAR download; cached after | +| Remote signing errors | Set `s3.remote-signing-enabled=false` | -### Wrong Warehouse +## Nested Namespaces -**Error:** Cannot create/load tables. -**Cause:** Warehouse ≠ bucket name. -**Solution:** Set `warehouse="bucket-name"` to match bucket exactly. +Control-plane URL separator for nested namespaces is **`%1F`** (Unit Separator), not `/` or `.`: `/namespaces/parent%1Fchild/tables`. -## Table and Schema Issues +## Debug Checklist -### Table/Namespace Already Exists +1. `npx wrangler r2 bucket catalog status ` — enabled? +2. Token has R2 Storage (Admin R&W) + R2 Data Catalog (R&W)? +3. `catalog.list_namespaces()` succeeds? +4. Catalog URI = `catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}`, warehouse = `{ACCOUNT_ID}_{BUCKET}`? +5. Namespace created before `create_table`? +6. Compaction enabled + `credential_status: present`? -**Error:** `"TableAlreadyExistsError"` -**Solution:** Use try/except to load existing or check first. +## See Also -### Namespace Not Found - -**Error:** Cannot create table. -**Solution:** Create namespace first: `catalog.create_namespace("ns")` - -### Schema Evolution Errors - -**Error:** `"422 Validation"` on schema update. -**Cause:** Incompatible change (required field, type shrink). -**Solution:** Only add nullable columns, compatible type widening (int→long, float→double). - -## Data and Query Issues - -### Empty Scan Results - -**Error:** Scan returns no data. -**Cause:** Incorrect filter or partition column. -**Solution:** Test without filter first: `table.scan().to_pandas()`. Verify partition column names. - -### Slow Queries - -**Error:** Performance degrades over time. -**Cause:** Too many small files. -**Solution:** Check file count, compact if >1000 or avg <10MB. See [api.md](api.md#compaction). - -### Type Mismatch - -**Error:** `"Cannot cast"` on append. -**Cause:** PyArrow types don't match Iceberg schema. -**Solution:** Cast to int64 (Iceberg default), not int32. Check `table.schema()`. - -## Compaction Issues - -### Compaction Issues - -**Problem:** File count unchanged or compaction takes hours. -**Cause:** Target size too large, or table too big for PyIceberg. -**Solution:** Only compact if avg <50MB. For >1TB tables, use Spark. Run during low-traffic periods. - -## Maintenance Issues - -### Snapshot/Orphan Issues - -**Problem:** Expiration fails or orphan cleanup deletes active data. -**Cause:** Too aggressive retention or wrong order. -**Solution:** Always expire snapshots first with `retain_last=10`, then cleanup orphans with 3+ day threshold. - -## Concurrency Issues - -### Concurrent Write Conflicts - -**Problem:** `CommitFailedException` with multiple writers. -**Cause:** Optimistic locking - simultaneous commits. -**Solution:** Add retry with exponential backoff (see [patterns.md](patterns.md#pattern-6-concurrent-writes-with-retry)). - -### Stale Metadata - -**Problem:** Old schema/data after external update. -**Cause:** Cached metadata. -**Solution:** Reload table: `table = catalog.load_table(("ns", "table"))` - -## Performance Optimization - -### Performance Tips - -**Scans:** Use `row_filter` and `selected_fields` to reduce data scanned. -**Partitions:** 100-1000 optimal. Avoid high cardinality (millions) or low (<10). -**Files:** Keep 100-500MB avg. Compact if <10MB or >10k files. - -## Limits - -| Resource | Recommended | Impact if Exceeded | -|----------|-------------|-------------------| -| Tables/namespace | <10k | Slow list ops | -| Files/table | <100k | Slow query planning | -| Partitions/table | 100-1k | Metadata overhead | -| Snapshots/table | Expire >7d | Metadata bloat | - -## Common Error Messages Reference - -| Error Message | Likely Cause | Fix | -|---------------|--------------|-----| -| `401 Unauthorized` | Missing/invalid token | Check token has catalog+storage permissions | -| `403 Forbidden` | Token lacks storage permissions | Add R2 Storage Bucket Item permission | -| `404 Not Found` | Catalog not enabled or wrong URI | Run `wrangler r2 bucket catalog enable` | -| `409 Conflict` | Table/namespace already exists | Use try/except or load existing | -| `422 Unprocessable Entity` | Schema validation failed | Check type compatibility, required fields | -| `CommitFailedException` | Concurrent write conflict | Add retry logic with backoff | -| `NamespaceAlreadyExistsError` | Namespace exists | Use try/except or load existing | -| `NoSuchTableError` | Table doesn't exist | Check namespace+table name, create first | -| `TypeError: Cannot cast` | PyArrow type mismatch | Cast data to match Iceberg schema | - -## Debugging Checklist - -When things go wrong, check in order: - -1. ✅ **Catalog enabled:** `npx wrangler r2 bucket catalog status ` -2. ✅ **Token permissions:** Both R2 Data Catalog + R2 Storage in dashboard -3. ✅ **Connection test:** `catalog.list_namespaces()` succeeds -4. ✅ **URI format:** HTTPS, includes `/iceberg/`, correct bucket name -5. ✅ **Warehouse name:** Matches bucket name exactly -6. ✅ **Namespace exists:** Create before `create_table()` -7. ✅ **Enable debug logging:** `logging.basicConfig(level=logging.DEBUG)` -8. ✅ **PyIceberg version:** `pip install --upgrade pyiceberg` (≥0.5.0) -9. ✅ **File health:** Compact if >1000 files or avg <10MB -10. ✅ **Snapshot count:** Expire if >100 snapshots - -## Enable Debug Logging - -```python -import logging -logging.basicConfig(level=logging.DEBUG) -# Now operations show HTTP requests/responses -``` - -## Resources - -- [Cloudflare Community](https://community.cloudflare.com/c/developers/workers/40) -- [Cloudflare Discord](https://discord.cloudflare.com) - #r2 channel -- [PyIceberg GitHub](https://github.com/apache/iceberg-python/issues) -- [Apache Iceberg Slack](https://iceberg.apache.org/community/) - -## Next Steps - -- [patterns.md](patterns.md) - Working examples -- [api.md](api.md) - API reference +- [configuration.md](configuration.md) · [api.md](api.md) · [patterns.md](patterns.md) diff --git a/skills/cloudflare/references/r2-data-catalog/patterns.md b/skills/cloudflare/references/r2-data-catalog/patterns.md index b6b181f..1a80c6e 100644 --- a/skills/cloudflare/references/r2-data-catalog/patterns.md +++ b/skills/cloudflare/references/r2-data-catalog/patterns.md @@ -1,133 +1,102 @@ -# Common Patterns +# R2 Data Catalog Patterns -Practical patterns for R2 Data Catalog with PyIceberg. +Code templates with PyIceberg (lightweight, no JVM) and PySpark (full Iceberg ecosystem). For per-engine config (DuckDB, Trino, Snowflake, StarRocks) and partitioning/maintenance best practices, pull `https://developers.cloudflare.com/r2/data-catalog/config-examples/` and `.../table-maintenance/`. -## PyIceberg Connection +| Need | Tool | +|------|------| +| Catalog ops, append/scan, small-medium loads | PyIceberg | +| Batch ETL, INSERT INTO SELECT, DELETE/MERGE, write-back, >1 TB maintenance | PySpark | +| Pure SQL analytics (no writes) | [R2 SQL](../r2-sql/) | + +## PyIceberg: Connect, Create, Load ```python -import os +import os, pyarrow as pa from pyiceberg.catalog.rest import RestCatalog -from pyiceberg.exceptions import NamespaceAlreadyExistsError catalog = RestCatalog( - name="r2_catalog", - warehouse=os.getenv("R2_WAREHOUSE"), # bucket name - uri=os.getenv("R2_CATALOG_URI"), # catalog endpoint - token=os.getenv("R2_TOKEN"), # API token + name="r2", + warehouse=os.environ["R2_WAREHOUSE"], # {ACCOUNT_ID}_{BUCKET} + uri=os.environ["R2_CATALOG_URI"], # https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET} + token=os.environ["R2_TOKEN"], ) +catalog.create_namespace_if_not_exists("analytics") -# Create namespace (idempotent) -try: - catalog.create_namespace("default") -except NamespaceAlreadyExistsError: - pass +schema = pa.schema([("id", pa.int64()), ("name", pa.string()), ("amount", pa.float64())]) +table = catalog.create_table(("analytics", "events"), schema=schema) +table.append(pa.table({"id": [1, 2], "name": ["a", "b"], "amount": [80.0, 92.5]})) +print(table.scan().to_arrow().to_pandas()) ``` -## Pattern 1: Log Analytics Pipeline - -Ingest logs incrementally, query by time/level. +## PyIceberg: Partitioned Time-Series Table ```python -import pyarrow as pa -from datetime import datetime from pyiceberg.schema import Schema -from pyiceberg.types import NestedField, TimestampType, StringType, IntegerType +from pyiceberg.types import NestedField, TimestampType, StringType from pyiceberg.partitioning import PartitionSpec, PartitionField from pyiceberg.transforms import DayTransform -# Create partitioned table (once) schema = Schema( NestedField(1, "timestamp", TimestampType(), required=True), NestedField(2, "level", StringType(), required=True), - NestedField(3, "service", StringType(), required=True), - NestedField(4, "message", StringType(), required=False), + NestedField(3, "message", StringType(), required=False), ) - -partition_spec = PartitionSpec( - PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day") -) - -catalog.create_namespace("logs") -table = catalog.create_table(("logs", "app_logs"), schema=schema, partition_spec=partition_spec) - -# Append logs (incremental) -data = pa.table({ - "timestamp": [datetime(2026, 1, 27, 10, 30, 0)], - "level": ["ERROR"], - "service": ["auth-service"], - "message": ["Failed login"], -}) -table.append(data) - -# Query by time + level (leverages partitioning) -scan = table.scan(row_filter="level = 'ERROR' AND day = '2026-01-27'") -errors = scan.to_pandas() +spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day")) +table = catalog.create_table(("logs", "app_logs"), schema=schema, partition_spec=spec) +errors = table.scan(row_filter="level = 'ERROR'").to_pandas() # partition pruning ``` -## Pattern 2: Time-Travel Queries +## PySpark Session -```python -from datetime import datetime, timedelta - -table = catalog.load_table(("logs", "app_logs")) - -# Query specific snapshot -snapshot_id = table.current_snapshot().snapshot_id -data = table.scan(snapshot_id=snapshot_id).to_pandas() - -# Query as of timestamp (yesterday) -yesterday_ms = int((datetime.now() - timedelta(days=1)).timestamp() * 1000) -data = table.scan(as_of_timestamp=yesterday_ms).to_pandas() -``` - -## Pattern 3: Schema Evolution +Verified template — requires Iceberg **1.6.1** and vended credentials. S3 keys are only needed for orphan-file removal. (If this drifts, cross-check `config-examples/spark-python/`.) ```python -from pyiceberg.types import StringType - -table = catalog.load_table(("users", "profiles")) - -with table.update_schema() as update: - update.add_column("email", StringType(), required=False) - update.rename_column("name", "full_name") -# Old readers ignore new columns, new readers see nulls for old data +from pyspark.sql import SparkSession + +spark = SparkSession.builder \ + .appName("R2DataCatalog") \ + .config('spark.jars.packages', + 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,' + 'org.apache.iceberg:iceberg-aws-bundle:1.6.1,' + 'org.apache.hadoop:hadoop-aws:3.3.4,' + 'com.amazonaws:aws-java-sdk-bundle:1.12.262') \ + .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \ + .config("spark.sql.catalog.r2dc", "org.apache.iceberg.spark.SparkCatalog") \ + .config("spark.sql.catalog.r2dc.type", "rest") \ + .config("spark.sql.catalog.r2dc.uri", CATALOG_URI) \ + .config("spark.sql.catalog.r2dc.warehouse", WAREHOUSE) \ + .config("spark.sql.catalog.r2dc.token", TOKEN) \ + .config("spark.sql.catalog.r2dc.header.X-Iceberg-Access-Delegation", "vended-credentials") \ + .config("spark.sql.catalog.r2dc.s3.remote-signing-enabled", "false") \ + .config("spark.sql.defaultCatalog", "r2dc") \ + .config("spark.hadoop.fs.s3a.access.key", S3_ACCESS_KEY) \ + .config("spark.hadoop.fs.s3a.secret.key", S3_SECRET_KEY) \ + .config("spark.hadoop.fs.s3a.endpoint", S3_ENDPOINT) \ + .config("spark.hadoop.fs.s3a.path.style.access", "true") \ + .getOrCreate() +spark.sql("USE r2dc") ``` -## Pattern 4: Partitioned Tables +> `X-Iceberg-Access-Delegation: vended-credentials` is required; `s3.remote-signing-enabled` must be `false`. First startup ~30–60s for JAR downloads (cached after). -```python -from pyiceberg.partitioning import PartitionSpec, PartitionField -from pyiceberg.transforms import DayTransform, IdentityTransform - -# Partition by day + country -partition_spec = PartitionSpec( - PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day"), - PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="country"), -) -table = catalog.create_table(("events", "user_events"), schema=schema, partition_spec=partition_spec) - -# Queries prune partitions automatically -scan = table.scan(row_filter="country = 'US' AND day = '2026-01-27'") -``` - -## Pattern 5: Table Maintenance +## PySpark: Batch ETL ```python -from datetime import datetime, timedelta - -table = catalog.load_table(("logs", "app_logs")) - -# Compact → expire → cleanup (in order) -table.rewrite_data_files(target_file_size_bytes=128 * 1024 * 1024) -seven_days_ms = int((datetime.now() - timedelta(days=7)).timestamp() * 1000) -table.expire_snapshots(older_than=seven_days_ms, retain_last=10) -three_days_ms = int((datetime.now() - timedelta(days=3)).timestamp() * 1000) -table.delete_orphan_files(older_than=three_days_ms) +spark.sql(""" +CREATE TABLE IF NOT EXISTS my_ns.events ( + __ingest_ts TIMESTAMP, event_id STRING, category STRING, amount DOUBLE +) PARTITIONED BY (days(__ingest_ts)) +""") + +spark.read.option("header","true").csv("data.csv").writeTo("my_ns.events").append() +spark.read.parquet("data.parquet").writeTo("my_ns.events").append() +spark.sql("INSERT INTO my_ns.target SELECT col1, col2 FROM my_ns.source WHERE col1 > 0") +spark.sql("DELETE FROM my_ns.events WHERE amount < 0") ``` -See [api.md](api.md#table-maintenance) for detailed parameters. +> Partition large tables (`PARTITIONED BY (days(__ingest_ts))`). Unpartitioned works for small datasets (<1000 files) but degrades at scale. -## Pattern 6: Concurrent Writes with Retry +## Concurrent Writes with Retry (PyIceberg) ```python from pyiceberg.exceptions import CommitFailedException @@ -136,56 +105,18 @@ import time def append_with_retry(table, data, max_retries=3): for attempt in range(max_retries): try: - table.append(data) - return + table.append(data); return except CommitFailedException: - if attempt == max_retries - 1: - raise + if attempt == max_retries - 1: raise time.sleep(2 ** attempt) ``` -## Pattern 7: Upsert Simulation - -```python -import pandas as pd -import pyarrow as pa - -# Read → merge → overwrite (not atomic, use Spark MERGE INTO for production) -existing = table.scan().to_pandas() -new_data = pd.DataFrame({"id": [1, 3], "value": [100, 300]}) -merged = pd.concat([existing, new_data]).drop_duplicates(subset=["id"], keep="last") -table.overwrite(pa.Table.from_pandas(merged)) -``` - -## Pattern 8: DuckDB Integration - -```python -import duckdb - -arrow_table = table.scan().to_arrow() -con = duckdb.connect() -con.register("logs", arrow_table) -result = con.execute("SELECT level, COUNT(*) FROM logs GROUP BY level").fetchdf() -``` - -## Pattern 9: Monitor Table Health +Optimistic locking: concurrent commits to the same table may conflict; different-partition writes are safe. -```python -files = table.scan().plan_files() -avg_mb = sum(f.file_size_in_bytes for f in files) / len(files) / (1024**2) -print(f"Files: {len(files)}, Avg: {avg_mb:.1f}MB, Snapshots: {len(table.snapshots())}") +## Connecting Any Iceberg Engine -if avg_mb < 10 or len(files) > 1000: - print("⚠️ Needs compaction") -``` +Engines connect with the Iceberg REST catalog config — Catalog URI `https://catalog.cloudflarestorage.com/{ACCOUNT_ID}/{BUCKET}`, warehouse `{ACCOUNT_ID}_{BUCKET}`, your token, and header `X-Iceberg-Access-Delegation: vended-credentials`. Copy-paste configs per engine: `config-examples/`. -## Best Practices +## See Also -| Area | Guideline | -|------|-----------| -| **Partitioning** | Use day/hour for time-series; 100-1000 partitions; avoid high cardinality | -| **File sizes** | Target 128-512MB; compact when avg <10MB or >10k files | -| **Schema** | Add columns as nullable (`required=False`); batch changes | -| **Maintenance** | Compact high-write daily/weekly; expire snapshots 7-30d; cleanup orphans after | -| **Concurrency** | Reads automatic; writes to different partitions safe; retry same partition | -| **Performance** | Filter on partitions; select only needed columns; batch appends 100MB+ | +- [api.md](api.md) · [gotchas.md](gotchas.md) · [pipelines/patterns.md](../pipelines/patterns.md) · [r2-sql/patterns.md](../r2-sql/patterns.md) diff --git a/skills/cloudflare/references/r2-sql/README.md b/skills/cloudflare/references/r2-sql/README.md index c59a161..03b3f6e 100644 --- a/skills/cloudflare/references/r2-sql/README.md +++ b/skills/cloudflare/references/r2-sql/README.md @@ -1,128 +1,64 @@ -# Cloudflare R2 SQL Skill Reference +# Cloudflare R2 SQL -Expert guidance for Cloudflare R2 SQL - serverless distributed query engine for Apache Iceberg tables. +Serverless, distributed, **read-only** query engine (Apache DataFusion) for Apache Iceberg tables in R2 Data Catalog. -## Reading Order - -**New to R2 SQL?** Start here: -1. Read "What is R2 SQL?" and "When to Use" below -2. [configuration.md](configuration.md) - Enable catalog, create tokens -3. [patterns.md](patterns.md) - Wrangler CLI and integration examples -4. [api.md](api.md) - SQL syntax and query reference -5. [gotchas.md](gotchas.md) - Limitations and troubleshooting - -**Quick reference?** Jump to: -- [Run a query via Wrangler](patterns.md#wrangler-cli-query) -- [SQL syntax reference](api.md#sql-syntax) -- [ORDER BY limitations](gotchas.md#order-by-limitations) - -## What is R2 SQL? - -R2 SQL is Cloudflare's **serverless distributed analytics query engine** for querying Apache Iceberg tables in R2 Data Catalog. Features: - -- **Serverless** - No clusters to manage, no infrastructure -- **Distributed** - Leverages Cloudflare's global network for parallel execution -- **SQL interface** - Familiar SQL syntax for analytics queries -- **Zero egress fees** - Query from any cloud/region without data transfer costs -- **Open beta** - Free during beta (standard R2 storage costs apply) - -### What is Apache Iceberg? - -Open table format for large-scale analytics datasets in object storage: -- **ACID transactions** - Safe concurrent reads/writes -- **Metadata optimization** - Fast queries without full table scans -- **Schema evolution** - Add/rename/drop columns without rewrites -- **Partitioning** - Organize data for efficient pruning - -## When to Use +## Documentation -**Use R2 SQL for:** -- **Log analytics** - Query application/system logs with WHERE filters and aggregations -- **BI dashboards** - Generate reports from large analytical datasets -- **Fraud detection** - Analyze transaction patterns with GROUP BY/HAVING -- **Multi-cloud analytics** - Query data from any cloud without egress fees -- **Ad-hoc exploration** - Run SQL queries on Iceberg tables via Wrangler CLI +For full function lists, data types, and pricing, **retrieve the live docs** — use the `cloudflare-docs` MCP/search tool if available, otherwise `webfetch`. -**Don't use R2 SQL for:** -- **Workers/Pages runtime** - R2 SQL has no Workers binding, use HTTP API from external systems -- **Real-time queries (<100ms)** - Optimized for analytical batch queries, not OLTP -- **Complex joins/CTEs** - Limited SQL feature set (no JOINs, subqueries, CTEs currently) -- **Small datasets (<1GB)** - Setup overhead not justified +| Topic | URL | +|-------|-----| +| Overview / get started | `https://developers.cloudflare.com/r2-sql/get-started/` | +| Query data | `https://developers.cloudflare.com/r2-sql/query-data/` | +| SQL reference | `https://developers.cloudflare.com/r2-sql/sql-reference/` | +| Aggregate functions | `https://developers.cloudflare.com/r2-sql/sql-reference/aggregate-functions/` | +| Scalar functions | `https://developers.cloudflare.com/r2-sql/sql-reference/scalar-functions/` | +| Complex types | `https://developers.cloudflare.com/r2-sql/sql-reference/complex-types/` | +| Limitations & best practices | `https://developers.cloudflare.com/r2-sql/reference/limitations-best-practices/` | +| Wrangler commands | `https://developers.cloudflare.com/r2-sql/reference/wrangler-commands/` | +| Pricing | `https://developers.cloudflare.com/r2-sql/platform/pricing/` | -## Decision Tree: Need to Query R2 Data? +## Connection Values -``` -Do you need to query structured data in R2? -├─ YES, data is in Iceberg tables -│ ├─ Need SQL interface? → Use R2 SQL (this reference) -│ ├─ Need Python API? → See r2-data-catalog reference (PyIceberg) -│ └─ Need other engine? → See r2-data-catalog reference (Spark, Trino, etc.) -│ -├─ YES, but not in Iceberg format -│ ├─ Streaming data? → Use Pipelines to write to Data Catalog, then R2 SQL -│ └─ Static files? → Use PyIceberg to create Iceberg tables, then R2 SQL -│ -└─ NO, just need object storage → Use R2 reference (not R2 SQL) -``` - -## Architecture Overview - -**Query Planner:** -- Top-down metadata investigation with multi-layer pruning -- Partition-level, column-level, and row-group pruning -- Streaming pipeline - execution starts before planning completes -- Early termination with LIMIT - stops when result complete - -**Query Execution:** -- Coordinator distributes work to workers across Cloudflare network -- Workers run Apache DataFusion for parallel query execution -- Parquet column pruning - reads only required columns -- Ranged reads from R2 for efficiency +| Value | Format | +|-------|--------| +| REST endpoint | `https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/r2-sql/query/{BUCKET}` | +| Wrangler | `npx wrangler r2 sql query "{WAREHOUSE}" ""` with `WRANGLER_R2_SQL_AUTH_TOKEN` set | +| Warehouse | `{ACCOUNT_ID}_{BUCKET}` | -**Aggregation Strategies:** -- Scatter-gather - simple aggregations (SUM, COUNT, AVG) -- Shuffling - ORDER BY/HAVING on aggregates via hash partitioning +> The REST endpoint is `api.sql.cloudflarestorage.com` — **not** `api.cloudflare.com/.../r2/sql`. ## Quick Start ```bash -# 1. Enable R2 Data Catalog on bucket -npx wrangler r2 bucket catalog enable my-bucket +npx wrangler r2 bucket catalog enable my-bucket # 1. enable catalog +export WRANGLER_R2_SQL_AUTH_TOKEN= # 2. auth (Admin R&W + R2 SQL Read) +npx wrangler r2 sql query "$ACCOUNT_ID"_my-bucket \ + "SELECT * FROM default.my_table LIMIT 10" # 3. query +``` -# 2. Create API token (Admin Read & Write) -# Dashboard: R2 → Manage API tokens → Create API token +## SQL Surface -# 3. Set environment variable -export WRANGLER_R2_SQL_AUTH_TOKEN= +R2 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. -# 4. Run query -npx wrangler r2 sql query "my-bucket" "SELECT * FROM default.my_table LIMIT 10" -``` +## When to Use -## Important Limitations +**Use for:** SQL analytics over Iceberg (logs, BI, fraud, ad-hoc), multi-cloud queries without egress, dashboards (query from a Worker via HTTP). -**CRITICAL: No Workers Binding** -- R2 SQL cannot be called directly from Workers/Pages code -- For programmatic access, use HTTP API from external systems -- Or query via PyIceberg, Spark, etc. (see r2-data-catalog reference) +**Don't use for:** writes (use PySpark/PyIceberg) or real-time OLTP (<100 ms). -**SQL Feature Set:** -- No JOINs, CTEs, subqueries, window functions -- ORDER BY supports aggregation columns (not just partition keys) -- LIMIT max 10,000 (default 500) -- See [gotchas.md](gotchas.md) for complete limitations +## No Workers Binding -## In This Reference +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 -- **[configuration.md](configuration.md)** - Enable catalog, create API tokens -- **[api.md](api.md)** - SQL syntax, functions, operators, data types -- **[patterns.md](patterns.md)** - Wrangler CLI, HTTP API, Pipelines, PyIceberg -- **[gotchas.md](gotchas.md)** - Limitations, troubleshooting, performance tips +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 -- [r2-data-catalog](../r2-data-catalog/) - PyIceberg, REST API, external engines -- [pipelines](../pipelines/) - Streaming ingestion to Iceberg tables -- [r2](../r2/) - R2 object storage fundamentals -- [Cloudflare R2 SQL Docs](https://developers.cloudflare.com/r2-sql/) -- [R2 SQL Deep Dive Blog](https://blog.cloudflare.com/r2-sql-deep-dive/) +- [r2-data-catalog](../r2-data-catalog/) — PyIceberg/PySpark, table management +- [pipelines](../pipelines/) — streaming ingest into queryable tables diff --git a/skills/cloudflare/references/r2-sql/SKILL.md.backup b/skills/cloudflare/references/r2-sql/SKILL.md.backup deleted file mode 100644 index c1a3584..0000000 --- a/skills/cloudflare/references/r2-sql/SKILL.md.backup +++ /dev/null @@ -1,512 +0,0 @@ -# Cloudflare R2 SQL Skill - -Guide for using Cloudflare R2 SQL - serverless distributed query engine for Apache Iceberg tables in R2 Data Catalog. - -## Overview - -R2 SQL is Cloudflare's serverless distributed analytics query engine for querying Apache Iceberg tables in R2 Data Catalog. Features: -- Serverless - no clusters to manage -- Distributed - leverages Cloudflare's global network -- Zero egress fees - query from any cloud/region -- Open beta - free during beta (standard R2 storage costs apply) - -## Core Concepts - -### Apache Iceberg Table Format -- Open table format for large-scale analytics datasets -- ACID transactions for reliable concurrent reads/writes -- Schema evolution - add/rename/drop columns without rewriting data -- Optimized metadata - avoids full table scans via indexed metadata -- Supported by Spark, Trino, Snowflake, DuckDB, ClickHouse, PyIceberg - -### R2 Data Catalog -- Managed Apache Iceberg catalog built into R2 bucket -- Exposes standard Iceberg REST catalog interface -- Single source of truth for table metadata -- Tracks table state via immutable snapshots -- Supports multiple query engines safely accessing same tables - -### Architecture -**Query Planner**: -- Top-down metadata investigation -- Multi-layer pruning (partition-level, column-level, row-group level) -- Streaming pipeline - execution starts before planning completes -- Early termination - stops when result complete without full scan -- Uses partition stats and column stats (min/max, null counts) - -**Query Execution**: -- Coordinator distributes work to workers across Cloudflare network -- Workers run Apache DataFusion for parallel query execution -- Arrow IPC format for inter-process communication -- Parquet column pruning - reads only required columns -- Ranged reads from R2 for efficiency - -**Aggregation Strategies**: -- Scatter-gather - for simple aggregations (sum, count, avg) -- Shuffling - for ORDER BY/HAVING on aggregates via hash partitioning - -## Setup & Configuration - -### 1. Enable R2 Data Catalog - -CLI: -```bash -npx wrangler r2 bucket catalog enable -``` - -Note the Warehouse name and Catalog URI from output. - -Dashboard: -1. R2 Object Storage → Select bucket -2. Settings tab → R2 Data Catalog → Enable -3. Note Catalog URI and Warehouse name - -### 2. Create API Token - -Required permissions: R2 Admin Read & Write (includes R2 SQL Read) - -Dashboard: -1. R2 Object Storage → Manage API tokens -2. Create API token → Admin Read & Write -3. Save token value - -### 3. Configure Environment - -```bash -export WRANGLER_R2_SQL_AUTH_TOKEN= -``` - -Or `.env` file: -``` -WRANGLER_R2_SQL_AUTH_TOKEN= -``` - -## Common Code Patterns - -### Wrangler CLI Query - -```bash -npx wrangler r2 sql query "" " - SELECT * - FROM namespace.table_name - WHERE condition - LIMIT 10" -``` - -### PyIceberg Setup - -```python -from pyiceberg.catalog.rest import RestCatalog - -catalog = RestCatalog( - name="my_catalog", - warehouse="", - uri="", - token="", -) - -# Create namespace -catalog.create_namespace_if_not_exists("default") -``` - -### Create Table - -```python -import pyarrow as pa - -# Define schema -df = pa.table({ - "id": [1, 2, 3], - "name": ["Alice", "Bob", "Charlie"], - "score": [80.0, 92.5, 88.0], -}) - -# Create table -table = catalog.create_table( - ("default", "people"), - schema=df.schema, -) -``` - -### Append Data - -```python -table.append(df) -``` - -### Query Table - -```python -# Scan and convert to Pandas -scanned = table.scan().to_arrow() -print(scanned.to_pandas()) -``` - -## SQL Reference - -### Query Structure - -```sql -SELECT column_list | aggregation_function -FROM table_name -WHERE conditions -[GROUP BY column_list] -[HAVING conditions] -[ORDER BY partition_key [DESC | ASC]] -[LIMIT number] -``` - -### Schema Discovery - -```sql --- List namespaces -SHOW DATABASES; -SHOW NAMESPACES; - --- List tables -SHOW TABLES IN namespace_name; - --- Describe table -DESCRIBE namespace_name.table_name; -``` - -### SELECT Patterns - -```sql --- All columns -SELECT * FROM ns.table; - --- Specific columns -SELECT user_id, timestamp, status FROM ns.table; - --- With conditions -SELECT * FROM ns.table -WHERE timestamp BETWEEN '2025-01-01T00:00:00Z' AND '2025-01-31T23:59:59Z' - AND status = 200 -LIMIT 100; - --- Complex conditions -SELECT * FROM ns.table -WHERE (status = 404 OR status = 500) - AND method = 'POST' - AND user_agent IS NOT NULL -ORDER BY timestamp DESC; -``` - -### Aggregations - -Supported functions: COUNT(*), SUM(col), AVG(col), MIN(col), MAX(col) - -```sql --- Count by group -SELECT department, COUNT(*) -FROM ns.sales_data -GROUP BY department; - --- Multiple aggregates -SELECT region, MIN(price), MAX(price), AVG(price) -FROM ns.products -GROUP BY region -ORDER BY AVG(price) DESC; - --- With HAVING filter -SELECT category, SUM(amount) -FROM ns.sales -WHERE sale_date >= '2024-01-01' -GROUP BY category -HAVING SUM(amount) > 10000 -LIMIT 10; -``` - -### Data Types - -| Type | Description | Example | -|------|-------------|---------| -| integer | Whole numbers | 1, 42, -10 | -| float | Decimals | 1.5, 3.14 | -| string | Text (quoted) | 'hello', 'GET' | -| boolean | true/false | true, false | -| timestamp | RFC3339 | '2025-01-01T00:00:00Z' | -| date | YYYY-MM-DD | '2025-01-01' | - -### Operators - -Comparison: =, !=, <, <=, >, >=, LIKE, BETWEEN, IS NULL, IS NOT NULL -Logical: AND (higher precedence), OR (lower precedence) - -### ORDER BY Limitations - -**CRITICAL**: ORDER BY only supports partition key columns - -```sql --- Valid if timestamp is partition key -SELECT * FROM ns.logs ORDER BY timestamp DESC LIMIT 100; - --- Invalid if column not in partition key -SELECT * FROM ns.logs ORDER BY user_id; -- ERROR -``` - -### LIMIT Defaults - -- Range: 1 to 10,000 -- Default: 500 if not specified - -## Pipelines Integration - -### Create Pipeline with Data Catalog Sink - -Schema file (`schema.json`): -```json -{ - "fields": [ - {"name": "user_id", "type": "string", "required": true}, - {"name": "event_type", "type": "string", "required": true}, - {"name": "amount", "type": "float64", "required": false} - ] -} -``` - -Setup: -```bash -npx wrangler pipelines setup -``` - -Configuration: -- Pipeline name: ecommerce -- Enable HTTP endpoint: yes -- Schema: Load from file → schema.json -- Destination: Data Catalog Table -- R2 bucket: your-bucket -- Namespace: default -- Table name: events -- Catalog token: -- Compression: zstd -- Roll file time: 10 seconds (dev), 300+ (prod) - -### Send Data to Pipeline - -```bash -curl -X POST https://{stream-id}.ingest.cloudflare.com \ - -H "Content-Type: application/json" \ - -d '[ - { - "user_id": "user_123", - "event_type": "purchase", - "amount": 29.99 - } - ]' -``` - -## Common Use Cases - -### Log Analytics -- Ingest logs via Pipelines to Iceberg table -- Partition by day(timestamp) for efficient queries -- Query specific time ranges with automatic pruning -- Aggregate by status codes, endpoints, user agents - -```sql -SELECT status, COUNT(*) -FROM logs.http_requests -WHERE timestamp BETWEEN '2025-01-01T00:00:00Z' AND '2025-01-31T23:59:59Z' - AND method = 'GET' -GROUP BY status -ORDER BY COUNT(*) DESC; -``` - -### Fraud Detection -- Stream transaction events to catalog -- Query suspicious patterns with WHERE filters -- Aggregate by location, merchant, time windows - -```sql -SELECT location, COUNT(*), AVG(amount) -FROM fraud.transactions -WHERE is_fraud = true - AND transaction_timestamp >= '2025-01-01' -GROUP BY location -HAVING COUNT(*) > 10; -``` - -### Business Intelligence -- ETL data into partitioned Iceberg tables -- Run analytical queries across large datasets -- Generate reports with GROUP BY aggregations -- No egress fees when querying from BI tools - -```sql -SELECT - department, - SUM(revenue) as total_revenue, - AVG(revenue) as avg_revenue -FROM sales.transactions -WHERE sale_date >= '2024-01-01' -GROUP BY department -ORDER BY SUM(revenue) DESC -LIMIT 10; -``` - -## Performance Optimization - -### Partitioning Strategy -- Choose partition key based on common query patterns -- Typical: day(timestamp), hour(timestamp), region, category -- Enables metadata pruning to skip entire partitions -- Required for ORDER BY optimization - -### Query Optimization -- Use WHERE filters to leverage partition/column stats -- Specify LIMIT to enable early termination -- ORDER BY partition key columns only -- Filter on high-selectivity columns first - -### Data Organization -- Smaller files → slower queries (overhead) -- Larger files → better compression, fewer metadata ops -- Recommended: 100-500MB Parquet files after compression -- Use appropriate roll intervals in Pipelines (300+ seconds for prod) - -### File Pruning -Automatic at three levels: -1. Partition-level: Skip manifests not matching query -2. File-level: Skip Parquet files via column stats -3. Row-group level: Skip row groups within files - -## Iceberg Metadata Structure - -``` -bucket/ - metadata/ - snap-{id}.avro # Snapshot (points to manifest list) - {uuid}-m0.avro # Manifest file (lists data files + stats) - version-hint.text # Current metadata version - v{n}.metadata.json # Table metadata (schema, snapshots) - data/ - 00000-0-{uuid}.parquet # Data files -``` - -**Metadata hierarchy**: -1. Table metadata JSON - schema, partition spec, snapshot log -2. Snapshot - points to manifest list -3. Manifest list - partition stats for each manifest -4. Manifest files - column stats for each data file -5. Parquet files - row group stats in footer - -## Limitations & Best Practices - -### Current Limitations (Open Beta) -- ORDER BY only on partition key columns -- COUNT(*) only - COUNT(column) not supported -- No aliases in SELECT -- No subqueries, joins, or CTEs -- No nested column access -- LIMIT max 10,000 - -### Best Practices -- Partition by time dimension for time-series data -- Use BETWEEN for time ranges (leverages partition pruning) -- Combine filters with AND for better pruning -- Set appropriate LIMIT based on use case -- Use compression (zstd recommended) -- Monitor query performance and adjust partitioning - -### Type Safety -- Quote string values: 'value' -- Use RFC3339 for timestamps: '2025-01-01T00:00:00Z' -- Use YYYY-MM-DD for dates: '2025-01-01' -- No implicit type conversions - -## Connecting Other Engines - -R2 Data Catalog supports standard Iceberg REST catalog API. - -### Spark (Scala) -```scala -val spark = SparkSession.builder() - .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") - .config("spark.sql.catalog.my_catalog.catalog-impl", "org.apache.iceberg.rest.RESTCatalog") - .config("spark.sql.catalog.my_catalog.uri", catalogUri) - .config("spark.sql.catalog.my_catalog.token", token) - .config("spark.sql.catalog.my_catalog.warehouse", warehouse) - .getOrCreate() -``` - -### Snowflake -- Create external Iceberg catalog connection -- Configure with Catalog URI and R2 credentials -- Query tables via SQL interface - -### DuckDB, Trino, ClickHouse -- Supported via Iceberg REST catalog protocol -- Refer to engine-specific documentation for configuration - -## Pricing (Future) - -Currently in open beta - no charges beyond standard R2 costs. - -Planned future pricing: -- R2 storage: $0.015/GB-month -- Class A operations: $4.50/million -- Class B operations: $0.36/million -- Catalog operations: $9.00/million (create table, get metadata, etc) -- Compaction: $0.05/GB + $4.00/million objects processed -- Egress: $0 (always free) - -30+ days notice before billing begins. - -## Troubleshooting - -### Common Errors - -**"ORDER BY column not in partition key"** -- Only partition key columns can be used in ORDER BY -- Check table partition spec with DESCRIBE -- Remove ORDER BY or adjust table partitioning - -**"Token authentication failed"** -- Verify WRANGLER_R2_SQL_AUTH_TOKEN is set -- Ensure token has R2 Admin Read & Write + SQL Read permissions -- Token may be expired - create new one - -**"Table not found"** -- Verify namespace exists: SHOW DATABASES -- Check table name: SHOW TABLES IN namespace -- Ensure catalog enabled on bucket - -**"No data returned"** -- Check WHERE conditions match data -- Verify time range in BETWEEN clause -- Try removing filters to confirm data exists - -### Performance Issues - -**Slow queries**: -- Check partition pruning effectiveness -- Reduce LIMIT if scanning too much data -- Ensure filters on partition key columns -- Review Parquet file sizes (aim for 100-500MB) - -**Query timeout**: -- Add more restrictive WHERE filters -- Reduce LIMIT -- Consider better partitioning strategy - -## Resources - -- Docs: https://developers.cloudflare.com/r2-sql/ -- Data Catalog: https://developers.cloudflare.com/r2/data-catalog/ -- Blog: https://blog.cloudflare.com/r2-sql-deep-dive/ -- Discord: https://discord.cloudflare.com/ - -## Key Reminders - -1. R2 SQL queries ONLY Apache Iceberg tables in R2 Data Catalog -2. Enable catalog on bucket before use -3. Create API token with R2 + catalog permissions -4. Partition by time for time-series data -5. ORDER BY limited to partition key columns -6. Use LIMIT and WHERE for optimal performance -7. Zero egress fees - query from anywhere -8. Open beta - free during testing phase -9. Serverless - no infrastructure management -10. Leverage Cloudflare's global network for distributed execution diff --git a/skills/cloudflare/references/r2-sql/api.md b/skills/cloudflare/references/r2-sql/api.md index 7e67c8f..3411613 100644 --- a/skills/cloudflare/references/r2-sql/api.md +++ b/skills/cloudflare/references/r2-sql/api.md @@ -1,158 +1,121 @@ # R2 SQL API Reference -SQL syntax, functions, operators, and data types for R2 SQL queries. +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/`. -## SQL Syntax +## Query Endpoint -```sql -SELECT column_list | aggregation_function -FROM [namespace.]table_name -WHERE conditions -[GROUP BY column_list] -[HAVING conditions] -[ORDER BY column | aggregation_function [DESC | ASC]] -[LIMIT number] ``` - -## Schema Discovery - -```sql -SHOW DATABASES; -- List namespaces -SHOW NAMESPACES; -- Alias for SHOW DATABASES -SHOW SCHEMAS; -- Alias for SHOW DATABASES -SHOW TABLES IN namespace; -- List tables in namespace -DESCRIBE namespace.table; -- Show table schema, partition keys +POST https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/r2-sql/query/{BUCKET} +Authorization: Bearer +Content-Type: application/json +Body: {"query": ""} ``` -## SELECT Clause +CLI: `npx wrangler r2 sql query "{WAREHOUSE}" ""` (with `WRANGLER_R2_SQL_AUTH_TOKEN`). -```sql --- All columns -SELECT * FROM logs.http_requests; +## Response Format --- Specific columns -SELECT user_id, timestamp, status FROM logs.http_requests; +```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": [] +} ``` -**Limitations:** No column aliases, expressions, or nested column access - -## WHERE Clause - -### Operators - -| Operator | Example | -|----------|---------| -| `=`, `!=`, `<`, `<=`, `>`, `>=` | `status = 200` | -| `LIKE` | `user_agent LIKE '%Chrome%'` | -| `BETWEEN` | `timestamp BETWEEN '2025-01-01T00:00:00Z' AND '2025-01-31T23:59:59Z'` | -| `IS NULL`, `IS NOT NULL` | `email IS NOT NULL` | -| `AND`, `OR` | `status = 200 AND method = 'GET'` | - -Use parentheses for precedence: `(status = 404 OR status = 500) AND method = 'POST'` +Error: `{"result": null, "success": false, "errors": [{"code": 40003, "message": "..."}]}`. `bytes_scanned` ≈ billable data. -## Aggregation Functions - -| Function | Description | -|----------|-------------| -| `COUNT(*)` | Count all rows | -| `COUNT(column)` | Count non-null values | -| `COUNT(DISTINCT column)` | Count unique values | -| `SUM(column)`, `AVG(column)` | Numeric aggregations | -| `MIN(column)`, `MAX(column)` | Min/max values | +## Query Structure ```sql --- Multiple aggregations with GROUP BY -SELECT region, COUNT(*), SUM(amount), AVG(amount) -FROM sales.transactions -WHERE sale_date >= '2024-01-01' -GROUP BY region; +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 ``` -## HAVING Clause - -Filter aggregated results (after GROUP BY): +## Schema Discovery ```sql -SELECT category, SUM(amount) -FROM sales.transactions -GROUP BY category -HAVING SUM(amount) > 10000; +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) ``` -## ORDER BY Clause - -Sort results by: -- **Partition key columns** - Always supported -- **Aggregation functions** - Supported via shuffle strategy +## JOINs / Subqueries / CTEs / Set Ops ```sql --- Order by partition key -SELECT * FROM logs.requests ORDER BY timestamp DESC LIMIT 100; - --- Order by aggregation (repeat function, aliases not supported) -SELECT region, SUM(amount) -FROM sales.transactions -GROUP BY region -ORDER BY SUM(amount) DESC; +-- 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; ``` -**Limitations:** Cannot order by non-partition columns. See [gotchas.md](gotchas.md#order-by-limitations) +## Window Functions -## LIMIT Clause +Use inline `OVER (...)`. See the SQL reference for the full list of supported window functions and frame syntax. ```sql -SELECT * FROM logs.requests LIMIT 100; +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; ``` -| Setting | Value | -|---------|-------| -| Min | 1 | -| Max | 10,000 | -| Default | 500 | +## Functions -**Always use LIMIT** to enable early termination optimization. +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 -| Type | SQL Literal | Example | -|------|-------------|---------| -| `integer` | Unquoted number | `42`, `-10` | -| `float` | Decimal number | `3.14`, `-0.5` | -| `string` | Single quotes | `'hello'`, `'GET'` | -| `boolean` | Keyword | `true`, `false` | -| `timestamp` | RFC3339 string | `'2025-01-01T00:00:00Z'` | -| `date` | ISO 8601 date | `'2025-01-01'` | +`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/`. -### Type Safety +```sql +WHERE status = 200 AND method = 'GET' -- not '200', not GET + AND ts >= '2026-01-01T00:00:00Z' -- not '2026-01-01' +``` -- Quote strings with single quotes: `'value'` -- Timestamps must be RFC3339: `'2025-01-01T00:00:00Z'` (include timezone) -- Dates must be ISO 8601: `'2025-01-01'` (YYYY-MM-DD) -- No implicit conversions +## Complex Types (quick examples; full ref in docs) ```sql --- ✅ Correct -WHERE status = 200 AND method = 'GET' AND timestamp > '2025-01-01T00:00:00Z' - --- ❌ Wrong -WHERE status = '200' -- string instead of integer -WHERE timestamp > '2025-01-01' -- missing time/timezone -WHERE method = GET -- unquoted string +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 ``` -## Query Result Format - -JSON array of objects: +## Errors -```json -[ - {"user_id": "user_123", "timestamp": "2025-01-15T10:30:00Z", "status": 200}, - {"user_id": "user_456", "timestamp": "2025-01-15T10:31:00Z", "status": 404} -] -``` +Failed queries return `{"success": false, "errors": [{"code": ..., "message": ...}]}`. For error codes and troubleshooting, see `https://developers.cloudflare.com/r2-sql/troubleshooting/`. ## See Also -- [patterns.md](patterns.md) - Query examples and use cases -- [gotchas.md](gotchas.md) - SQL limitations and error handling -- [configuration.md](configuration.md) - Setup and authentication +- [patterns.md](patterns.md) — query examples · [gotchas.md](gotchas.md) — limits & workarounds · [configuration.md](configuration.md) diff --git a/skills/cloudflare/references/r2-sql/configuration.md b/skills/cloudflare/references/r2-sql/configuration.md index 3c5cfb2..4dbc049 100644 --- a/skills/cloudflare/references/r2-sql/configuration.md +++ b/skills/cloudflare/references/r2-sql/configuration.md @@ -1,147 +1,50 @@ # R2 SQL Configuration -Setup and configuration for R2 SQL queries. +Auth and setup. For the current permission matrix and wrangler flags, pull `https://developers.cloudflare.com/r2-sql/reference/wrangler-commands/` and the R2 Data Catalog manage-catalogs doc. ## Prerequisites -- R2 bucket with Data Catalog enabled -- API token with R2 permissions -- Wrangler CLI installed (for CLI queries) +- R2 bucket with Data Catalog enabled ([r2-data-catalog/configuration.md](../r2-data-catalog/configuration.md)) +- R2 API token: **R2 Storage Admin Read & Write** (includes R2 SQL Read), or add **R2 SQL Read** explicitly +- Wrangler CLI (for CLI queries) -## Enable R2 Data Catalog +> Open-beta limitation: R2 Storage **Admin Read & Write is required even for read-only R2 SQL queries**. -R2 SQL queries Apache Iceberg tables in R2 Data Catalog. Must enable catalog on bucket first. - -### Via Wrangler CLI +## Enable Catalog + Get Warehouse ```bash -npx wrangler r2 bucket catalog enable -``` - -Output includes: -- **Warehouse name** - Typically same as bucket name -- **Catalog URI** - REST endpoint for catalog operations - -Example output: -``` -Catalog enabled successfully -Warehouse: my-bucket -Catalog URI: https://abc123.r2.cloudflarestorage.com/iceberg/my-bucket +npx wrangler r2 bucket catalog enable my-bucket ``` -### Via Dashboard - -1. Navigate to **R2 Object Storage** → Select your bucket -2. Click **Settings** tab -3. Scroll to **R2 Data Catalog** section -4. Click **Enable** -5. Note the **Catalog URI** and **Warehouse** name - -**Important:** Enabling catalog creates metadata directories in bucket but does not modify existing objects. - -## Create API Token - -R2 SQL requires API token with R2 permissions. +You query by **warehouse** name (`{ACCOUNT_ID}_{BUCKET}`), shown in the output alongside the Catalog URI. -### Required Permission - -**R2 Admin Read & Write** (includes R2 SQL Read permission) - -### Via Dashboard - -1. Navigate to **R2 Object Storage** -2. Click **Manage API tokens** (top right) -3. Click **Create API token** -4. Select **Admin Read & Write** permission -5. Click **Create API Token** -6. **Copy token value** - shown only once - -### Permission Scope - -| Permission | Grants Access To | -|------------|------------------| -| R2 Admin Read & Write | R2 storage operations + R2 SQL queries + Data Catalog operations | -| R2 SQL Read | SQL queries only (no storage writes) | - -**Note:** R2 SQL Read permission not yet available via Dashboard - use Admin Read & Write. - -## Configure Environment +## Configure Auth ### Wrangler CLI -Set environment variable for Wrangler to use: - ```bash export WRANGLER_R2_SQL_AUTH_TOKEN= +# or a .env file in the project dir (auto-loaded): WRANGLER_R2_SQL_AUTH_TOKEN= ``` -Or create `.env` file in project directory: - -``` -WRANGLER_R2_SQL_AUTH_TOKEN= -``` - -Wrangler automatically loads `.env` file when running commands. +> Wrangler does **not** use the `wrangler login` OAuth session for R2 SQL — the env var is required. -### HTTP API - -For programmatic access (non-Wrangler), pass token in Authorization header: +### REST API ```bash -curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/r2/sql/query \ - -H "Authorization: Bearer " \ - -H "Content-Type: application/json" \ - -d '{ - "warehouse": "my-bucket", - "query": "SELECT * FROM default.my_table LIMIT 10" - }' +curl -X POST \ + "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/r2-sql/query/$BUCKET" \ + -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \ + -d '{"query": "SELECT * FROM default.my_table LIMIT 10"}' ``` -**Note:** HTTP API endpoint URL may vary - see [patterns.md](patterns.md#http-api-query) for current endpoint. - ## Verify Setup -Test configuration by querying system tables: - ```bash -# List namespaces -npx wrangler r2 sql query "my-bucket" "SHOW DATABASES" - -# List tables in namespace -npx wrangler r2 sql query "my-bucket" "SHOW TABLES IN default" +npx wrangler r2 sql query "${ACCOUNT_ID}_my-bucket" "SHOW DATABASES" +npx wrangler r2 sql query "${ACCOUNT_ID}_my-bucket" "SHOW TABLES IN default" ``` -If successful, returns JSON array of results. - -## Troubleshooting - -### "Token authentication failed" - -**Cause:** Invalid or missing token - -**Solution:** -- Verify `WRANGLER_R2_SQL_AUTH_TOKEN` environment variable set -- Check token has Admin Read & Write permission -- Create new token if expired - -### "Catalog not enabled on bucket" - -**Cause:** Data Catalog not enabled - -**Solution:** -- Run `npx wrangler r2 bucket catalog enable ` -- Or enable via Dashboard (R2 → bucket → Settings → R2 Data Catalog) - -### "Permission denied" - -**Cause:** Token lacks required permissions - -**Solution:** -- Verify token has **Admin Read & Write** permission -- Create new token with correct permissions - ## See Also -- [r2-data-catalog/configuration.md](../r2-data-catalog/configuration.md) - Detailed token setup and PyIceberg connection -- [patterns.md](patterns.md) - Query examples using configuration -- [gotchas.md](gotchas.md) - Common configuration errors +- [api.md](api.md) — SQL syntax · [patterns.md](patterns.md) — query examples · [gotchas.md](gotchas.md) — troubleshooting diff --git a/skills/cloudflare/references/r2-sql/gotchas.md b/skills/cloudflare/references/r2-sql/gotchas.md index d16de94..b75824d 100644 --- a/skills/cloudflare/references/r2-sql/gotchas.md +++ b/skills/cloudflare/references/r2-sql/gotchas.md @@ -1,212 +1,39 @@ # R2 SQL Gotchas -Limitations, troubleshooting, and common pitfalls for R2 SQL. +Operational pitfalls. For the authoritative list of supported features, unsupported features, and recommended workarounds, pull `https://developers.cloudflare.com/r2-sql/reference/limitations-best-practices/` and `https://developers.cloudflare.com/r2-sql/troubleshooting/`. -## Critical Limitations +## Access -### No Workers Binding +- **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_R2_SQL_AUTH_TOKEN` — it does **not** reuse the `wrangler login` OAuth session. +- Open beta: R2 Storage **Admin Read & Write is required even for read-only** queries. -**Cannot call R2 SQL from Workers/Pages code** - no binding exists. +## Type Safety -```typescript -// ❌ This doesn't exist -export default { - async fetch(request, env) { - const result = await env.R2_SQL.query("SELECT * FROM table"); // Not possible - return Response.json(result); - } -}; -``` - -**Solutions:** -- HTTP API from external systems (not Workers) -- PyIceberg/Spark via r2-data-catalog REST API -- For Workers, use D1 or external databases - -### ORDER BY Limitations - -Can only order by: -1. **Partition key columns** - Always supported -2. **Aggregation functions** - Supported via shuffle strategy - -**Cannot order by** regular non-partition columns. - -```sql --- ✅ Valid: ORDER BY partition key -SELECT * FROM logs.requests ORDER BY timestamp DESC LIMIT 100; - --- ✅ Valid: ORDER BY aggregation -SELECT region, SUM(amount) FROM sales.transactions -GROUP BY region ORDER BY SUM(amount) DESC; - --- ❌ Invalid: ORDER BY non-partition column -SELECT * FROM logs.requests ORDER BY user_id; - --- ❌ Invalid: ORDER BY alias (must repeat function) -SELECT region, SUM(amount) as total FROM sales.transactions -GROUP BY region ORDER BY total; -- Use ORDER BY SUM(amount) -``` - -Check partition spec: `DESCRIBE namespace.table_name` - -## SQL Feature Limitations - -| Feature | Supported | Notes | -|---------|-----------|-------| -| SELECT, WHERE, GROUP BY, HAVING | ✅ | Standard support | -| COUNT, SUM, AVG, MIN, MAX | ✅ | Standard aggregations | -| ORDER BY partition/aggregation | ✅ | See above | -| LIMIT | ✅ | Max 10,000 | -| Column aliases | ❌ | No AS alias | -| Expressions in SELECT | ❌ | No col1 + col2 | -| ORDER BY non-partition | ❌ | Fails at runtime | -| JOINs, subqueries, CTEs | ❌ | Denormalize at write time | -| Window functions, UNION | ❌ | Use external engines | -| INSERT/UPDATE/DELETE | ❌ | Use PyIceberg/Pipelines | -| Nested columns, arrays, JSON | ❌ | Flatten at write time | - -**Workarounds:** -- No JOINs: Denormalize data or use Spark/PyIceberg -- No subqueries: Split into multiple queries -- No aliases: Accept generated names, transform in app - -## Common Errors - -### "Column not found" -**Cause:** Typo, column doesn't exist, or case mismatch -**Solution:** `DESCRIBE namespace.table_name` to check schema - -### "Type mismatch" -```sql --- ❌ Wrong types -WHERE status = '200' -- string instead of integer -WHERE timestamp > '2025-01-01' -- missing time/timezone - --- ✅ Correct types -WHERE status = 200 -WHERE timestamp > '2025-01-01T00:00:00Z' -``` - -### "ORDER BY column not in partition key" -**Cause:** Ordering by non-partition column -**Solution:** Use partition key, aggregation, or remove ORDER BY. Check: `DESCRIBE table` - -### "Token authentication failed" -```bash -# Check/set token -echo $WRANGLER_R2_SQL_AUTH_TOKEN -export WRANGLER_R2_SQL_AUTH_TOKEN= - -# Or .env file -echo "WRANGLER_R2_SQL_AUTH_TOKEN=" > .env -``` - -### "Table not found" ```sql --- Verify catalog and tables -SHOW DATABASES; -SHOW TABLES IN namespace_name; +-- ❌ 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' ``` -Enable catalog: `npx wrangler r2 bucket catalog enable ` - -### "LIMIT exceeds maximum" -Max LIMIT is 10,000. For pagination, use WHERE filters with partition keys. - -### "No data returned" (unexpected) -**Debug steps:** -1. `SELECT COUNT(*) FROM table` - verify data exists -2. Remove WHERE filters incrementally -3. `SELECT * FROM table LIMIT 10` - inspect actual data/types - -## Performance Issues - -### Slow Queries - -**Causes:** Too many partitions, large LIMIT, no filters, small files - -```sql --- ❌ Slow: No filters -SELECT * FROM logs.requests LIMIT 10000; - --- ✅ Fast: Filter on partition key -SELECT * FROM logs.requests -WHERE timestamp >= '2025-01-15T00:00:00Z' AND timestamp < '2025-01-16T00:00:00Z' -LIMIT 1000; - --- ✅ Faster: Multiple filters -SELECT * FROM logs.requests -WHERE timestamp >= '2025-01-15T00:00:00Z' AND status = 404 AND method = 'GET' -LIMIT 1000; -``` - -**File optimization:** -- Target Parquet size: 100-500MB compressed -- Pipelines roll interval: 300+ sec (prod), 10 sec (dev) -- Run compaction to merge small files - -### Query Timeout - -**Solution:** Add restrictive WHERE filters, reduce time range, query smaller intervals - -```sql --- ❌ Times out: Year-long aggregation -SELECT status, COUNT(*) FROM logs.requests -WHERE timestamp >= '2024-01-01T00:00:00Z' GROUP BY status; - --- ✅ Faster: Month-long aggregation -SELECT status, COUNT(*) FROM logs.requests -WHERE timestamp >= '2025-01-01T00:00:00Z' AND timestamp < '2025-02-01T00:00:00Z' -GROUP BY status; -``` - -## Best Practices - -### Partitioning -- **Time-series:** Partition by day/hour on timestamp -- **Avoid:** High-cardinality keys (user_id), >10,000 partitions - -```python -from pyiceberg.partitioning import PartitionSpec, PartitionField -from pyiceberg.transforms import DayTransform - -PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day")) -``` - -### Query Writing -- **Always use LIMIT** for early termination -- **Filter on partition keys first** for pruning -- **Combine filters with AND** for more pruning - -```sql --- Good -WHERE timestamp >= '2025-01-15T00:00:00Z' AND status = 404 AND method = 'GET' LIMIT 100 -``` +No implicit conversions. Timestamps must be RFC3339 with timezone; dates ISO 8601. -### Type Safety -- Quote strings: `'GET'` not `GET` -- RFC3339 timestamps: `'2025-01-01T00:00:00Z'` not `'2025-01-01'` -- ISO dates: `'2025-01-15'` not `'01/15/2025'` +## Performance -### Data Organization -- **Pipelines:** Dev `roll_file_time: 10`, Prod `roll_file_time: 300+` -- **Compression:** Use `zstd` -- **Maintenance:** Compaction for small files, expire old snapshots +- **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. -## Debugging Checklist +## Debug Checklist -1. `npx wrangler r2 bucket catalog enable ` - Verify catalog -2. `echo $WRANGLER_R2_SQL_AUTH_TOKEN` - Check token -3. `SHOW DATABASES` - List namespaces -4. `SHOW TABLES IN namespace` - List tables -5. `DESCRIBE namespace.table` - Check schema -6. `SELECT COUNT(*) FROM namespace.table` - Verify data -7. `SELECT * FROM namespace.table LIMIT 10` - Test simple query -8. Add filters incrementally +1. `wrangler r2 bucket catalog enable ` — catalog on? +2. `echo $WRANGLER_R2_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) - SQL syntax -- [patterns.md](patterns.md) - Query optimization -- [configuration.md](configuration.md) - Setup -- [Cloudflare R2 SQL Docs](https://developers.cloudflare.com/r2-sql/) +- [api.md](api.md) · [patterns.md](patterns.md) · [configuration.md](configuration.md) diff --git a/skills/cloudflare/references/r2-sql/patterns.md b/skills/cloudflare/references/r2-sql/patterns.md index 53de7d3..0359e9a 100644 --- a/skills/cloudflare/references/r2-sql/patterns.md +++ b/skills/cloudflare/references/r2-sql/patterns.md @@ -1,222 +1,118 @@ # R2 SQL Patterns -Common patterns, use cases, and integration examples for R2 SQL. +Code templates for CLI, REST, and Worker access. For performance/partitioning best practices, pull `https://developers.cloudflare.com/r2-sql/reference/limitations-best-practices/`. -## Wrangler CLI Query +## Wrangler CLI ```bash -# Basic query -npx wrangler r2 sql query "my-bucket" "SELECT * FROM default.logs LIMIT 10" - -# Multi-line query -npx wrangler r2 sql query "my-bucket" " - SELECT status, COUNT(*), AVG(response_time) - FROM logs.http_requests - WHERE timestamp >= '2025-01-01T00:00:00Z' - GROUP BY status - ORDER BY COUNT(*) DESC - LIMIT 100 -" - -# Use environment variable -export R2_SQL_WAREHOUSE="my-bucket" -npx wrangler r2 sql query "$R2_SQL_WAREHOUSE" "SELECT * FROM default.logs" +export WRANGLER_R2_SQL_AUTH_TOKEN=$API_TOKEN + +npx wrangler r2 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" ``` -## HTTP API Query +## REST API (Python) -For programmatic access from external systems (not Workers - see gotchas.md). +```python +import requests -```bash -curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/r2/sql/query \ - -H "Authorization: Bearer " \ - -H "Content-Type: application/json" \ - -d '{ - "warehouse": "my-bucket", - "query": "SELECT * FROM default.my_table WHERE status = 200 LIMIT 100" - }' -``` +API = f"https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/r2-sql/query/{BUCKET}" +HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"} -Response: -```json -{ - "success": true, - "result": [{"user_id": "user_123", "timestamp": "2025-01-15T10:30:00Z", "status": 200}], - "errors": [] -} -``` +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"]) -## Pipelines Integration +rows, metrics = r2sql("SELECT category, COUNT(*) AS cnt FROM analytics.events GROUP BY category LIMIT 10") +``` -Stream data to Iceberg tables via Pipelines, then query with R2 SQL. +## REST API (curl) ```bash -# Setup pipeline (select Data Catalog Table destination) -npx wrangler pipelines setup - -# Key settings: -# - Destination: Data Catalog Table -# - Compression: zstd (recommended) -# - Roll file time: 300+ sec (production), 10 sec (dev) - -# Send data to pipeline -curl -X POST https://{stream-id}.ingest.cloudflare.com \ - -H "Content-Type: application/json" \ - -d '[{"user_id": "user_123", "event_type": "purchase", "timestamp": "2025-01-15T10:30:00Z", "amount": 29.99}]' - -# Query ingested data (wait for roll interval) -npx wrangler r2 sql query "my-bucket" " - SELECT event_type, COUNT(*), SUM(amount) - FROM default.events - WHERE timestamp >= '2025-01-15T00:00:00Z' - GROUP BY event_type -" +curl -X POST \ + "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/r2-sql/query/$BUCKET" \ + -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \ + -d '{"query": "SELECT COUNT(*) AS total FROM analytics.events"}' ``` -See [pipelines/patterns.md](../pipelines/patterns.md) for detailed setup. +## Dashboard Worker -## PyIceberg Integration +No R2 SQL binding exists — query the REST endpoint via `fetch()`. -Create and populate Iceberg tables with PyIceberg, then query with R2 SQL. +```typescript +interface Env { ACCOUNT_ID: string; BUCKET: string; R2_SQL_TOKEN: string; } -```python -from pyiceberg.catalog.rest import RestCatalog -import pyarrow as pa -import pandas as pd - -# Setup catalog -catalog = RestCatalog( - name="my_catalog", - warehouse="my-bucket", - uri="https://.r2.cloudflarestorage.com/iceberg/my-bucket", - token="", -) -catalog.create_namespace_if_not_exists("analytics") - -# Create table -schema = pa.schema([ - pa.field("user_id", pa.string(), nullable=False), - pa.field("event_time", pa.timestamp("us", tz="UTC"), nullable=False), - pa.field("page_views", pa.int64(), nullable=False), -]) -table = catalog.create_table(("analytics", "user_metrics"), schema=schema) - -# Append data -df = pd.DataFrame({ - "user_id": ["user_1", "user_2"], - "event_time": pd.to_datetime(["2025-01-15 10:00:00", "2025-01-15 11:00:00"], utc=True), - "page_views": [10, 25], -}) -table.append(pa.Table.from_pandas(df, schema=schema)) +async function queryR2SQL(env: Env, query: string) { + const url = `https://api.sql.cloudflarestorage.com/api/v1/accounts/${env.ACCOUNT_ID}/r2-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(`R2 SQL ${resp.status}: ${await resp.text()}`); + return (await resp.json() as any).result; +} + +export default { + async fetch(req: Request, env: Env): Promise { + 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 }); + }, +}; ``` -Query with R2 SQL: ```bash -npx wrangler r2 sql query "my-bucket" " - SELECT user_id, SUM(page_views) - FROM analytics.user_metrics - WHERE event_time >= '2025-01-15T00:00:00Z' - GROUP BY user_id -" +npx wrangler secret put R2_SQL_TOKEN ``` -See [r2-data-catalog/patterns.md](../r2-data-catalog/patterns.md) for advanced PyIceberg patterns. - -## Use Cases +## Example Queries -### Log Analytics ```sql -- Error rate by endpoint -SELECT path, COUNT(*), SUM(CASE WHEN status >= 400 THEN 1 ELSE 0 END) as errors -FROM logs.http_requests -WHERE timestamp BETWEEN '2025-01-01T00:00:00Z' AND '2025-01-31T23:59:59Z' +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; --- Response time stats -SELECT method, MIN(response_time_ms), AVG(response_time_ms), MAX(response_time_ms) -FROM logs.http_requests WHERE timestamp >= '2025-01-15T00:00:00Z' GROUP BY method; +-- 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; --- Traffic by status -SELECT status, COUNT(*) FROM logs.http_requests -WHERE timestamp >= '2025-01-15T00:00:00Z' AND method = 'GET' -GROUP BY status ORDER BY COUNT(*) DESC; +-- 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; ``` -### Fraud Detection -```sql --- High-value transactions -SELECT location, COUNT(*), SUM(amount), AVG(amount) -FROM fraud.transactions WHERE transaction_timestamp >= '2025-01-01T00:00:00Z' AND amount > 1000.0 -GROUP BY location ORDER BY SUM(amount) DESC LIMIT 20; - --- Flagged transactions -SELECT merchant_category, COUNT(*), AVG(amount) FROM fraud.transactions -WHERE is_fraud_flag = true AND transaction_timestamp >= '2025-01-01T00:00:00Z' -GROUP BY merchant_category HAVING COUNT(*) > 10 ORDER BY COUNT(*) DESC; -``` +## Cursor-Based Pagination -### Business Intelligence -```sql --- Sales by department -SELECT department, SUM(revenue), AVG(revenue), COUNT(*) FROM sales.transactions -WHERE sale_date >= '2024-01-01' GROUP BY department ORDER BY SUM(revenue) DESC LIMIT 10; - --- Product performance -SELECT category, COUNT(DISTINCT product_id), SUM(units_sold), SUM(revenue) -FROM sales.product_sales WHERE sale_date BETWEEN '2024-10-01' AND '2024-12-31' -GROUP BY category ORDER BY SUM(revenue) DESC; -``` - -## Connecting External Engines - -R2 Data Catalog exposes Iceberg REST API. Connect Spark, Snowflake, Trino, DuckDB, etc. - -```scala -// Apache Spark example -val spark = SparkSession.builder() - .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") - .config("spark.sql.catalog.my_catalog.catalog-impl", "org.apache.iceberg.rest.RESTCatalog") - .config("spark.sql.catalog.my_catalog.uri", "https://.r2.cloudflarestorage.com/iceberg/my-bucket") - .config("spark.sql.catalog.my_catalog.token", "") - .getOrCreate() +Paginate on a sortable (ideally partition) column rather than `OFFSET`: -spark.sql("SELECT * FROM my_catalog.default.my_table LIMIT 10").show() +```sql +SELECT * FROM logs.requests ORDER BY __ingest_ts DESC LIMIT 500; -- page 1 +SELECT * FROM logs.requests WHERE __ingest_ts < '' ORDER BY __ingest_ts DESC LIMIT 500; -- page 2 ``` -See [r2-data-catalog/patterns.md](../r2-data-catalog/patterns.md) for more engines. - -## Performance Optimization +## Performance (essentials) -### Partitioning -- **Time-series:** day/hour on timestamp -- **Geographic:** region/country -- **Avoid:** High-cardinality keys (user_id) +- **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 [r2-data-catalog](../r2-data-catalog/configuration.md)). +- Read response `metrics` (`files_scanned`, `bytes_scanned`) to tune. Full guidance: limitations-best-practices doc. -```python -from pyiceberg.partitioning import PartitionSpec, PartitionField -from pyiceberg.transforms import DayTransform - -PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day")) -``` - -### Query Optimization -- **Always use LIMIT** for early termination -- **Filter on partition keys first** -- **Multiple filters** for better pruning - -```sql --- Better: Multiple filters on partition key -SELECT * FROM logs.requests -WHERE timestamp >= '2025-01-15T00:00:00Z' AND status = 404 AND method = 'GET' LIMIT 100; -``` +## Pipelines → R2 SQL -### File Organization -- **Pipelines roll:** Dev 10-30s, Prod 300+s -- **Target Parquet:** 100-500MB compressed +After `npx wrangler pipelines setup` (Data 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) - SQL syntax reference -- [gotchas.md](gotchas.md) - Limitations and troubleshooting -- [r2-data-catalog/patterns.md](../r2-data-catalog/patterns.md) - PyIceberg advanced patterns -- [pipelines/patterns.md](../pipelines/patterns.md) - Streaming ingestion patterns +- [api.md](api.md) · [gotchas.md](gotchas.md) · [r2-data-catalog/patterns.md](../r2-data-catalog/patterns.md)