SQL: Auto union Iceberg and Redpanda topic

modules/ROOT/nav.adoc

@@ -538,12 +538,15 @@
 ** xref:reference:sql/index.adoc[Redpanda SQL Reference]
 *** xref:reference:sql/sql-statements/index.adoc[Statements]
 **** xref:reference:sql/sql-statements/keywords.adoc[]
+**** xref:reference:sql/sql-statements/alter-iceberg-catalog.adoc[]
 **** xref:reference:sql/sql-statements/alter-redpanda-catalog.adoc[]
 **** xref:reference:sql/sql-statements/alter-storage.adoc[]
 **** xref:reference:sql/sql-statements/alter-table.adoc[]
+**** xref:reference:sql/sql-statements/create-iceberg-catalog.adoc[]
 **** xref:reference:sql/sql-statements/create-redpanda-catalog.adoc[]
 **** xref:reference:sql/sql-statements/create-storage.adoc[]
 **** xref:reference:sql/sql-statements/create-table.adoc[]
+**** xref:reference:sql/sql-statements/drop-iceberg-catalog.adoc[]
 **** xref:reference:sql/sql-statements/drop-redpanda-catalog.adoc[]
 **** xref:reference:sql/sql-statements/drop-storage.adoc[]
 **** xref:reference:sql/sql-statements/drop-table.adoc[]

modules/reference/pages/sql/sql-statements/alter-iceberg-catalog.adoc

@@ -0,0 +1,40 @@
+= ALTER ICEBERG CATALOG
+:description: The ALTER ICEBERG CATALOG statement modifies connection properties of an existing Iceberg catalog.
+:page-topic-type: reference
+
+The `ALTER ICEBERG CATALOG` statement modifies connection properties of an existing Iceberg catalog. You must repeat the `STORAGE` clause when altering, even if the storage connection is not changing.
+
+== Syntax
+
+[source,sql]
+----
+ALTER ICEBERG CATALOG [IF EXISTS] catalog_name STORAGE storage_name
+WITH (option = 'value' [, ...]);
+----
+
+* `catalog_name`: Name of the Iceberg catalog to modify.
+* `IF EXISTS`: Optional. Prevents an error if the Iceberg catalog does not exist.
+* `storage_name`: Name of the storage connection backing the catalog.
+* `option = 'value'`: One or more connection options to update. See xref:reference:sql/sql-statements/create-iceberg-catalog.adoc[CREATE ICEBERG CATALOG] for the full list of options.
+
+== Examples
+
+Update the REST catalog URI for an existing Iceberg catalog:
+
+[source,sql]
+----
+ALTER ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+WITH (uri = 'https://new-catalog.example.com');
+----
+
+Rotate the basic-authentication credentials on an existing Iceberg catalog:
+
+[source,sql]
+----
+ALTER ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+WITH (
+  auth_type = 'basic',
+  username = '<new-username>',
+  password = '<new-password>'
+);
+----

modules/reference/pages/sql/sql-statements/create-iceberg-catalog.adoc

