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,208 @@
+= 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. The Iceberg catalog can be queried directly or linked to a Redpanda catalog with `USING CATALOG` so that bridge queries return a unified view of live and Iceberg-committed records. 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].
+
+== 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 for bridge queries.
+|===

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

@@ -9,11 +9,13 @@ 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.
 
 == Options
 
@@ -60,6 +62,11 @@ WITH (option = 'value' [, ...]);
 |STRING
 |No
 |Schema Registry authentication password.
+
+|`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
@@ -91,3 +98,18 @@ WITH (
   schema_registry_password = 'sr_pass'
 );
 ----
+
+=== 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 a non-overlapping continuum of data from the Redpanda topic and its corresponding Iceberg table in a single result.
+
+[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'
+  );
+----

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

@@ -0,0 +1,31 @@
+= DROP ICEBERG CATALOG
+:description: The DROP ICEBERG CATALOG statement removes an Iceberg catalog connection.
+:page-topic-type: reference
+
+The `DROP ICEBERG CATALOG` statement removes an Iceberg catalog connection.
+
+== Syntax
+
+[source,sql]
+----
+DROP ICEBERG CATALOG [IF EXISTS] catalog_name;
+----
+
+* `catalog_name`: Name of the Iceberg catalog to remove.
+* `IF EXISTS`: Optional. Prevents an error if the Iceberg catalog does not exist.
+
+== Examples
+
+Drop an Iceberg catalog:
+
+[source,sql]
+----
+DROP ICEBERG CATALOG lakehouse_catalog;
+----
+
+Drop the catalog only if it exists:
+
+[source,sql]
+----
+DROP ICEBERG CATALOG IF EXISTS lakehouse_catalog;
+----