sql-federated-queries

# Redpanda SQL: Federated & External Queries Oxla can query external data — Kafka/Redpanda topics, Apache Iceberg tables, and object-store files — without ETL, joining them directly with native columnar tables using standard SQL. External sources are accessed through named catalog objects that Oxla stores in its own metastore. Schemas are decoded automatically from Avro, Protobuf, or JSON via the Schema Registry and can evolve transparently. Connect to Oxla via the PostgreSQL wire protocol. The port is configurable via `network.postgresql.port` (conventional default: 5432): ```bash psql -h <oxla-host> -p <port> -U <user> ``` --- ## Quickstart ### 1. Kafka Catalog — query a Redpanda/Kafka topic ```sql -- Step 1: Create the catalog (connection to the broker + schema registry). -- Required options: initial_brokers, schema_registry_url CREATE KAFKA CATALOG my_kafka WITH ( initial_brokers = 'localhost:9092', schema_registry_url = 'http://localhost:8081' ); -- Step 2: Register the topic as a table inside the catalog. -- Required option: topic CREATE TABLE my_kafka=>orders WITH (topic = 'orders-topic'); -- Step 3: Refresh the schema (pulls Avro/Protobuf/JSON schema from registry). REFRESH my_kafka=>orders; -- Step 4: Query it like any table. SELECT order_id, customer_id, amount FROM my_kafka=>orders WHERE amount > 100; -- Filter on Kafka metadata (partition, offset, timestamp). SELECT order_id, (redpanda).partition, (redpanda)."offset" FROM my_kafka=>orders WHERE (redpanda).timestamp > TIMESTAMP '2024-06-01 00:00:00'; ``` ### 2. Iceberg Catalog — query Apache Iceberg tables ```sql -- Step 0: Create a storage connection (required before any ICEBERG CATALOG). CREATE STORAGE my_s3 TYPE = S3 WITH (region = 'us-east-1', access_key_id = '...', secret_access_key = '...'); -- Step 1: Create an Iceberg REST catalog (STORAGE clause is mandatory). CREATE ICEBERG CATALOG my_ice STORAGE public.my_s3 WITH ( uri = 'https://iceberg.example.com', auth_type = 'oauth2', oauth2_client_id = 'my-client-id', oauth2_client_secret = 'my-client-secret' ); -- Step 2: Query an Iceberg table using the catalog=>namespace.table syntax. SELECT customer_id, total FROM my_ice=>sales.orders WHERE order_date >= DATE '2024-01-01'; -- Step 3: Inspect what tables are known. SELECT * FROM system.iceberg_tables WHERE catalog_name = 'my_ice'; ``` ### 3. File-based — COPY FROM parquet on S3 ```sql -- Load a parquet file from S3 into a native table. COPY my_table FROM 's3://my-bucket/data/file.parquet' (FORMAT parquet); -- Export a native table to parquet on S3. COPY my_table TO 's3://my-bucket/exports/file.parquet' (FORMAT parquet); -- Load ORC. COPY my_table FROM 's3://my-bucket/data/file.orc' (FORMAT ORC); ``` --- ## Kafka Catalogs Kafka catalogs connect Oxla to a Kafka or Redpanda cluster. Both `CREATE KAFKA CATALOG` and `CREATE REDPANDA CATALOG` are accepted (synonyms). ### Connection options (`CREATE KAFKA CATALOG ... WITH (...)`) | Option | Required | Description | |--------|----------|-------------| | `initial_brokers` | Yes | Bootstrap broker list (e.g. `'broker1:9092,broker2:9092'`) | | `schema_registry_url` | Yes | Schema Registry URL (`http://` or `https://`) | | `pandaproxy_url` | No | Panda Proxy URL for transparent Iceberg queries | | `sasl_mechanism` | No | `'SCRAM-SHA-256'` or `'SCRAM-SHA-512'` | | `sasl_user` | No | SASL username | | `sasl_password` | No | SASL password (stored encrypted) | | `truststore` | No | PEM CA bundle for TLS verification | | `key_store_key` | No | PEM private key for mTLS | | `key_store_cert` | No | PEM certificate for mTLS | | `connection_timeout` | No | Timeout in milliseconds (integer) | | `rd_kafka_debug` | No | librdkafka debug contexts (e.g. `'all'`) | ### Table options (`CREATE TABLE catalog=>name WITH (...)`) | Option | Required | Default | Description | |--------|----------|---------|-------------| | `topic` | Yes | — | Kafka topic name | | `schema_subject` | No | `<topic>-value` | Schema Registry subject name | | `schema_lookup_policy` | No | `LATEST` | `LATEST` (pin to latest schema) or `SCHEMA_ID` (per-record schema ID) | | `error_handling_policy` | No | `FAIL` | `FAIL`, `FILL_NULL`, or `DROP_RECORD` on decode errors | | `struct_mapping_policy` | No | `COMPOUND` | `COMPOUND` (nested struct, and map fields as map columns) or `JSON` (collapse nested records and maps to SQL JSON) | | `output_schema_message_full_name` | No | — | Protobuf message full name (e.g. `com.example.Order`) | | `confluent_wire_protocol` | No | `true` | `true` or `false`; only for `LATEST` policy | | `key_decode_mode` | No | `binary` | Decode the record key: `binary`, `string`, `schema_latest`, or `schema_id_prefix` | | `key_schema_subject` | No | — | Schema Registry subject for the key schema (`schema_*` key modes) | | `key_confluent_wire_protocol` | No | `false` | `true`/`false`; only when `key_decode_mode = 'schema_latest'` | | `key_schema_message_full_name` | No | — | Protobuf key message full name (`schema_*` key modes) | | `header_value_type` | No | `binary` | Decode header values as `binary` (`BYTEA`) or `string` (`TEXT`) | `key_decode_mode` and `header_value_type` are fixed once set — `ALTER` rejects changing them (see `references/kafka-catalogs.md`). ### Querying Use the `catalog=>table` syntax in any SELECT: ```sql -- Simple select SELECT * FROM my_kafka=>orders; -- Join with a native table SELECT o.order_id, c.name, o.amount FROM my_kafka=>orders o JOIN customers c ON o.customer_id = c.id; -- Access Kafka metadata via the `redpanda` struct column SELECT (redpanda).partition, (redpanda)."offset", (redpanda).timestamp, order_id FROM my_kafka=>orders; ``` ### Altering and refreshing ```sql -- Change table options after creation ALTER TABLE my_kafka=>orders WITH (error_handling_policy = 'FILL_NULL'); -- Change catalog-level options ALTER KAFKA CATALOG my_kafka WITH (connection_timeout = 30000); -- Relink a Kafka catalog to a different Iceberg catalog (for transparent queries) ALTER KAFKA CATALOG my_kafka USING CATALOG my_ice; -- Detach the Iceberg link ALTER KAFKA CATALOG my_kafka USING CATALOG NULL; -- Refresh schema from Schema Registry REFRESH my_kafka=>orders; ``` --- ## Iceberg Catalogs Oxla connects to Apache Iceberg REST catalogs (e.g., Polaris, AWS Glue, Tabular). ### Connection options (`CREATE ICEBERG CATALOG ... WITH (...)`) | Option | Required | Description | |--------|----------|-------------| | `uri` | Yes | REST catalog endpoint URL | | `warehouse` | No | Warehouse path or identifier | | `auth_type` | No | `'oauth2'`, `'basic'`, or `'aws_sigv4'` | | `oauth2_client_id` | OAuth2 | OAuth2 client ID | | `oauth2_client_secret` | OAuth2 | OAuth2 client secret (stored encrypted) | | `oauth2_scope` | No | OAuth2 scope (default: `PRINCIPAL_ROLE:ALL`) | | `oauth2_token_endpoint_url` | No | Token endpoint URL (auto-discovered from /config if omitted) | | `oauth2_token_refresh_margin_seconds` | No | Seconds before expiry to refresh token | | `username` | Basic | HTTP basic auth username | | `password` | Basic | HTTP basic auth password (stored encrypted) | | `aws_region` | SigV4 | AWS region (e.g., `'us-east-1'`) | | `aws_access_key_id` | SigV4 | AWS access key (omit to use instance IAM role) | | `aws_secret_access_key` | SigV4 | AWS secret key | | `aws_service_name` | No | SigV4 service name (default: `'glue'`) | | `ssl_verify` | No | TLS verification: `'true'` or `'false'` | | `ssl_ca_info` | No | Path to CA bundle | | `ssl_ca_path` | No | Path to directory of CA certs | | `ssl_crl_file` | No | Path to CRL file | ### Querying ```sql -- Single-level namespace SELECT id, name FROM my_ice=>ns.table1; -- Multi-level namespace SELECT age, name FROM my_ice=>my.path.table1; -- Partition filter (pushes predicate to Iceberg scan plan) SELECT id FROM my_ice=>my.path.partitioned_table WHERE id = 1; -- Range filter on a partition column SELECT id, name FROM my_ice=>my.path.range_table WHERE id < 3; -- Date/timestamp partition filters SELECT id FROM my_ice=>my.path.partitioned_table WHERE val_date = DATE '2024-01-02'; SELECT id FROM my_ice=>my.path.partitioned_table WHERE val_ts = TIMESTAMP '2024-01-03 12:00:00'; ``` ### Creating and dropping Iceberg tables and namespaces (DDL) Oxla can also issue DDL against the Iceberg REST catalog itself. A table path always needs a namespace (`catalog=>namespace.table`). ```sql -- Namespaces CREATE NAMESPACE my_ice=>sales; DROP NAMESPACE my_ice=>sales; -- RESTRICT (default); must be empty -- Tables (explicit column list; optional PARTITION BY and WITH properties) CREATE TABLE my_ice=>sales.orders (id INT, amount BIGINT NOT NULL, label TEXT); CREATE TABLE my_ice=>sales.events (id INT, d DATE) PARTITION BY (day(d), bucket(8, id)); DROP TABLE my_ice=>sales.orders; -- add PURGE to also delete data files ``` `NOT NULL` columns become required Iceberg fields; `NUMERIC(p,s)` maps to `decimal(p,s)`; types with no Iceberg equivalent (`INTERVAL`, `JSON`/`JSONB`, `UUID`, `INT16`/`INT32`, geo types) are rejected. `CREATE TABLE AS SELECT` and `DROP NAMESPACE ... CASCADE` parse but are not supported. Iceberg DDL is authorized against the catalog's owner. See [iceberg.md](references/iceberg.md) for the full grammar, partition transforms, and type mapping. --- ## File-Based External Data (COPY FROM/TO) Oxla supports `COPY FROM` and `COPY TO` with parquet and ORC on S3, GCS, and Azure. ```sql -- Load parquet COPY my_table FROM 's3://bucket/path/file.parquet' (FORMAT parquet); -- Load ORC COPY my_table FROM 's3://bucket/path/file.orc' (FORMAT ORC); -- Export parquet COPY my_table TO 's3://bucket/path/out.parquet' (FORMAT parquet); -- Export ORC COPY my_table TO 's3://bucket/path/out.orc' (FORMAT ORC); -- Load from STDIN (CSV) COPY my_table FROM STDIN (FORMAT CSV); ``` Storage credentials can be configured either via `CREATE STORAGE` (persistent named connection; see the sql-admin-api skill) or inline per-statement using `AWS_CRED(...)`, `GCS_CRED(...)`, or `AZURE_CRED(...)` options on the `COPY` statement (see [files-and-system-tables.md](references/files-and-system-tables.md)). Path protocols: `s3://`, `gs://`, `az://` or `wasbs://` (Azure), `local://`. --- ## System Tables for External Metadata These virtual tables live in the `system` schema and are accessible in every database. ```sql -- Kafka catalog connections SELECT * FROM system.kafka_connections; -- Columns: database_name, namespace_name, name, options, iceberg_catalog (nullable) -- Kafka topic sources SELECT * FROM system.kafka_sources; -- Columns: database_name, namespace_name, name, connection_name, topic_name, -- subject_name, lookup_policy, error_handling_policy, struct_mapping_policy, -- output_schema_full_message_name -- Iceberg catalogs SELECT * FROM system.iceberg_catalogs; -- Columns: name, uri, warehouse, auth_type, namespace_name, database_name -- Iceberg tables (REFRESHed into local catalog) SELECT * FROM system.iceberg_tables; -- Columns: database_name, namespace_name, catalog_name, name, oid