@@ -0,0 +1,210 @@
+= CREATE ICEBERG CATALOG
+:description: The CREATE ICEBERG CATALOG statement creates a named connection to an Iceberg REST catalog, enabling Redpanda SQL to query Iceberg-translated topic data.
+:page-topic-type: reference
+
+The `CREATE ICEBERG CATALOG` statement creates a named connection to an Iceberg REST catalog. Link the Iceberg catalog to a Redpanda catalog with `USING CATALOG` so that queries against the linked Redpanda catalog return both live and Iceberg-translated records. Standalone querying against an Iceberg catalog is not supported. See xref:sql:query-data/query-iceberg-topics.adoc[Query Iceberg topics] for the end-to-end workflow.
+
+The statement requires an existing xref:reference:sql/sql-statements/create-storage.adoc[storage connection] that holds the object-storage credentials for the Iceberg warehouse.
+
+== Syntax
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG [IF NOT EXISTS] catalog_name STORAGE storage_name
+  WITH (option = 'value' [, ...]);
+----
+
+* `catalog_name`: Name for the new Iceberg catalog.
+* `IF NOT EXISTS`: Optional. Prevents an error if an Iceberg catalog with the same name already exists.
+* `storage_name`: Name of an existing storage connection. Create it first with xref:reference:sql/sql-statements/create-storage.adoc[CREATE STORAGE].
+
+NOTE: Catalogs are created in the current schema (`public` by default). To create a catalog in a different schema, qualify the name as `schema.catalog_name`. The auto-created `default_iceberg_catalog` is in `public`.
+
+== Options
+
+[cols="<30%,<15%,<10%,<45%",options="header"]
+|===
+|Option |Type |Required |Description
+
+|`uri`
+|STRING
+|Yes
+|REST catalog endpoint URI.
+
+|`warehouse`
+|STRING
+|No
+|Iceberg warehouse identifier or location.
+
+|`auth_type`
+|STRING
+|No
+|Authentication type for the REST catalog. One of `oauth2`, `basic`, or `aws_sigv4`. If omitted, the catalog connects without authentication. Providing an auth-specific option (such as `username` or `aws_region`) without `auth_type` is rejected.
+
+|`oauth2_client_id`
+|STRING
+|Required when `auth_type = 'oauth2'`
+|OAuth2 client ID.
+
+|`oauth2_client_secret`
+|STRING
+|Required when `auth_type = 'oauth2'`
+|OAuth2 client secret.
+
+|`oauth2_scope`
+|STRING
+|No
+|OAuth2 scope to request.
+
+|`oauth2_token_endpoint_url`
+|STRING
+|No
+|OAuth2 token endpoint URL. Use to override the catalog's default token endpoint.
+
+|`oauth2_token_refresh_margin_seconds`
+|INTEGER
+|No
+|Number of seconds before token expiry to refresh. Must be between 0 and 2147483647.
+
+|`username`
+|STRING
+|Required when `auth_type = 'basic'`
+|Basic authentication username.
+
+|`password`
+|STRING
+|Required when `auth_type = 'basic'`
+|Basic authentication password.
+
+|`aws_region`
+|STRING
+|Required when `auth_type = 'aws_sigv4'`
+|AWS region for SigV4 request signing (for example, `us-west-2`).
+
+|`aws_access_key_id`
+|STRING
+|No
+|AWS access key ID for SigV4 signing. Must be set together with `aws_secret_access_key`. If both are omitted, the catalog uses the AWS default credential chain (environment variables, shared config, STS web identity, IMDSv2/ECS).
+
+|`aws_secret_access_key`
+|STRING
+|No
+|AWS secret access key for SigV4 signing. See `aws_access_key_id` for credential-chain behavior.
+
+|`ssl_verify`
+|STRING
+|No
+|`'true'` (default) or `'false'`. Whether to verify the REST catalog's TLS certificate.
+
+|`ssl_ca_info`
+|STRING
+|No
+|Path to a CA certificate file used to verify the REST catalog's TLS certificate.
+
+|`ssl_ca_path`
+|STRING
+|No
+|Path to a directory containing CA certificates.
+
+|`ssl_crl_file`
+|STRING
+|No
+|Path to a certificate revocation list (CRL) file.
+|===
+
+== Examples
+
+=== Create a basic Iceberg catalog
+
+Connect to a REST catalog without authentication. The catalog uses TLS verification by default.
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+  WITH (
+    uri = 'https://catalog.example.com',
+    warehouse = 's3://warehouse/'
+  );
+----
+
+=== Create an Iceberg catalog with OAuth2 authentication
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+  WITH (
+    uri = 'https://catalog.example.com',
+    warehouse = 's3://lakehouse-data/',
+    auth_type = 'oauth2',
+    oauth2_client_id = '<client-id>',
+    oauth2_client_secret = '<client-secret>',
+    oauth2_scope = 'PRINCIPAL_ROLE:ALL',
+    oauth2_token_endpoint_url = 'https://auth.example.com/token',
+    oauth2_token_refresh_margin_seconds = 300
+  );
+----
+
+=== Create an Iceberg catalog with basic authentication
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+  WITH (
+    uri = 'https://catalog.example.com',
+    warehouse = 's3://warehouse/',
+    auth_type = 'basic',
+    username = '<username>',
+    password = '<password>'
+  );
+----
+
+=== Create an Iceberg catalog with AWS SigV4 authentication
+
+Use for REST catalogs fronted by AWS services (such as AWS Glue).
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+  WITH (
+    uri = 'https://catalog.example.com',
+    warehouse = 's3://warehouse/',
+    auth_type = 'aws_sigv4',
+    aws_region = 'us-west-2',
+    aws_access_key_id = '<access-key-id>',
+    aws_secret_access_key = '<secret-access-key>'
+  );
+----
+
+To use the AWS default credential chain (for example, an EC2 instance-profile role), omit `aws_access_key_id` and `aws_secret_access_key`. They must be set together or omitted together.
+
+=== Create an Iceberg catalog with custom TLS settings
+
+[source,sql]
+----
+CREATE ICEBERG CATALOG lakehouse_catalog STORAGE iceberg_storage
+  WITH (
+    uri = 'https://catalog.example.com',
+    warehouse = 's3://warehouse/',
+    ssl_verify = 'true',
+    ssl_ca_info = '/etc/ssl/certs/catalog-ca.pem'
+  );
+----
+
+== Related statements
+
+[cols="<40%,<60%",options="header"]
+|===
+|Statement |Description
+
+|xref:reference:sql/sql-statements/alter-iceberg-catalog.adoc[ALTER ICEBERG CATALOG]
+|Modify connection properties of an existing Iceberg catalog.
+
+|xref:reference:sql/sql-statements/drop-iceberg-catalog.adoc[DROP ICEBERG CATALOG]
+|Remove an Iceberg catalog.
+
+|xref:reference:sql/sql-statements/create-storage.adoc[CREATE STORAGE]
+|Create the storage connection that backs the Iceberg catalog.
+
+|xref:reference:sql/sql-statements/create-redpanda-catalog.adoc[CREATE REDPANDA CATALOG]
+|Create a Redpanda catalog. Use `USING CATALOG` to link a Redpanda catalog to an Iceberg catalog so that queries return both live and Iceberg-translated records.
+|===

