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):

psql -h <oxla-host> -p <port> -U <user>

Quickstart

1. Kafka Catalog — query a Redpanda/Kafka topic

-- 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

-- 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

-- 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 (...))

Table options (CREATE TABLE catalog=>name WITH (...))

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:

-- 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

-- 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 (...))

Querying

-- 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).

-- 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 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.

-- 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). 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.

-- 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

-- Examples
SELECT name, topic_name, lookup_policy
FROM   system.kafka_sources
WHERE  connection_name = 'my_kafka';

SELECT name, uri, auth_type
FROM   system.iceberg_catalogs;

Redpanda Enterprise Source Config (producing the data Oxla reads)

The Iceberg tables, Tiered Storage segments, and validated schemas that Oxla queries are produced on the Redpanda side by Enterprise Edition features (valid license required). When Oxla reads an Iceberg table or runs a transparent Kafka-Iceberg query, the Redpanda cluster must first be configured to write it.

# Redpanda side (Enterprise): produce an Iceberg table Oxla can read.
rpk cluster config set cloud_storage_enabled true        # Tiered Storage (prereq)
rpk cluster config set iceberg_enabled true              # Iceberg Topics
rpk cluster config set iceberg_catalog_type rest
rpk cluster config set iceberg_rest_catalog_endpoint http://catalog:8181
rpk topic alter-config clicks --set redpanda.iceberg.mode=value_schema_id_prefix

The redpanda.iceberg.* topic properties (mode, delete, invalid.record.action, partition.spec, target.lag.ms), the iceberg_rest_catalog_* cluster properties, and the schema-validation knobs are documented in redpanda-iceberg-source-config.md. Point Oxla's CREATE ICEBERG CATALOG uri / auth_type at the same REST catalog Redpanda writes to (iceberg_rest_catalog_endpoint / iceberg_rest_catalog_authentication_mode).

Reference Directory

• kafka-catalogs.md: Complete reference for CREATE KAFKA CATALOG, CREATE TABLE catalog=>name, ALTER KAFKA CATALOG/TABLE, REFRESH, all options, schema decoding (Avro/Protobuf/JSON) including map columns and their lookup semantics, the redpanda metadata struct with metadata-predicate read pruning, and transparent Kafka-Iceberg queries.
• iceberg.md: Querying Apache Iceberg REST catalogs — CREATE ICEBERG CATALOG, the catalog=>path.table syntax, partition pruning and range/date/timestamp filters, all auth modes (OAuth2/Basic/SigV4), and multi-file partition scans. Also covers reading map columns, concurrent-commit conflicts (SQLSTATE 40001), and catalog DDL: CREATE/DROP TABLE (with PARTITION BY transforms, PURGE, and the SQL→Iceberg type mapping) and CREATE/DROP NAMESPACE.
• files-and-system-tables.md: COPY FROM/TO with parquet and ORC on S3/GCS/Azure, storage path protocols (s3://, gs://, az:///wasbs://, local://), CREATE STORAGE credential configuration, inline per-statement AWS_CRED/GCS_CRED/AZURE_CRED credentials, and the four external-metadata system tables with their column schemas and example queries.
• redpanda-iceberg-source-config.md: The Redpanda Enterprise features that produce the data Oxla reads — Iceberg Topics (iceberg_enabled, the redpanda.iceberg.mode/delete/invalid.record.action/partition.spec/target.lag.ms topic properties, the four Iceberg modes, DLQ tables), the REST and object_storage catalog backing (iceberg_catalog_type, all iceberg_rest_catalog_* properties) and how to align them with Oxla's CREATE ICEBERG CATALOG, Tiered Storage as a prerequisite (cloud_storage_enabled), and Server-Side Schema ID Validation (enable_schema_id_validation, redpanda.key/value.schema.id.validation, subject-name strategies). Notes license requirements and expiration behavior.