modules/reference/pages/sql/sql-statements/create-redpanda-catalog.adoc

@@ -9,11 +9,15 @@ The `CREATE REDPANDA CATALOG` statement creates a named connection to a Redpanda
 [source,sql]
 ----
 CREATE REDPANDA CATALOG [IF NOT EXISTS] catalog_name
-WITH (option = 'value' [, ...]);
+  [USING CATALOG [schema.]iceberg_catalog_name]
+  WITH (option = 'value' [, ...]);
 ----
 
 * `catalog_name`: Name for the new catalog connection.
 * `IF NOT EXISTS`: Optional. Prevents an error if a catalog with the same name already exists.
+* `USING CATALOG iceberg_catalog_name`: Optional. Links the Redpanda catalog to an existing Iceberg catalog so that queries against tables in this catalog can return data from both the Redpanda topic and its corresponding Iceberg table in a single result. The Iceberg catalog must already exist. You can qualify the Iceberg catalog name with a schema prefix.
+
+NOTE: Catalogs are created in the current schema (`public` by default). To create a catalog in a different schema, qualify the name as `schema.catalog_name`. The auto-created `default_redpanda_catalog` is in `public`.
 
 == Options
 
@@ -34,32 +38,37 @@ WITH (option = 'value' [, ...]);
 |`sasl_mechanism`
 |STRING
 |No
-|SASL authentication mechanism (for example, `SCRAM-SHA-256`).
+|SASL authentication mechanism. Only `SCRAM-SHA-256` and `SCRAM-SHA-512` are accepted.
 
-|`sasl_username`
+|`sasl_user`
 |STRING
 |No
-|SASL username.
+|SASL username. Also used as the basic-authentication username for Schema Registry (and, when present, the HTTP Proxy endpoint set by `pandaproxy_url`).
 
 |`sasl_password`
 |STRING
 |No
-|SASL password.
+|SASL password. Also used as the basic-authentication password for Schema Registry (and, when present, the HTTP Proxy endpoint set by `pandaproxy_url`).
 
-|`tls_enabled`
+|`truststore`
 |STRING
 |No
-|Enable TLS for broker connections (`true` or `false`).
+|PEM-encoded CA certificate used to verify the TLS certificates presented by the Redpanda brokers, the Schema Registry, and (when set) the HTTP Proxy. Use TLS by specifying an `https://` URL for `schema_registry_url` and `pandaproxy_url`.
 
-|`schema_registry_username`
+|`key_store_key`
 |STRING
 |No
-|Schema Registry authentication username.
+|PEM-encoded client private key for mutual TLS (mTLS) to the Redpanda brokers. Must be set together with `key_store_cert`.
 
-|`schema_registry_password`
+|`key_store_cert`
 |STRING
 |No
-|Schema Registry authentication password.
+|PEM-encoded client certificate for mutual TLS (mTLS) to the Redpanda brokers. Must be set together with `key_store_key`.
+
+|`pandaproxy_url`
+|STRING
+|Conditional
+|Base URL of the Redpanda HTTP Proxy REST API. Required when the catalog includes a `USING CATALOG` clause; Redpanda SQL uses this endpoint to fetch Iceberg translation state for queries that span the topic and its Iceberg table.
 |===
 
 == Examples
@@ -75,7 +84,9 @@ WITH (
 );
 ----
 
-=== Create a catalog with authentication
+=== Create a catalog with SASL authentication and TLS
+
+The `sasl_user` and `sasl_password` values are used as the basic-authentication credentials for Schema Registry as well as the SASL credentials for the brokers. Use an `https://` URL for `schema_registry_url` and provide `truststore` (PEM-encoded CA) to verify the TLS certificate.
 
 [source,sql]
 ----
@@ -84,10 +95,39 @@ WITH (
   initial_brokers = 'broker1:9092',
   schema_registry_url = 'https://schema-registry:8081',
   sasl_mechanism = 'SCRAM-SHA-256',
-  sasl_username = 'admin',
+  sasl_user = 'admin',
   sasl_password = 'secret',
-  tls_enabled = 'true',
-  schema_registry_username = 'sr_user',
-  schema_registry_password = 'sr_pass'
+  truststore = '-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----'
+);
+----
+
+=== Create a catalog with mutual TLS
+
+Provide a client certificate and key for mTLS to the brokers in addition to the truststore.
+
+[source,sql]
+----
+CREATE REDPANDA CATALOG production_redpanda_mtls
+WITH (
+  initial_brokers = 'broker1:9092',
+  schema_registry_url = 'https://schema-registry:8081',
+  truststore = '-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----',
+  key_store_cert = '-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----',
+  key_store_key = '-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----'
 );
 ----
+
+=== Create a catalog linked to an Iceberg catalog
+
+Use the `USING CATALOG` clause to link a Redpanda catalog to an existing Iceberg catalog. Queries against tables in the linked Redpanda catalog can then return both live records from the Redpanda topic and Iceberg-translated records from its corresponding Iceberg table in a single result, with no overlap between them.
+
+[source,sql]
+----
+CREATE REDPANDA CATALOG redpanda_with_iceberg
+  USING CATALOG lakehouse_catalog
+  WITH (
+    initial_brokers = 'broker1:9092',
+    schema_registry_url = 'http://schema-registry:8081',
+    pandaproxy_url = 'http://redpanda:8082'
+  );
+----