You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/08/27 06:37:54 UTC
(fluss) branch main updated: [gateway] Add catalog and DDL backend operations with REST APIs (#4107)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 417216235 [gateway] Add catalog and DDL backend operations with REST APIs (#4107)
417216235 is described below
commit 41721623552c6281cc96fb1a23abf0ac155e9824
Author: Junbo Wang <be...@gmail.com>
AuthorDate: Thu Aug 27 14:37:50 2026 +0800
[gateway] Add catalog and DDL backend operations with REST APIs (#4107)
* [gateway] Add catalog and DDL backend operations with REST APIs
* [gateway] Simplify DDL conversion, test fixtures and comments
* [gateway] Simplify REST handling and align API status codes
* [gateway] Centralize REST query checks and shared helpers
* [gateway] Align DDL responses, properties and error reporting
* [gateway] Reserve table prefix for Fluss properties
* [gateway] Distinguish Fluss validation errors
---
fluss-gateway/Cargo.toml | 4 +-
fluss-gateway/openapi.yaml | 1708 ++++++++++++++++++++++++-
fluss-gateway/src/backend/client.rs | 144 ++-
fluss-gateway/src/backend/connection.rs | 4 +-
fluss-gateway/src/backend/errors.rs | 261 +++-
fluss-gateway/src/backend/fake.rs | 367 ++++--
fluss-gateway/src/backend/mod.rs | 82 +-
fluss-gateway/src/config.rs | 5 +-
fluss-gateway/src/error.rs | 38 +-
fluss-gateway/src/lifecycle.rs | 2 +-
fluss-gateway/src/observability.rs | 48 +-
fluss-gateway/src/protocol/rest/datatype.rs | 671 ++++++++++
fluss-gateway/src/protocol/rest/ddl.rs | 1426 +++++++++++++++++++++
fluss-gateway/src/protocol/rest/health.rs | 10 +-
fluss-gateway/src/protocol/rest/metadata.rs | 539 ++++++--
fluss-gateway/src/protocol/rest/mod.rs | 220 ++--
fluss-gateway/src/protocol/rest/openapi.rs | 167 ++-
fluss-gateway/src/protocol/rest/pagination.rs | 52 +-
fluss-gateway/tests/e2e_cluster.rs | 189 ++-
fluss-gateway/tests/http_api.rs | 2 +-
fluss-gateway/tests/support/mod.rs | 52 +
21 files changed, 5411 insertions(+), 580 deletions(-)
diff --git a/fluss-gateway/Cargo.toml b/fluss-gateway/Cargo.toml
index eaaae24ec..c59e34895 100644
--- a/fluss-gateway/Cargo.toml
+++ b/fluss-gateway/Cargo.toml
@@ -66,7 +66,7 @@ hyper = { version = "1", default-features = false, features = ["http1", "server"
# GracefulShutdown and the tower-to-hyper service adapter for the serve loop above. `http1` gates
# the `GracefulConnection` impl `graceful.watch` needs, so it is requested rather than inherited.
hyper-util = { version = "0.1", default-features = false, features = ["http1", "server-graceful", "service", "tokio"] }
-# Process CPU/RSS/fd sampling for the FIP-49 process_* metric families.
+# Process CPU/RSS/fd sampling for the process_* metric families.
libc = "0.2"
log = "0.4"
metrics = "0.24"
@@ -77,7 +77,7 @@ serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
# Preserves the nested typed field path after file, env, and CLI values are merged.
serde_path_to_error = "0.1"
-# The gateway.yaml configuration file is YAML with flat dotted keys (FIP-49).
+# The gateway.yaml configuration file is YAML with flat dotted keys.
# serde_yaml_ng is the maintained, API-compatible fork of the archived serde_yaml crate.
serde_yaml_ng = "0.10"
# Runtime capabilities used by the server; test-only I/O/time controls are enabled below.
diff --git a/fluss-gateway/openapi.yaml b/fluss-gateway/openapi.yaml
index 19a43af22..20fe78888 100644
--- a/fluss-gateway/openapi.yaml
+++ b/fluss-gateway/openapi.yaml
@@ -1,5 +1,17 @@
components:
schemas:
+ AlterTableBody:
+ additionalProperties: false
+ description: Body of `PATCH /v1/clusters/{cluster}/databases/{database}/tables/{table}`.
+ properties:
+ changes:
+ description: The server currently rejects mixing schema and config changes.
+ items:
+ $ref: '#/components/schemas/TableChangeBody'
+ type: array
+ required:
+ - changes
+ type: object
ClustersResponse:
description: 'Response of `GET /v1/clusters`: the IDs used in the cluster-scoped paths of the other APIs.'
properties:
@@ -10,6 +22,350 @@ components:
required:
- clusters
type: object
+ ColumnBody:
+ additionalProperties: false
+ description: One column; top-level nullability belongs here, not inside `data_type`.
+ properties:
+ comment:
+ type:
+ - string
+ - 'null'
+ data_type:
+ $ref: '#/components/schemas/ColumnDataType'
+ name:
+ type: string
+ nullable:
+ description: Primary-key columns default to false and reject true; other columns default to true.
+ type:
+ - boolean
+ - 'null'
+ required:
+ - name
+ - data_type
+ type: object
+ ColumnDataType:
+ description: A Fluss column type; top-level nullability is declared by the column.
+ oneOf:
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - BOOLEAN
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - TINYINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - SMALLINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - INTEGER
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - BIGINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - FLOAT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - DOUBLE
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ length:
+ format: int32
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - CHAR
+ type: string
+ required:
+ - length
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - STRING
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ precision:
+ format: int32
+ maximum: 38
+ minimum: 1
+ type: integer
+ scale:
+ format: int32
+ maximum: 38
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - DECIMAL
+ type: string
+ required:
+ - precision
+ - scale
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - DATE
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIME_WITHOUT_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIMESTAMP_WITHOUT_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ type:
+ enum:
+ - BYTES
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ length:
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - BINARY
+ type: string
+ required:
+ - length
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ element_type:
+ $ref: '#/components/schemas/WireDataType'
+ type:
+ enum:
+ - ARRAY
+ type: string
+ required:
+ - element_type
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ key_type:
+ $ref: '#/components/schemas/WireDataType'
+ type:
+ enum:
+ - MAP
+ type: string
+ value_type:
+ $ref: '#/components/schemas/WireDataType'
+ required:
+ - key_type
+ - value_type
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ fields:
+ items:
+ $ref: '#/components/schemas/WireRowField'
+ type: array
+ type:
+ enum:
+ - ROW
+ type: string
+ required:
+ - fields
+ - type
+ type: object
+ ColumnResponse:
+ description: One column of a table schema.
+ properties:
+ comment:
+ type: string
+ data_type:
+ $ref: '#/components/schemas/ColumnDataType'
+ name:
+ type: string
+ nullable:
+ type: boolean
+ required:
+ - name
+ - data_type
+ - nullable
+ type: object
+ CreateDatabaseBody:
+ additionalProperties: false
+ description: Body of `POST /v1/clusters/{cluster}/databases`.
+ properties:
+ database:
+ type: string
+ required:
+ - database
+ type: object
+ CreatePartitionBody:
+ additionalProperties: false
+ description: Body of `POST .../tables/{table}/partitions`.
+ properties:
+ partition:
+ additionalProperties:
+ type: string
+ propertyNames:
+ type: string
+ type: object
+ required:
+ - partition
+ type: object
+ CreateTableBody:
+ additionalProperties: false
+ description: Body of `POST /v1/clusters/{cluster}/databases/{database}/tables`.
+ properties:
+ columns:
+ items:
+ $ref: '#/components/schemas/ColumnBody'
+ type: array
+ comment:
+ type:
+ - string
+ - 'null'
+ configs:
+ additionalProperties:
+ type: string
+ description: Fluss table properties, validated by the server.
+ propertyNames:
+ type: string
+ type: object
+ custom_properties:
+ additionalProperties:
+ type: string
+ description: Custom table metadata; keys must not use the reserved `table.` prefix.
+ propertyNames:
+ type: string
+ type: object
+ distribution:
+ oneOf:
+ - type: 'null'
+ - $ref: '#/components/schemas/DistributionBody'
+ partitioned_by:
+ items:
+ type: string
+ type: array
+ primary_key:
+ description: Primary-key columns; omit for a log table.
+ items:
+ type: string
+ type:
+ - array
+ - 'null'
+ table_name:
+ type: string
+ validate_only:
+ description: Build and validate the definition locally without creating anything.
+ type: boolean
+ required:
+ - table_name
+ - columns
+ type: object
+ DatabaseResponse:
+ description: Response of a database creation.
+ properties:
+ database:
+ type: string
+ required:
+ - database
+ type: object
DatabasesResponse:
description: Response of `GET /v1/clusters/{cluster}/databases`.
properties:
@@ -19,12 +375,41 @@ components:
type: array
next_page_token:
description: Present only while more entries follow.
- type:
- - string
- - 'null'
+ type: string
required:
- databases
type: object
+ DistributionBody:
+ additionalProperties: false
+ description: The bucket distribution of a table definition.
+ properties:
+ bucket_count:
+ format: int32
+ minimum: 1
+ type: integer
+ bucket_keys:
+ items:
+ type: string
+ type: array
+ required:
+ - bucket_count
+ - bucket_keys
+ type: object
+ DistributionResponse:
+ description: The bucket distribution of a table.
+ properties:
+ bucket_count:
+ format: int32
+ minimum: 1
+ type: integer
+ bucket_keys:
+ items:
+ type: string
+ type: array
+ required:
+ - bucket_count
+ - bucket_keys
+ type: object
ErrorBody:
description: Body of the REST error envelope.
properties:
@@ -85,7 +470,7 @@ components:
- error
type: object
HealthResponse:
- description: 'Response of `GET /health` (FIP-49): liveness plus process uptime.'
+ description: 'Response of `GET /health`: liveness plus process uptime.'
properties:
status:
type: string
@@ -98,6 +483,53 @@ components:
- status
- uptime_ms
type: object
+ PartitionEntry:
+ description: One partition in a listing.
+ properties:
+ name:
+ description: Native partition name; percent-encode it as a path segment when deleting.
+ type: string
+ partition:
+ additionalProperties:
+ type: string
+ propertyNames:
+ type: string
+ type: object
+ required:
+ - name
+ - partition
+ type: object
+ PartitionResponse:
+ description: Response of a partition creation.
+ properties:
+ database:
+ type: string
+ partition:
+ additionalProperties:
+ type: string
+ propertyNames:
+ type: string
+ type: object
+ table:
+ type: string
+ required:
+ - database
+ - table
+ - partition
+ type: object
+ PartitionsResponse:
+ description: Response of `GET /v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions`.
+ properties:
+ next_page_token:
+ description: Present only while more entries follow.
+ type: string
+ partitions:
+ items:
+ $ref: '#/components/schemas/PartitionEntry'
+ type: array
+ required:
+ - partitions
+ type: object
ReadyResponse:
description: 'Response of `GET /ready`: the gateway accepts application traffic.'
properties:
@@ -106,14 +538,104 @@ components:
required:
- status
type: object
+ TableChangeBody:
+ description: |-
+ One supported table change. Config keys under `table.` address Fluss table properties;
+ other keys address custom properties.
+ oneOf:
+ - description: 'An added column must be nullable: existing rows have no value for it.'
+ properties:
+ comment:
+ type:
+ - string
+ - 'null'
+ data_type:
+ $ref: '#/components/schemas/ColumnDataType'
+ kind:
+ enum:
+ - add_column
+ type: string
+ name:
+ type: string
+ required:
+ - name
+ - data_type
+ - kind
+ type: object
+ - properties:
+ key:
+ type: string
+ kind:
+ enum:
+ - set_config
+ type: string
+ value:
+ type: string
+ required:
+ - key
+ - value
+ - kind
+ type: object
+ - description: Removes the explicitly configured value.
+ properties:
+ key:
+ type: string
+ kind:
+ enum:
+ - reset_config
+ type: string
+ required:
+ - key
+ - kind
+ type: object
+ TableResponse:
+ description: Response of `GET /v1/clusters/{cluster}/databases/{database}/tables/{table}`.
+ properties:
+ columns:
+ items:
+ $ref: '#/components/schemas/ColumnResponse'
+ type: array
+ comment:
+ type: string
+ configs:
+ additionalProperties:
+ type: string
+ description: Fluss table properties.
+ propertyNames:
+ type: string
+ type: object
+ custom_properties:
+ additionalProperties:
+ type: string
+ description: Custom table metadata.
+ propertyNames:
+ type: string
+ type: object
+ database:
+ type: string
+ distribution:
+ $ref: '#/components/schemas/DistributionResponse'
+ partitioned_by:
+ items:
+ type: string
+ type: array
+ primary_key:
+ items:
+ type: string
+ type: array
+ table:
+ type: string
+ required:
+ - database
+ - table
+ - columns
+ type: object
TablesResponse:
description: Response of `GET /v1/clusters/{cluster}/databases/{database}/tables`.
properties:
next_page_token:
description: Present only while more entries follow.
- type:
- - string
- - 'null'
+ type: string
tables:
items:
type: string
@@ -121,6 +643,308 @@ components:
required:
- tables
type: object
+ ValidateOnlyResponse:
+ description: The result of a successful local `validate_only` dry run.
+ properties:
+ column_count:
+ minimum: 0
+ type: integer
+ database:
+ type: string
+ primary_key:
+ items:
+ type: string
+ type: array
+ table:
+ type: string
+ validate_only:
+ type: boolean
+ required:
+ - validate_only
+ - database
+ - table
+ - column_count
+ - primary_key
+ type: object
+ WireDataType:
+ description: The exact recursive Fluss type as it appears on the wire.
+ oneOf:
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - BOOLEAN
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - TINYINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - SMALLINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - INTEGER
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - BIGINT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - FLOAT
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - DOUBLE
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ length:
+ format: int32
+ minimum: 0
+ type: integer
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - CHAR
+ type: string
+ required:
+ - length
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - STRING
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ precision:
+ format: int32
+ maximum: 38
+ minimum: 1
+ type: integer
+ scale:
+ format: int32
+ maximum: 38
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - DECIMAL
+ type: string
+ required:
+ - precision
+ - scale
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - DATE
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIME_WITHOUT_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIMESTAMP_WITHOUT_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ precision:
+ format: int32
+ maximum: 9
+ minimum: 0
+ type: integer
+ type:
+ enum:
+ - TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ type: string
+ required:
+ - precision
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - BYTES
+ type: string
+ required:
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ length:
+ minimum: 0
+ type: integer
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - BINARY
+ type: string
+ required:
+ - length
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ element_type:
+ $ref: '#/components/schemas/WireDataType'
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - ARRAY
+ type: string
+ required:
+ - element_type
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ key_type:
+ $ref: '#/components/schemas/WireDataType'
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - MAP
+ type: string
+ value_type:
+ $ref: '#/components/schemas/WireDataType'
+ required:
+ - key_type
+ - value_type
+ - type
+ type: object
+ - additionalProperties: false
+ properties:
+ fields:
+ items:
+ $ref: '#/components/schemas/WireRowField'
+ type: array
+ nullable:
+ type: boolean
+ type:
+ enum:
+ - ROW
+ type: string
+ required:
+ - fields
+ - type
+ type: object
+ WireRowField:
+ additionalProperties: false
+ description: One `ROW` field. Server-assigned field IDs are not exposed.
+ properties:
+ description:
+ type: string
+ field_type:
+ $ref: '#/components/schemas/WireDataType'
+ name:
+ type: string
+ required:
+ - name
+ - field_type
+ type: object
info:
description: Stateless REST gateway for Apache Fluss
license:
@@ -152,7 +976,7 @@ paths:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
description: Wrong method for this route
- summary: 'The FIP-49 health summary: `{status, uptime_ms}`, always 200 while the process answers.'
+ summary: Returns process liveness and uptime without a backend RPC.
tags:
- health
/ready:
@@ -283,6 +1107,12 @@ paths:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
description: Unknown cluster
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
'429':
content:
application/json:
@@ -295,6 +1125,12 @@ paths:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
'503':
content:
application/json:
@@ -310,9 +1146,8 @@ paths:
summary: Lists the databases of one configured cluster.
tags:
- metadata
- /v1/clusters/{cluster}/databases/{database}/tables:
- get:
- operationId: listTables
+ post:
+ operationId: createDatabase
parameters:
- description: Configured cluster ID
in: path
@@ -320,57 +1155,236 @@ paths:
required: true
schema:
type: string
- - description: Database name
- in: path
- name: database
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreateDatabaseBody'
required: true
- schema:
- type: string
- - description: Maximum entries to return. Defaults to 100, capped at 1000.
- in: query
- name: max_results
- required: false
- schema:
- maximum: 1000
- minimum: 1
- type: integer
- - description: Opaque token from the `next_page_token` of the previous response.
- in: query
- name: page_token
- required: false
- schema:
- type: string
responses:
- '200':
+ '201':
content:
application/json:
schema:
- $ref: '#/components/schemas/TablesResponse'
- description: Tables in lexical order
+ $ref: '#/components/schemas/DatabaseResponse'
+ description: The created database
+ headers:
+ Location:
+ description: Created database URL
+ schema:
+ type: string
'400':
content:
application/json:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
- description: Invalid page parameter or page token
+ description: Malformed body or invalid definition
'404':
content:
application/json:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
- description: Unknown cluster or database
- '429':
+ description: Unknown cluster
+ '409':
content:
application/json:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
- description: Metadata concurrency limit exceeded
- '500':
+ description: The database already exists
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '415':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The body is not JSON
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Creates one database.
+ tags:
+ - ddl
+ /v1/clusters/{cluster}/databases/{database}:
+ delete:
+ operationId: dropDatabase
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ responses:
+ '204':
+ description: The database was dropped
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unsupported query parameter
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster or database
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The database still holds tables
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Drops a database only if it contains no tables.
+ tags:
+ - ddl
+ /v1/clusters/{cluster}/databases/{database}/tables:
+ get:
+ operationId: listTables
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Maximum entries to return. Defaults to 100, capped at 1000.
+ in: query
+ name: max_results
+ required: false
+ schema:
+ maximum: 1000
+ minimum: 1
+ type: integer
+ - description: Opaque token from the `next_page_token` of the previous response.
+ in: query
+ name: page_token
+ required: false
+ schema:
+ type: string
+ responses:
+ '200':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/TablesResponse'
+ description: Tables in lexical order
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Invalid page parameter or page token
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster or database
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
content:
application/json:
schema:
$ref: '#/components/schemas/ErrorEnvelope'
description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
'503':
content:
application/json:
@@ -386,6 +1400,624 @@ paths:
summary: Lists the tables of one database.
tags:
- metadata
+ post:
+ operationId: createTable
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreateTableBody'
+ required: true
+ responses:
+ '200':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ValidateOnlyResponse'
+ description: 'Dry run: fluss-rs accepted the definition and nothing was created'
+ '201':
+ description: The table was created; GET Location to read its metadata
+ headers:
+ Location:
+ description: Created table URL
+ schema:
+ type: string
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Malformed body or invalid definition
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster or database
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The table already exists
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '415':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The body is not JSON
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Creates one table, or validates its definition without creating it.
+ tags:
+ - ddl
+ /v1/clusters/{cluster}/databases/{database}/tables/{table}:
+ delete:
+ operationId: dropTable
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ responses:
+ '204':
+ description: The table was dropped
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unsupported query parameter
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, or table
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Drops one table.
+ tags:
+ - ddl
+ get:
+ operationId: describeTable
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ responses:
+ '200':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/TableResponse'
+ description: The table metadata
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unsupported query parameter
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, or table
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: 'Describes one table: its schema, keys, distribution, partitioning, and properties.'
+ tags:
+ - metadata
+ patch:
+ operationId: alterTable
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/AlterTableBody'
+ required: true
+ responses:
+ '204':
+ description: The table was altered; GET the same URL to read its metadata
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Malformed body or invalid change
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, or table
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '415':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The body is not JSON
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Applies one group of schema or config changes to a table.
+ tags:
+ - ddl
+ /v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions:
+ get:
+ operationId: listPartitions
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ - description: Maximum entries to return. Defaults to 100, capped at 1000.
+ in: query
+ name: max_results
+ required: false
+ schema:
+ maximum: 1000
+ minimum: 1
+ type: integer
+ - description: Opaque token from the `next_page_token` of the previous response.
+ in: query
+ name: page_token
+ required: false
+ schema:
+ type: string
+ responses:
+ '200':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/PartitionsResponse'
+ description: Partitions in partition-name order
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Invalid page parameter or page token
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, or table
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Lists the partitions of one partitioned table.
+ tags:
+ - metadata
+ post:
+ operationId: createPartition
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreatePartitionBody'
+ required: true
+ responses:
+ '201':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/PartitionResponse'
+ description: The created partition
+ headers:
+ Location:
+ description: Created partition URL
+ schema:
+ type: string
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Malformed body, or a spec that does not match the partition keys
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, or table
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The partition already exists or the table's partition limit was reached
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '415':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: The body is not JSON
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Creates one partition of a partitioned table.
+ tags:
+ - ddl
+ /v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions/{partition}:
+ delete:
+ operationId: dropPartition
+ parameters:
+ - description: Configured cluster ID
+ in: path
+ name: cluster
+ required: true
+ schema:
+ type: string
+ - description: Database name
+ in: path
+ name: database
+ required: true
+ schema:
+ type: string
+ - description: Table name
+ in: path
+ name: table
+ required: true
+ schema:
+ type: string
+ - description: Fluss partition name; values follow partition-key order and are joined by `$`
+ in: path
+ name: partition
+ required: true
+ schema:
+ type: string
+ responses:
+ '204':
+ description: The partition was dropped
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unsupported query parameter, or a name that does not match the partition keys
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Unknown cluster, database, table, or partition
+ '413':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request body above the configured limit
+ '429':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Metadata concurrency limit exceeded
+ '500':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss backend failure
+ '501':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss does not support the operation or API version
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Fluss is unavailable, or the gateway is starting or shutting down
+ '504':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ErrorEnvelope'
+ description: Request deadline exceeded
+ summary: Drops one partition by its Fluss name.
+ tags:
+ - ddl
/v1/openapi.json:
get:
operationId: getOpenApi
diff --git a/fluss-gateway/src/backend/client.rs b/fluss-gateway/src/backend/client.rs
index 03c88bfb3..73642e654 100644
--- a/fluss-gateway/src/backend/client.rs
+++ b/fluss-gateway/src/backend/client.rs
@@ -15,15 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! The production [`FlussBackend`] over `fluss-rs`.
-//!
-//! It owns one `ConnectionCache` per configured cluster, which is the whole of the gateway's connection
-//! management: routing a request to its cluster and sharing its service-mode connection. None of that
-//! is visible above this module.
-//!
-//! This module and [`crate::backend::connection`] own native client handles, connection lifecycle, and
-//! transport error translation. Protocol adapters may still reuse stable `fluss-rs` domain types, such
-//! as metadata descriptors, when they fit the API instead of defining duplicate models.
+//! The production [`FlussBackend`] over one connection cache per configured cluster.
use crate::backend::FlussBackend;
use crate::backend::connection::{ConnectionCache, NativeConnector};
@@ -36,20 +28,20 @@ use crate::error::GatewayResult;
use async_trait::async_trait;
use fluss::client::FlussAdmin;
use fluss::error::Error as FlussClientError;
+use fluss::metadata::{
+ AlterTableChanges, PartitionInfo, PartitionSpec, TableDescriptor, TableInfo, TablePath,
+};
use futures_util::future::join_all;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
pub struct NativeFlussBackend {
- /// The cluster routing table. Built from configuration and never mutated, so it needs no lock and
- /// no request can add, remove, or reorder a cluster.
caches: BTreeMap<ClusterId, ConnectionCache<NativeConnector>>,
}
impl NativeFlussBackend {
- /// Performs no I/O: the process starts while Fluss is down, and the first request of a cluster
- /// dials it.
+ /// Builds the routing table without connecting.
pub fn from_config(config: &GatewayConfig) -> Self {
let caches = config
.clusters
@@ -65,10 +57,7 @@ impl NativeFlussBackend {
Self { caches }
}
- /// Closes every connection of every cluster within `timeout`. Idempotent.
- ///
- /// Concurrent because the budget is the whole shutdown's, not each cluster's: closing sequentially
- /// would let one slow cluster consume the time the others need.
+ /// Closes all clusters concurrently within one timeout.
pub(crate) async fn close(&self, timeout: Duration) -> GatewayResult<()> {
let closes = join_all(self.caches.values().map(|cache| cache.close(timeout))).await;
let mut first_failure = None;
@@ -81,26 +70,18 @@ impl NativeFlussBackend {
first_failure.map_or(Ok(()), Err)
}
- /// Runs one idle scan for every configured cluster. Each cache logs its own best-effort close
- /// failures, so a slow or unavailable cluster cannot suppress cleanup of the others.
+ /// Runs one idle scan for every configured cluster.
pub(crate) async fn clean_expired_connections(&self) {
join_all(self.caches.values().map(ConnectionCache::clean_expired)).await;
}
- /// The only place `cluster_not_found` originates.
fn cache_for(&self, ctx: &RequestContext) -> GatewayResult<&ConnectionCache<NativeConnector>> {
self.caches
.get(ctx.cluster_id())
.ok_or_else(|| unknown_cluster(ctx.cluster_id().as_str()))
}
- /// The single entry point of every admin call: route, run under the request budget, take the
- /// cluster's service connection, then classify the failure.
- ///
- /// A failure never evicts the connection. `fluss-rs` reports a broken transport per server and
- /// reconnects that server on the next use, so the logical client recovers on its own; discarding it
- /// would only throw away its cluster metadata and cached sub-clients.
- ///
+ /// Routes and bounds one native admin call, then classifies its failure.
async fn admin_call<T, F, Fut>(
&self,
ctx: &RequestContext,
@@ -118,7 +99,7 @@ impl NativeFlussBackend {
Ok(admin) => operation(admin).await,
Err(error) => Err(error),
};
- result.map_err(|native| map_fluss_error(what, native))
+ result.map_err(|native| map_fluss_error(what, native, Some(ctx)))
})
.await
}
@@ -146,9 +127,100 @@ impl FlussBackend for NativeFlussBackend {
ctx: &RequestContext,
database: &str,
) -> GatewayResult<Vec<String>> {
- let database = database.to_string();
self.admin_call(ctx, "list the tables", |admin| async move {
- admin.list_tables(&database).await
+ admin.list_tables(database).await
+ })
+ .await
+ }
+
+ async fn create_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()> {
+ self.admin_call(ctx, "create the database", |admin| async move {
+ admin.create_database(database, None, false).await
+ })
+ .await
+ }
+
+ async fn drop_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()> {
+ self.admin_call(ctx, "drop the database", |admin| async move {
+ admin.drop_database(database, false, false).await
+ })
+ .await
+ }
+
+ /// Reads through fluss-rs, which refreshes its metadata cache.
+ async fn describe_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<TableInfo> {
+ self.admin_call(ctx, "describe the table", |admin| async move {
+ admin.get_table_info(table).await
+ })
+ .await
+ }
+
+ async fn create_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ descriptor: &TableDescriptor,
+ ) -> GatewayResult<()> {
+ self.admin_call(ctx, "create the table", |admin| async move {
+ admin.create_table(table, descriptor, false).await
+ })
+ .await
+ }
+
+ async fn alter_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ changes: AlterTableChanges,
+ ) -> GatewayResult<()> {
+ self.admin_call(ctx, "alter the table", |admin| async move {
+ admin.alter_table(table, false, changes).await
+ })
+ .await
+ }
+
+ async fn drop_table(&self, ctx: &RequestContext, table: &TablePath) -> GatewayResult<()> {
+ self.admin_call(ctx, "drop the table", |admin| async move {
+ admin.drop_table(table, false).await
+ })
+ .await
+ }
+
+ async fn list_partitions(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<Vec<PartitionInfo>> {
+ self.admin_call(ctx, "list the partitions", |admin| async move {
+ admin.list_partition_infos(table).await
+ })
+ .await
+ }
+
+ async fn create_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()> {
+ self.admin_call(ctx, "create the partition", |admin| async move {
+ admin.create_partition(table, spec, false).await
+ })
+ .await
+ }
+
+ async fn drop_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()> {
+ self.admin_call(ctx, "drop the partition", |admin| async move {
+ admin.drop_partition(table, spec, false).await
})
.await
}
@@ -174,8 +246,6 @@ mod tests {
ClusterConfig::default()
}
- /// Every configured cluster is routable in lexical order, and a malformed or unconfigured ID is
- /// the same 404 — resolved from configuration alone, without any connection attempt.
#[tokio::test]
async fn routing_answers_only_from_configuration() {
let backend = backend(&[("zeta", service_cluster()), ("alpha", service_cluster())]);
@@ -200,12 +270,9 @@ mod tests {
);
}
- /// The backend constructs without touching the network, so both ways a dial can fail surface on
- /// the first request rather than at startup. Shutdown then has nothing to close.
#[tokio::test]
async fn a_dial_failure_reaches_the_caller_classified() {
let unreachable = ClusterConfig {
- // Port 1 has no listener in any test environment; the connect timeout bounds the attempt.
bootstrap_servers: "127.0.0.1:1".to_string(),
connect_timeout: ConfigDuration::from_millis(200),
..service_cluster()
@@ -235,12 +302,6 @@ mod tests {
(error.kind(), error.code())
}
- /// A server that accepts the connection and never answers must still end at the deadline.
- ///
- /// Nothing below the gateway ends this wait: `fluss-rs` has no per-RPC timeout, and the connect
- /// timeout is already satisfied by the accepted TCP connection. Only [`RequestContext::run`] does.
- /// The abandoned RPC stays registered on the native connection — see the note on
- /// `NativeConnector::dial`.
#[tokio::test]
async fn a_server_that_never_answers_ends_at_the_request_deadline() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -250,7 +311,6 @@ mod tests {
tokio::spawn(async move {
let mut accepted = Vec::new();
while let Ok((stream, _)) = listener.accept().await {
- // Held open and never answered.
accepted.push(stream);
}
});
diff --git a/fluss-gateway/src/backend/connection.rs b/fluss-gateway/src/backend/connection.rs
index 25c0edcf2..0f16f32d3 100644
--- a/fluss-gateway/src/backend/connection.rs
+++ b/fluss-gateway/src/backend/connection.rs
@@ -298,7 +298,7 @@ impl<K: Connector> ConnectionCache<K> {
for result in results {
if let Err(error) = result {
first_failure = first_failure
- .or_else(|| Some(map_fluss_error("close the Fluss connection", error)));
+ .or_else(|| Some(map_fluss_error("close the Fluss connection", error, None)));
}
observability::connection_closed(self.cluster.as_str(), "shutdown");
}
@@ -325,7 +325,7 @@ impl<K: Connector> ConnectionCache<K> {
.connector
.dial(self.identity.as_ref())
.await
- .map_err(|native| map_fluss_error("connect to Fluss", native))?;
+ .map_err(|native| map_fluss_error("connect to Fluss", native, None))?;
Ok(self.install(&entry, connection))
}
diff --git a/fluss-gateway/src/backend/errors.rs b/fluss-gateway/src/backend/errors.rs
index ee7c45e1f..8554dbda7 100644
--- a/fluss-gateway/src/backend/errors.rs
+++ b/fluss-gateway/src/backend/errors.rs
@@ -16,48 +16,45 @@
// under the License.
//! Classification of native `fluss-rs` failures.
-//!
-//! Its own module because both layers above it need it: the pool classifies a failed dial, and the
-//! backend classifies a failed operation. Nothing outside this module inspects a `fluss` error type.
-use crate::error::{GatewayError, Resource};
+use crate::backend::context::RequestContext;
+use crate::error::{ErrorKind, GatewayError, Resource};
use fluss::error::{Error as FlussClientError, FlussError};
-/// Classifies one native failure, with `what` naming the attempted operation in the client-facing
-/// message.
-///
-/// The rules the gateway commits to: an error code Fluss defines for a resource keeps that meaning,
-/// service authentication or authorization failures are backend deployment faults, a retriable
-/// transport failure is a transient outage, and anything the client cannot classify is reported as a
-/// backend failure rather than a gateway-internal one. Messages carry the operation, never the native
-/// error text, which can contain addresses and payload detail; that goes to the log.
-pub(crate) fn map_fluss_error(what: &str, error: FlussClientError) -> GatewayError {
- if let Some(api_error) = error.api_error()
- && let Some(mapped) = map_api_error(what, api_error)
- {
- return mapped;
- }
- match &error {
- FlussClientError::UnsupportedOperation { .. }
- | FlussClientError::UnsupportedVersion { .. } => {
- log::warn!("Fluss does not support the request while trying to {what}: {error}");
- GatewayError::unsupported(format!(
+/// Logs and classifies a native failure without exposing its detail to clients.
+pub(crate) fn map_fluss_error(
+ what: &str,
+ error: FlussClientError,
+ ctx: Option<&RequestContext>,
+) -> GatewayError {
+ let mapped = error
+ .api_error()
+ .and_then(|api_error| map_api_error(what, api_error))
+ .unwrap_or_else(|| match &error {
+ FlussClientError::UnsupportedOperation { .. }
+ | FlussClientError::UnsupportedVersion { .. } => GatewayError::unsupported(format!(
"Fluss does not support the request while trying to {what}"
- ))
- }
- FlussClientError::IllegalArgument { .. } => GatewayError::invalid_argument(format!(
- "Fluss rejected the request while trying to {what}"
- )),
- _ if error.is_retriable() => {
- log::warn!("Fluss is temporarily unavailable while trying to {what}: {error}");
- GatewayError::unavailable(format!("Fluss is unavailable while trying to {what}"))
- }
- _ => {
- log::error!("the Fluss request failed while trying to {what}: {error}");
- // An unclassifiable native failure is the backend's, not the gateway's (FIP-49 `backend`).
- GatewayError::backend(format!("Fluss failed while trying to {what}"))
- }
- }
+ )),
+ FlussClientError::IllegalArgument { .. } => GatewayError::invalid_argument(format!(
+ "Fluss rejected the request while trying to {what}"
+ )),
+ _ if error.is_retriable() => {
+ GatewayError::unavailable(format!("Fluss is unavailable while trying to {what}"))
+ }
+ _ => GatewayError::backend(format!("Fluss failed while trying to {what}")),
+ });
+ let level = match mapped.kind() {
+ ErrorKind::Backend | ErrorKind::Internal => log::Level::Error,
+ ErrorKind::Unavailable | ErrorKind::Unsupported => log::Level::Warn,
+ _ => log::Level::Debug,
+ };
+ log::log!(
+ level,
+ "request_id={} cluster={} operation={what:?} error={error}",
+ ctx.map_or("-", RequestContext::request_id),
+ ctx.map_or("-", |ctx| ctx.cluster_id().as_str()),
+ );
+ mapped
}
/// Maps the protocol error codes that carry a meaning of their own. `None` falls through to the
@@ -80,23 +77,67 @@ fn map_api_error(what: &str, api_error: FlussError) -> Option<GatewayError> {
GatewayError::already_exists(format!("the table already exists while trying to {what}"))
.with_resource(Resource::Table)
}
+ // A schema is not a resource the API addresses on its own, so a missing one is reported
+ // against the table it belongs to; that is the resource a caller can act on.
+ FlussError::SchemaNotExist => GatewayError::not_found(format!(
+ "the table schema does not exist while trying to {what}"
+ ))
+ .with_resource(Resource::Table),
+ FlussError::PartitionNotExists => GatewayError::not_found(format!(
+ "the partition does not exist while trying to {what}"
+ ))
+ .with_resource(Resource::Partition),
+ FlussError::PartitionAlreadyExists => GatewayError::already_exists(format!(
+ "the partition already exists while trying to {what}"
+ ))
+ .with_resource(Resource::Partition),
+ FlussError::TableNotPartitionedException => GatewayError::invalid_argument(format!(
+ "the table is not partitioned while trying to {what}"
+ ))
+ .with_resource(Resource::Partition),
+ FlussError::PartitionSpecInvalidException => GatewayError::invalid_argument(format!(
+ "Fluss rejected the partition spec while trying to {what}"
+ ))
+ .with_resource(Resource::Partition),
+ // The table is at its configured partition limit. A caller can act on it, by dropping a
+ // partition, which is what makes it a precondition rather than a gateway capacity failure.
+ FlussError::PartitionMaxNumException => GatewayError::failed_precondition(format!(
+ "the table holds the maximum number of partitions, {what} refused"
+ ))
+ .with_resource(Resource::Partition),
FlussError::DatabaseNotEmpty => {
GatewayError::failed_precondition(format!("the database is not empty, {what} refused"))
.with_resource(Resource::Database)
}
- FlussError::InvalidDatabaseException | FlussError::InvalidTableException => {
- GatewayError::invalid_argument(format!(
- "Fluss rejected the name while trying to {what}"
- ))
+ FlussError::InvalidDatabaseException => GatewayError::invalid_argument(format!(
+ "Fluss rejected the name while trying to {what}"
+ )),
+ FlussError::InvalidTableException => GatewayError::invalid_argument(format!(
+ "Fluss rejected the table request while trying to {what}"
+ )),
+ FlussError::InvalidConfigException => GatewayError::invalid_argument(format!(
+ "the configuration is invalid while trying to {what}"
+ )),
+ FlussError::InvalidAlterTableException => GatewayError::invalid_argument(format!(
+ "Fluss rejected the requested table alteration while trying to {what}"
+ )),
+ FlussError::InvalidReplicationFactor => GatewayError::invalid_argument(format!(
+ "the replication factor is invalid while trying to {what}"
+ )),
+ FlussError::BucketMaxNumException => GatewayError::invalid_argument(format!(
+ "the requested bucket count exceeds the maximum while trying to {what}"
+ )),
+ FlussError::RequestTimeOut => {
+ GatewayError::deadline_exceeded(format!("Fluss timed out while trying to {what}"))
}
- // In service mode, both failures are deployment faults, including protocol mismatches.
- // User mode may later map a caller-specific authorization failure to 403.
+ FlussError::UnsupportedVersion => GatewayError::unsupported(format!(
+ "Fluss does not support the request while trying to {what}"
+ )),
+ // TODO: Map caller authorization failures to 403 when fluss-rs supports user mode.
FlussError::AuthenticateException => {
- log::error!("Fluss rejected the gateway connection while trying to {what}");
GatewayError::backend(format!("Fluss rejected the gateway while trying to {what}"))
}
FlussError::AuthorizationException => {
- log::error!("Fluss denied the gateway while trying to {what}");
GatewayError::backend(format!("Fluss denied the gateway while trying to {what}"))
}
_ => return None,
@@ -106,9 +147,27 @@ fn map_api_error(what: &str, api_error: FlussError) -> Option<GatewayError> {
#[cfg(test)]
pub(crate) mod tests {
use super::*;
- use crate::error::ErrorKind;
+ use std::cell::RefCell;
+ use std::time::Duration;
+
+ thread_local! {
+ static LOGS: RefCell<Vec<(log::Level, String)>> = const { RefCell::new(Vec::new()) };
+ }
+
+ struct TestLogger;
+
+ impl log::Log for TestLogger {
+ fn enabled(&self, _: &log::Metadata<'_>) -> bool {
+ true
+ }
+
+ fn log(&self, record: &log::Record<'_>) {
+ LOGS.with_borrow_mut(|logs| logs.push((record.level(), record.args().to_string())));
+ }
+
+ fn flush(&self) {}
+ }
- /// A native failure carrying one Fluss protocol error code.
pub(crate) fn api_failure(error: FlussError) -> FlussClientError {
FlussClientError::FlussAPIError {
api_error: fluss::error::ApiError {
@@ -118,10 +177,12 @@ pub(crate) mod tests {
}
}
- /// Every mapped condition keeps the class and the wire code a caller can act on, and no message
- /// quotes the native error text.
#[test]
- fn native_failures_map_to_their_gateway_class_and_code() {
+ fn native_failures_map_to_gateway_errors_and_logs() {
+ log::set_logger(&TestLogger).expect("install the test logger");
+ log::set_max_level(log::LevelFilter::Debug);
+ let ctx = RequestContext::for_test("default", Duration::from_secs(5));
+ let operation = "complete the request";
let cases = [
(
api_failure(FlussError::DatabaseNotExist),
@@ -148,11 +209,31 @@ pub(crate) mod tests {
ErrorKind::FailedPrecondition,
"database_not_empty",
),
+ (
+ api_failure(FlussError::PartitionMaxNumException),
+ ErrorKind::FailedPrecondition,
+ "failed_precondition",
+ ),
+ (
+ api_failure(FlussError::TableNotPartitionedException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
+ (
+ api_failure(FlussError::PartitionSpecInvalidException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
(
api_failure(FlussError::InvalidDatabaseException),
ErrorKind::InvalidArgument,
"invalid_argument",
),
+ (
+ api_failure(FlussError::InvalidTableException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
(
api_failure(FlussError::AuthenticateException),
ErrorKind::Backend,
@@ -163,6 +244,36 @@ pub(crate) mod tests {
ErrorKind::Backend,
"backend",
),
+ (
+ api_failure(FlussError::InvalidConfigException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
+ (
+ api_failure(FlussError::InvalidAlterTableException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
+ (
+ api_failure(FlussError::InvalidReplicationFactor),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
+ (
+ api_failure(FlussError::BucketMaxNumException),
+ ErrorKind::InvalidArgument,
+ "invalid_argument",
+ ),
+ (
+ api_failure(FlussError::RequestTimeOut),
+ ErrorKind::DeadlineExceeded,
+ "timeout",
+ ),
+ (
+ api_failure(FlussError::UnsupportedVersion),
+ ErrorKind::Unsupported,
+ "unsupported",
+ ),
(
FlussClientError::UnsupportedVersion {
message: "server detail".to_string(),
@@ -196,20 +307,56 @@ pub(crate) mod tests {
),
];
for (native, expected_kind, expected_code) in cases {
+ let api_error = native.api_error();
let rendered = native.to_string();
- let mapped = map_fluss_error("list the databases", native);
+ let mapped = map_fluss_error(operation, native, Some(&ctx));
assert_eq!(mapped.kind(), expected_kind, "{rendered}");
assert_eq!(mapped.code(), expected_code, "{rendered}");
- assert!(
- mapped.message().contains("list the databases"),
- "{}",
- mapped.message()
- );
+ assert!(mapped.message().contains(operation), "{}", mapped.message());
assert!(
!mapped.message().contains("server detail"),
"the native detail must stay in the log: {}",
mapped.message()
);
+ let expected_reason = match api_error {
+ Some(FlussError::InvalidDatabaseException) => Some("Fluss rejected the name"),
+ Some(FlussError::InvalidTableException) => Some("Fluss rejected the table request"),
+ Some(FlussError::InvalidConfigException) => Some("the configuration is invalid"),
+ Some(FlussError::InvalidAlterTableException) => {
+ Some("Fluss rejected the requested table alteration")
+ }
+ Some(FlussError::InvalidReplicationFactor) => {
+ Some("the replication factor is invalid")
+ }
+ Some(FlussError::BucketMaxNumException) => {
+ Some("the requested bucket count exceeds the maximum")
+ }
+ Some(FlussError::TableNotPartitionedException) => {
+ Some("the table is not partitioned")
+ }
+ Some(FlussError::PartitionSpecInvalidException) => {
+ Some("Fluss rejected the partition spec")
+ }
+ _ => None,
+ };
+ if let Some(reason) = expected_reason {
+ assert_eq!(
+ mapped.message(),
+ format!("{reason} while trying to {operation}")
+ );
+ }
+ let logs = LOGS.take();
+ assert_eq!(logs.len(), 1, "{logs:?}");
+ if expected_kind == ErrorKind::InvalidArgument {
+ assert_eq!(logs[0].0, log::Level::Debug);
+ }
+ assert!(logs[0].1.contains(&rendered), "{logs:?}");
+ assert!(logs[0].1.contains("request_id=test-request"), "{logs:?}");
+ assert!(logs[0].1.contains("cluster=default"), "{logs:?}");
+ assert!(
+ logs[0].1.contains(&format!("operation={operation:?}")),
+ "{logs:?}"
+ );
}
}
}
diff --git a/fluss-gateway/src/backend/fake.rs b/fluss-gateway/src/backend/fake.rs
index 2eafc86fc..774d6b202 100644
--- a/fluss-gateway/src/backend/fake.rs
+++ b/fluss-gateway/src/backend/fake.rs
@@ -15,34 +15,60 @@
// specific language governing permissions and limitations
// under the License.
-//! In-memory [`FlussBackend`] for the protocol tests (FIP-49 test plan).
-//!
-//! It needs no Fluss cluster and is compiled only under `cfg(test)`, so it can never be part of a
-//! shippable path. Replacing the whole backend also replaces the connection pool, which is why the
-//! pool has its own tests against an injected connector.
+//! Fixed catalog fixtures, recorded mutations, and operation-scoped failures for protocol tests.
use crate::backend::context::RequestContext;
use crate::backend::types::ClusterId;
use crate::backend::{FlussBackend, unknown_cluster};
use crate::error::{GatewayError, GatewayResult, Resource};
use async_trait::async_trait;
-use std::collections::BTreeMap;
+use fluss::metadata::{
+ AlterTableChanges, DataType, PartitionInfo, PartitionSpec, Schema, TableDescriptor, TableInfo,
+ TablePath,
+};
+use std::collections::{BTreeMap, HashMap, btree_map::Entry};
use std::sync::{Mutex, MutexGuard, PoisonError};
-use std::time::Duration;
-/// The injected behaviour of one fixture backend.
+const FIXTURE_TIME: i64 = 1_700_000_000_000;
+
+struct FakeTable {
+ info: TableInfo,
+ partitions: Vec<PartitionInfo>,
+}
+
+#[derive(Default)]
struct FakeState {
- databases: BTreeMap<String, Vec<String>>,
- /// Returned by every catalog call instead of the fixture content.
- failure: Option<GatewayError>,
- /// Delay before answering, for the deadline tests.
- latency: Duration,
+ databases: BTreeMap<String, BTreeMap<String, FakeTable>>,
+ calls: Vec<FakeCall>,
+ failures: HashMap<Operation, GatewayError>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Operation {
+ ListDatabases,
+ CreateDatabase,
+ DropDatabase,
+ ListTables,
+ DescribeTable,
+ CreateTable,
+ AlterTable,
+ DropTable,
+ ListPartitions,
+ CreatePartition,
+ DropPartition,
+}
+
+#[derive(Debug, Clone)]
+pub enum FakeCall {
+ CreateDatabase(String),
+ DropDatabase(String),
+ CreateTable(TablePath, TableDescriptor),
+ AlterTable(TablePath, AlterTableChanges),
+ DropTable(TablePath),
+ CreatePartition(TablePath, PartitionSpec),
+ DropPartition(TablePath, PartitionSpec),
}
-/// A backend whose clusters and catalog are fixed and whose failures are injected by the test.
-///
-/// The catalog is shared by every configured cluster: the tests that need cluster isolation are the
-/// pool's, not the protocol's.
pub struct FakeFlussBackend {
clusters: Vec<ClusterId>,
state: Mutex<FakeState>,
@@ -55,72 +81,90 @@ impl Default for FakeFlussBackend {
}
impl FakeFlussBackend {
- /// One cluster named `default` with an empty catalog.
pub fn new() -> Self {
Self::with_catalog(&[])
}
- /// One cluster named `default` over a catalog of databases and their tables, in any order.
pub fn with_catalog(databases: &[(&str, &[&str])]) -> Self {
- Self {
+ let backend = Self {
clusters: vec![cluster_id("default")],
- state: Mutex::new(FakeState {
- databases: databases
- .iter()
- .map(|(database, tables)| {
- (
- (*database).to_string(),
- tables.iter().map(|table| (*table).to_string()).collect(),
- )
- })
- .collect(),
- failure: None,
- latency: Duration::ZERO,
- }),
+ state: Mutex::default(),
+ };
+ for (database, tables) in databases {
+ backend.define_database(database);
+ for table in *tables {
+ backend.define_table(fixture_table(TablePath::new(*database, *table)));
+ }
}
+ backend
}
- /// Several clusters with an empty catalog, for the discovery tests.
pub fn with_clusters(ids: &[&str]) -> Self {
let mut clusters: Vec<ClusterId> = ids.iter().map(|id| cluster_id(id)).collect();
clusters.sort();
Self {
clusters,
- ..Self::with_catalog(&[])
+ ..Self::new()
+ }
+ }
+
+ pub fn define_database(&self, name: &str) {
+ self.state().databases.entry(name.to_string()).or_default();
+ }
+
+ pub fn define_table(&self, info: TableInfo) {
+ let mut state = self.state();
+ let tables = state
+ .databases
+ .entry(info.table_path.database().to_string())
+ .or_default();
+ match tables.entry(info.table_path.table().to_string()) {
+ Entry::Occupied(mut entry) => entry.get_mut().info = info,
+ Entry::Vacant(entry) => {
+ entry.insert(FakeTable {
+ info,
+ partitions: Vec::new(),
+ });
+ }
}
}
- /// Makes every catalog call fail, the way an unreachable or refusing cluster would.
- pub fn fail_with(&self, error: GatewayError) {
- self.state().failure = Some(error);
+ pub fn define_partition(&self, table: &TablePath, info: PartitionInfo) {
+ let mut state = self.state();
+ let entry = state
+ .databases
+ .get_mut(table.database())
+ .and_then(|tables| tables.get_mut(table.table()))
+ .expect("the fixture table is defined");
+ entry.partitions.push(info);
}
- /// Delays every catalog call, so a test can drive a request into its deadline.
- pub fn set_latency(&self, latency: Duration) {
- self.state().latency = latency;
+ pub fn fail_once(&self, operation: Operation, error: GatewayError) {
+ self.state().failures.insert(operation, error);
}
- /// Runs one fixture answer under the request's budget, like the production backend does.
- async fn answer<T>(
+ pub fn calls(&self) -> Vec<FakeCall> {
+ self.state().calls.clone()
+ }
+
+ fn call<T>(
&self,
ctx: &RequestContext,
- produce: impl FnOnce(&FakeState) -> GatewayResult<T>,
+ operation: Operation,
+ mutation: Option<FakeCall>,
+ answer: impl FnOnce(&FakeState) -> GatewayResult<T>,
) -> GatewayResult<T> {
if !self.has_cluster(ctx.cluster_id().as_str()) {
return Err(unknown_cluster(ctx.cluster_id().as_str()));
}
- let latency = self.state().latency;
- ctx.run(async move {
- if !latency.is_zero() {
- tokio::time::sleep(latency).await;
- }
- let state = self.state();
- match &state.failure {
- Some(failure) => Err(failure.clone()),
- None => produce(&state),
- }
- })
- .await
+ let mut state = self.state();
+ if let Some(call) = mutation {
+ state.calls.push(call);
+ }
+ if let Some(error) = state.failures.remove(&operation) {
+ return Err(error);
+ }
+ answer(&state)
}
fn state(&self) -> MutexGuard<'_, FakeState> {
@@ -132,6 +176,41 @@ fn cluster_id(id: &str) -> ClusterId {
ClusterId::try_from(id).expect("valid fixture cluster ID")
}
+fn fixture_table(table: TablePath) -> TableInfo {
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(fluss::metadata::BigIntType::new()))
+ .build()
+ .expect("the fixture schema is valid");
+ let descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .distributed_by(Some(1), Vec::new())
+ .build()
+ .expect("the fixture descriptor is valid");
+ TableInfo::of(table, 1, 1, descriptor, FIXTURE_TIME, FIXTURE_TIME)
+}
+
+fn database_of<'state>(
+ state: &'state FakeState,
+ database: &str,
+) -> GatewayResult<&'state BTreeMap<String, FakeTable>> {
+ state.databases.get(database).ok_or_else(|| {
+ GatewayError::not_found(format!("database `{database}` does not exist"))
+ .with_resource(Resource::Database)
+ })
+}
+
+fn table_of<'state>(
+ state: &'state FakeState,
+ table: &TablePath,
+) -> GatewayResult<&'state FakeTable> {
+ database_of(state, table.database())?
+ .get(table.table())
+ .ok_or_else(|| {
+ GatewayError::not_found(format!("table `{table}` does not exist"))
+ .with_resource(Resource::Table)
+ })
+}
+
#[async_trait]
impl FlussBackend for FakeFlussBackend {
fn clusters(&self) -> Vec<ClusterId> {
@@ -143,8 +222,27 @@ impl FlussBackend for FakeFlussBackend {
}
async fn list_databases(&self, ctx: &RequestContext) -> GatewayResult<Vec<String>> {
- self.answer(ctx, |state| Ok(state.databases.keys().cloned().collect()))
- .await
+ self.call(ctx, Operation::ListDatabases, None, |state| {
+ Ok(state.databases.keys().cloned().collect())
+ })
+ }
+
+ async fn create_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::CreateDatabase,
+ Some(FakeCall::CreateDatabase(database.to_string())),
+ |_| Ok(()),
+ )
+ }
+
+ async fn drop_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::DropDatabase,
+ Some(FakeCall::DropDatabase(database.to_string())),
+ |_| Ok(()),
+ )
}
async fn list_tables(
@@ -152,76 +250,93 @@ impl FlussBackend for FakeFlussBackend {
ctx: &RequestContext,
database: &str,
) -> GatewayResult<Vec<String>> {
- self.answer(ctx, |state| {
- state.databases.get(database).cloned().ok_or_else(|| {
- GatewayError::not_found(format!("database `{database}` does not exist"))
- .with_resource(Resource::Database)
- })
+ self.call(ctx, Operation::ListTables, None, |state| {
+ Ok(database_of(state, database)?.keys().cloned().collect())
})
- .await
}
-}
-#[cfg(test)]
-mod tests {
- use super::*;
- use crate::error::ErrorKind;
-
- #[tokio::test]
- async fn the_fixture_answers_its_catalog_until_a_failure_is_injected() {
- let backend = FakeFlussBackend::with_catalog(&[("sales", &["orders"]), ("ops", &[])]);
- let ctx = RequestContext::for_test("default", Duration::from_secs(5));
-
- assert_eq!(
- backend.list_databases(&ctx).await.unwrap(),
- ["ops", "sales"]
- );
- assert_eq!(
- backend.list_tables(&ctx, "sales").await.unwrap(),
- ["orders"]
- );
- assert_eq!(
- backend.list_tables(&ctx, "nope").await.unwrap_err().code(),
- "database_not_found"
- );
-
- // A cluster the fixture does not serve is the same 404 the production backend answers.
- let elsewhere = RequestContext::for_test("other", Duration::from_secs(5));
- assert_eq!(
- backend.list_databases(&elsewhere).await.unwrap_err().code(),
- "cluster_not_found"
- );
-
- backend.fail_with(GatewayError::unavailable("cluster is down"));
- assert_eq!(
- backend.list_databases(&ctx).await.unwrap_err().kind(),
- ErrorKind::Unavailable
- );
- }
-
- /// An injected delay is bounded by the request, not by the fixture.
- #[tokio::test]
- async fn a_slow_answer_runs_into_the_request_deadline() {
- let backend = FakeFlussBackend::new();
- backend.set_latency(Duration::from_secs(30));
- let ctx = RequestContext::for_test("default", Duration::from_millis(20));
-
- assert_eq!(
- backend.list_databases(&ctx).await.unwrap_err().kind(),
- ErrorKind::DeadlineExceeded
- );
- }
-
- #[test]
- fn several_fixture_clusters_are_discovered_in_order() {
- let backend = FakeFlussBackend::with_clusters(&["zeta", "alpha"]);
- assert_eq!(
- backend
- .clusters()
- .iter()
- .map(ClusterId::as_str)
- .collect::<Vec<_>>(),
- ["alpha", "zeta"]
- );
+ async fn describe_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<TableInfo> {
+ self.call(ctx, Operation::DescribeTable, None, |state| {
+ Ok(table_of(state, table)?.info.clone())
+ })
+ }
+
+ async fn create_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ descriptor: &TableDescriptor,
+ ) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::CreateTable,
+ Some(FakeCall::CreateTable(table.clone(), descriptor.clone())),
+ |_| Ok(()),
+ )
+ }
+
+ async fn alter_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ changes: AlterTableChanges,
+ ) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::AlterTable,
+ Some(FakeCall::AlterTable(table.clone(), changes)),
+ |_| Ok(()),
+ )
+ }
+
+ async fn drop_table(&self, ctx: &RequestContext, table: &TablePath) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::DropTable,
+ Some(FakeCall::DropTable(table.clone())),
+ |_| Ok(()),
+ )
+ }
+
+ async fn list_partitions(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<Vec<PartitionInfo>> {
+ self.call(ctx, Operation::ListPartitions, None, |state| {
+ Ok(table_of(state, table)?.partitions.clone())
+ })
+ }
+
+ async fn create_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::CreatePartition,
+ Some(FakeCall::CreatePartition(table.clone(), spec.clone())),
+ |_| Ok(()),
+ )
+ }
+
+ async fn drop_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()> {
+ self.call(
+ ctx,
+ Operation::DropPartition,
+ Some(FakeCall::DropPartition(table.clone(), spec.clone())),
+ |_| Ok(()),
+ )
}
}
diff --git a/fluss-gateway/src/backend/mod.rs b/fluss-gateway/src/backend/mod.rs
index 12f2bbd26..d0fecec0d 100644
--- a/fluss-gateway/src/backend/mod.rs
+++ b/fluss-gateway/src/backend/mod.rs
@@ -15,20 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! The gateway's HTTP-independent backend component.
-//!
-//! Protocol adapters reach Fluss only through [`FlussBackend`], which is their single dependency: the
-//! cluster a request names, the identity it acts as, and the connection that serves it are all resolved
-//! below this trait. That boundary is what lets the REST layer be tested without a cluster and lets a
-//! second protocol be added without reimplementing any of it.
-//!
-//! A backend **owns its connections**. Shutdown is a method on the concrete type rather than on this
-//! trait, so a protocol adapter cannot reach it.
-//!
-//! Every capability method is a complete request-response operation: the contract deliberately exposes
-//! **no** way to open a stream, scanner, cursor, or any other handle that would outlive the call,
-//! because the gateway keeps no request-spanning state. There is also no cluster-health method —
-//! reachability is not a queryable state, it is the outcome of the next request.
+//! HTTP-independent Fluss operations used by protocol adapters.
pub mod client;
pub mod connection;
@@ -43,33 +30,76 @@ use crate::backend::context::RequestContext;
use crate::backend::types::ClusterId;
use crate::error::{GatewayError, GatewayResult, Resource};
use async_trait::async_trait;
+use fluss::metadata::{
+ AlterTableChanges, PartitionInfo, PartitionSpec, TableDescriptor, TableInfo, TablePath,
+};
/// The backend capabilities the protocol adapters depend on.
///
-/// Implementations never return HTTP or JSON types: the adapter owns status mapping, and the backend
-/// owns the translation from native failures into [`GatewayError`].
-///
-/// The remaining FIP-49 capabilities — describe table, partitions, DDL, write, lookup — are appended
-/// here in the same shape, so carrying them changes neither the connection layer nor the protocol layer.
+/// Native metadata crosses this boundary unchanged; protocol adapters own wire shapes.
#[async_trait]
pub trait FlussBackend: Send + Sync + 'static {
- /// The clusters this gateway serves, in lexical ID order.
- ///
- /// A configuration echo, which is why it is synchronous, cannot fail, and takes no request.
+ /// The configured clusters, in lexical order.
fn clusters(&self) -> Vec<ClusterId>;
- /// Whether `id` names a cluster this gateway serves. Allocation-free, for per-request validation.
+ /// Whether `id` names a configured cluster.
fn has_cluster(&self, id: &str) -> bool;
async fn list_databases(&self, ctx: &RequestContext) -> GatewayResult<Vec<String>>;
+ async fn create_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()>;
+
+ /// Drops an empty database without cascading.
+ async fn drop_database(&self, ctx: &RequestContext, database: &str) -> GatewayResult<()>;
+
async fn list_tables(&self, ctx: &RequestContext, database: &str)
-> GatewayResult<Vec<String>>;
+
+ async fn describe_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<TableInfo>;
+
+ async fn create_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ descriptor: &TableDescriptor,
+ ) -> GatewayResult<()>;
+
+ /// Applies the change group in one native request.
+ async fn alter_table(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ changes: AlterTableChanges,
+ ) -> GatewayResult<()>;
+
+ async fn drop_table(&self, ctx: &RequestContext, table: &TablePath) -> GatewayResult<()>;
+
+ async fn list_partitions(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ ) -> GatewayResult<Vec<PartitionInfo>>;
+
+ async fn create_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()>;
+
+ async fn drop_partition(
+ &self,
+ ctx: &RequestContext,
+ table: &TablePath,
+ spec: &PartitionSpec,
+ ) -> GatewayResult<()>;
}
-/// The answer to a cluster this gateway does not serve.
-///
-/// A malformed ID and an unconfigured one are the same answer to a caller, so they share one error.
+/// Returns the error for an unconfigured cluster.
pub fn unknown_cluster(id: &str) -> GatewayError {
GatewayError::not_found(format!("unknown cluster `{id}`")).with_resource(Resource::Cluster)
}
diff --git a/fluss-gateway/src/config.rs b/fluss-gateway/src/config.rs
index 75b123b52..27363af87 100644
--- a/fluss-gateway/src/config.rs
+++ b/fluss-gateway/src/config.rs
@@ -1023,9 +1023,8 @@ impl GatewayConfig {
}
if cluster.identity_mode == IdentityMode::User {
- // Refused rather than silently downgraded to the shared service account: turning
- // "isolated per caller" into "everyone shares one identity" is a security problem, not a
- // degradation (FIP-49: configuring user mode fails at startup until Fluss ships act-as).
+ // Reject user mode until per-caller identities are supported; never fall back to
+ // the shared service account.
problems.push(format!(
"{} user is not supported yet: fluss-rust cannot send a SASL authorization ID, and \
client authentication is not implemented",
diff --git a/fluss-gateway/src/error.rs b/fluss-gateway/src/error.rs
index b9214853c..00b8ef5c9 100644
--- a/fluss-gateway/src/error.rs
+++ b/fluss-gateway/src/error.rs
@@ -26,10 +26,9 @@
//! and the OpenAPI `ErrorCode` schema is generated from it, so the published contract cannot drift from the
//! taxonomy.
//!
-//! FIP-49 error-model notes: the FIP's `database_not_empty` (409) condition is carried by
-//! [`ErrorKind::FailedPrecondition`], and its `*_not_found` / `*_already_exists` families are the
-//! [`ErrorKind::NotFound`] / [`ErrorKind::AlreadyExists`] kinds qualified by a [`Resource`], keeping one
-//! stable code per condition.
+//! `database_not_empty` (409) uses [`ErrorKind::FailedPrecondition`]. The `*_not_found` and
+//! `*_already_exists` families use [`ErrorKind::NotFound`] and [`ErrorKind::AlreadyExists`]
+//! qualified by a [`Resource`].
use serde::Serialize;
use std::any::Any;
@@ -80,8 +79,7 @@ macro_rules! error_kinds {
}
}
- /// Whether a response carrying this kind advertises `Retry-After`, which FIP-49 requires of
- /// every 429 and which the gateway also sends for a transient backend outage.
+ /// Whether responses advertise `Retry-After`: all 429s and transient backend outages.
pub fn retry_after(self) -> bool {
match self {
$( Self::$variant => $retry_after, )+
@@ -129,7 +127,7 @@ error_kinds! {
DeadlineExceeded => 504, "timeout", retry_after: false;
}
-/// A resource a failure can name, selecting the FIP-49 resource-specific wire code.
+/// A resource a failure can name, selecting a resource-specific wire code.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Resource {
Cluster,
@@ -138,10 +136,7 @@ pub enum Resource {
Partition,
}
-/// The FIP-49 resource-specific codes.
-///
-/// Only the combinations the FIP names appear here; any other kind/resource pair keeps the kind's generic
-/// code, so the gateway never invents a code for a condition the contract does not describe.
+/// Resource-specific codes; unlisted pairs keep the kind's generic code.
const RESOURCE_CODES: &[(ErrorKind, Resource, &str)] = &[
(ErrorKind::NotFound, Resource::Cluster, "cluster_not_found"),
(
@@ -175,7 +170,7 @@ const RESOURCE_CODES: &[(ErrorKind, Resource, &str)] = &[
Resource::Partition,
"partition_already_exists",
),
- // The one precondition the FIP names: dropping a non-empty database.
+ // Dropping a non-empty database has its own precondition code.
(
ErrorKind::FailedPrecondition,
Resource::Database,
@@ -234,8 +229,7 @@ pub type GatewayResult<T> = Result<T, GatewayError>;
pub struct GatewayError {
kind: ErrorKind,
message: String,
- /// Selects the resource-specific wire code. Never serialized: the resource is already named in the
- /// message, and FIP-49's envelope carries no structured details.
+ /// Selects a resource-specific wire code without adding a field to the error envelope.
resource: Option<Resource>,
}
@@ -310,8 +304,7 @@ impl GatewayError {
Self::new(ErrorKind::Unavailable, message)
}
- /// A Fluss backend failure the gateway cannot classify further. Answered with HTTP 500 and the
- /// FIP-49 `backend` code, so callers can tell it from a gateway-internal failure.
+ /// An unclassified Fluss failure, returned as HTTP 500 with the `backend` code.
pub fn backend(message: impl Into<String>) -> Self {
Self::new(ErrorKind::Backend, message)
}
@@ -333,8 +326,7 @@ impl GatewayError {
/// Stable code carried in the error envelope.
///
- /// An error that names a resource answers the resource-specific code that FIP-49 defines for the pair;
- /// any other error keeps its kind's generic code.
+ /// Unlisted kind/resource pairs fall back to the kind's generic code.
pub fn code(&self) -> &'static str {
self.resource
.and_then(|resource| resource_code(self.kind, resource))
@@ -411,10 +403,9 @@ impl ErrorEnvelope {
mod tests {
use super::*;
- /// The mappings FIP-49 pins down. The table generates every accessor, so this guards the table's own
- /// rows against an accidental edit.
+ /// Keeps published status and error-code mappings stable.
#[test]
- fn the_fip_mappings_are_frozen() {
+ fn http_status_and_error_code_mappings_are_stable() {
for (kind, status, code) in [
(ErrorKind::InvalidArgument, 400, "invalid_argument"),
(ErrorKind::Unauthenticated, 401, "unauthenticated"),
@@ -479,8 +470,7 @@ mod tests {
);
}
- /// The FIP-49 vocabulary: an error naming a resource answers the resource-specific code; one without a
- /// resource keeps its kind's generic code.
+ /// An error naming a resource uses its specific code, falling back to the generic code otherwise.
#[test]
fn resource_context_specialises_the_wire_code() {
let cases: [(GatewayError, Resource, &str); 5] = [
@@ -527,7 +517,7 @@ mod tests {
.code(),
"failed_precondition"
);
- // Backend and internal stay distinguishable (FIP-49 `backend` / `internal`).
+ // Backend and internal failures stay distinguishable.
assert_eq!(GatewayError::backend("x").code(), "backend");
assert_eq!(GatewayError::internal("x").code(), "internal");
}
diff --git a/fluss-gateway/src/lifecycle.rs b/fluss-gateway/src/lifecycle.rs
index 48c7512ef..59677a61c 100644
--- a/fluss-gateway/src/lifecycle.rs
+++ b/fluss-gateway/src/lifecycle.rs
@@ -339,7 +339,7 @@ async fn start_internal(
shutdown.clone(),
),
);
- // Samples the FIP-49 process_* and tokio_* gauges alongside the exporter they feed.
+ // Sample process and Tokio metrics alongside their exporter.
let sampler_shutdown = shutdown.clone();
spawn_named(&mut tasks, "runtime metrics sampler", async move {
let mut interval = tokio::time::interval(Duration::from_secs(10));
diff --git a/fluss-gateway/src/observability.rs b/fluss-gateway/src/observability.rs
index 9b62dc500..e1224ef20 100644
--- a/fluss-gateway/src/observability.rs
+++ b/fluss-gateway/src/observability.rs
@@ -17,11 +17,8 @@
//! Process logging and the gateway metric inventory.
//!
-//! [`METRIC_DEFINITIONS`] tracks the FIP-49 metric table: every family declared here is one the
-//! specification names, with its kind, unit, description, and label set. Families the FIP defines for
-//! capabilities that do not exist yet — the backend write and connection-pool families, the re-exported
-//! client backpressure families — arrive with those capabilities. The inventory never grows a family of its
-//! own; the tests below enforce that against the table.
+//! [`METRIC_DEFINITIONS`] lists implemented metric families with their kinds, units, descriptions,
+//! and label sets. Families for future capabilities are added alongside their implementations.
//!
//! Labels describe an operation or a bounded outcome. `cluster`, sourced from validated configuration, is the
//! only resource-name label the gateway itself emits.
@@ -59,8 +56,7 @@ static METRICS_HANDLE: OnceLock<PrometheusHandle> = OnceLock::new();
/// configured deadline.
///
/// Without explicit buckets `metrics-exporter-prometheus` renders every histogram as a summary with
-/// pre-computed quantiles, which cannot be aggregated across gateway instances — and FIP-49 asks for a
-/// histogram.
+/// pre-computed quantiles, which cannot be aggregated across gateway instances.
const DURATION_BUCKETS: &[f64] = &[
0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0,
];
@@ -84,16 +80,13 @@ pub struct MetricDefinition {
pub labels: &'static [&'static str],
}
-/// The gateway's slice of the FIP-49 metric table.
-///
-/// Adding an emission site means adding its family here first, and a family may only be added if the FIP
-/// declares it.
+/// Implemented metric families and their labels.
pub const METRIC_DEFINITIONS: &[MetricDefinition] = &[
metric(
"fluss_gateway_rest_requests_total",
MetricKind::Counter,
None,
- "Completed REST requests. `operation` is the matched route template (FIP-49), `code` \
+ "Completed REST requests. `operation` is the matched route template, `code` \
the HTTP status, and `cluster` the bounded configured-cluster label (`none` for \
cluster-free routes, `unknown` for unconfigured IDs).",
&["cluster", "method", "operation", "code"],
@@ -105,7 +98,7 @@ pub const METRIC_DEFINITIONS: &[MetricDefinition] = &[
"REST request duration.",
&["cluster", "method", "operation"],
),
- // FIP-49 connection families, reported per configured cluster by the backend runtime. The per-user
+ // Connection families, reported per configured cluster by the backend runtime. The per-user
// act-as pool of the user identity mode reports into the same families when it arrives: they count the
// Fluss connections the gateway holds for a cluster, whoever they are opened for.
metric(
@@ -129,7 +122,7 @@ pub const METRIC_DEFINITIONS: &[MetricDefinition] = &[
"Fluss connections released for a configured cluster. `reason` is `idle` or `shutdown`.",
&["cluster", "reason"],
),
- // FIP-49 process and Tokio runtime families, sampled periodically by the runtime sampler.
+ // Process and Tokio runtime families, sampled periodically by the runtime sampler.
metric(
"process_cpu_seconds_total",
MetricKind::Counter,
@@ -172,7 +165,7 @@ pub const METRIC_DEFINITIONS: &[MetricDefinition] = &[
"Tasks waiting in the Tokio injection queue.",
&[],
),
- // FIP-49 also lists `tokio_worker_busy_seconds_total`; it needs the `tokio_unstable` runtime
+ // `tokio_worker_busy_seconds_total` needs the `tokio_unstable` runtime
// metrics and is added once the build enables them.
];
@@ -222,10 +215,8 @@ pub fn init_metrics(enabled: bool) -> Result<(), String> {
/// Records one completed REST request against the matched route template, never the raw URI.
///
-/// `operation` and `code` are the FIP-49 label names: the operation is the matched route
-/// template, the code the HTTP status. `cluster` is already bounded by the caller: a configured
-/// cluster ID, `unknown` for a request that named an unconfigured one, or `none` for routes
-/// without a cluster segment.
+/// `operation` is the matched route template; `code` is the HTTP status. The caller bounds
+/// `cluster` to a configured ID, `unknown` for unconfigured IDs, or `none` for cluster-free routes.
pub fn http_request(cluster: &str, method: &str, operation: &str, code: u16, duration: Duration) {
metrics::counter!(
"fluss_gateway_rest_requests_total",
@@ -279,7 +270,7 @@ pub fn connections_active(cluster: &str, active: usize) {
.set(active as f64);
}
-/// Samples the FIP-49 process and Tokio runtime gauges once.
+/// Samples process and Tokio runtime gauges once.
///
/// Called periodically by the lifecycle's runtime sampler; each source that a platform cannot
/// provide is skipped rather than published as zero.
@@ -359,8 +350,7 @@ fn process_open_fds() -> Option<f64> {
None
}
-/// The soft file descriptor limit, from `getrlimit(2)`. FIP-49 pairs it with `process_open_fds` so the count
-/// can be read against its ceiling.
+/// The soft file descriptor limit from `getrlimit(2)`, paired with `process_open_fds`.
#[cfg(unix)]
fn process_max_fds() -> Option<f64> {
let mut limit = std::mem::MaybeUninit::<libc::rlimit>::zeroed();
@@ -433,8 +423,8 @@ mod tests {
assert_eq!(parse_level("module=debug"), LevelFilter::Info);
}
- /// The FIP-49 metric table, verbatim. The gateway may implement a subset of it and nothing outside it.
- const FIP_49_FAMILIES: &[&str] = &[
+ /// Allowed metric families, including those reserved for future capabilities.
+ const ALLOWED_METRIC_FAMILIES: &[&str] = &[
"fluss_gateway_rest_requests_total",
"fluss_gateway_rest_request_duration_seconds",
"fluss_gateway_backend_write_rows_total",
@@ -453,14 +443,12 @@ mod tests {
"tokio_worker_busy_seconds_total",
];
- /// Every declared family is one FIP-49 names — the inventory tracks the specification instead of
- /// growing families of its own.
#[test]
- fn the_inventory_declares_nothing_the_fip_does_not() {
+ fn the_inventory_uses_only_allowed_families() {
for definition in METRIC_DEFINITIONS {
assert!(
- FIP_49_FAMILIES.contains(&definition.name),
- "{} is not in the FIP-49 metric table",
+ ALLOWED_METRIC_FAMILIES.contains(&definition.name),
+ "{} is not an allowed metric family",
definition.name
);
}
@@ -492,7 +480,7 @@ mod tests {
assert_eq!(names.len(), total, "duplicate metric family declared");
}
- /// Gateway-owned families keep their label sets bounded. The FIP's re-exported
+ /// Gateway-owned families keep their label sets bounded. Re-exported
/// `fluss_client_writer_kv_backpressure_*` families do carry `database` / `table`; they come from the
/// client recorder, not from here, and this rule is relaxed for them when they arrive.
#[test]
diff --git a/fluss-gateway/src/protocol/rest/datatype.rs b/fluss-gateway/src/protocol/rest/datatype.rs
new file mode 100644
index 000000000..726c2a9f7
--- /dev/null
+++ b/fluss-gateway/src/protocol/rest/datatype.rs
@@ -0,0 +1,671 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! REST data types using Fluss's persisted structured vocabulary.
+
+use crate::error::{GatewayError, GatewayResult};
+use fluss::metadata::{
+ ArrayType, BigIntType, BinaryType, BooleanType, BytesType, CharType, DataField, DataType,
+ DateType, DecimalType, DoubleType, FloatType, IntType, MapType, RowType, SmallIntType,
+ StringType, TimeType, TimestampLTzType, TimestampType, TinyIntType,
+};
+use serde::{Deserialize, Deserializer, Serialize};
+use utoipa::openapi::{RefOr, Schema};
+use utoipa::{PartialSchema, ToSchema};
+
+const MAX_TYPE_NESTING: usize = 64;
+
+/// The exact recursive Fluss type as it appears on the wire.
+#[derive(Debug, Deserialize, Serialize, ToSchema)]
+#[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE", deny_unknown_fields)]
+#[schema(no_recursion)]
+pub enum WireDataType {
+ Boolean {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ #[serde(rename = "TINYINT")]
+ TinyInt {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ #[serde(rename = "SMALLINT")]
+ SmallInt {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ #[serde(rename = "INTEGER")]
+ Int {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ #[serde(rename = "BIGINT")]
+ BigInt {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ Float {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ Double {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ Char {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ length: u32,
+ },
+ String {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ Decimal {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ #[schema(minimum = 1, maximum = 38)]
+ precision: u32,
+ #[schema(maximum = 38)]
+ scale: u32,
+ },
+ Date {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ #[serde(rename = "TIME_WITHOUT_TIME_ZONE")]
+ Time {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ #[schema(maximum = 9)]
+ precision: u32,
+ },
+ #[serde(rename = "TIMESTAMP_WITHOUT_TIME_ZONE")]
+ Timestamp {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ #[schema(maximum = 9)]
+ precision: u32,
+ },
+ #[serde(rename = "TIMESTAMP_WITH_LOCAL_TIME_ZONE")]
+ TimestampLtz {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ #[schema(maximum = 9)]
+ precision: u32,
+ },
+ Bytes {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ },
+ Binary {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ length: usize,
+ },
+ Array {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ element_type: Box<WireDataType>,
+ },
+ Map {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ key_type: Box<WireDataType>,
+ value_type: Box<WireDataType>,
+ },
+ Row {
+ #[serde(default = "default_nullable", skip_serializing_if = "is_true")]
+ nullable: bool,
+ fields: Vec<WireRowField>,
+ },
+}
+
+fn default_nullable() -> bool {
+ true
+}
+
+fn is_true(value: &bool) -> bool {
+ *value
+}
+
+/// One `ROW` field. Server-assigned field IDs are not exposed.
+#[derive(Debug, Deserialize, Serialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct WireRowField {
+ pub name: String,
+ #[schema(no_recursion)]
+ pub field_type: WireDataType,
+ #[serde(
+ default,
+ deserialize_with = "optional_non_null",
+ skip_serializing_if = "Option::is_none"
+ )]
+ #[schema(nullable = false)]
+ pub description: Option<String>,
+}
+
+/// Missing fields default to `None`; present fields must contain a non-null value.
+fn optional_non_null<'de, D, T>(deserializer: D) -> Result<Option<T>, D::Error>
+where
+ D: Deserializer<'de>,
+ T: Deserialize<'de>,
+{
+ T::deserialize(deserializer).map(Some)
+}
+
+impl WireDataType {
+ /// Replaces only the top-level nullability.
+ pub fn with_root_nullable(mut self, root: bool) -> Self {
+ let nullable = match &mut self {
+ Self::Boolean { nullable }
+ | Self::TinyInt { nullable }
+ | Self::SmallInt { nullable }
+ | Self::Int { nullable }
+ | Self::BigInt { nullable }
+ | Self::Float { nullable }
+ | Self::Double { nullable }
+ | Self::String { nullable }
+ | Self::Date { nullable }
+ | Self::Bytes { nullable }
+ | Self::Char { nullable, .. }
+ | Self::Decimal { nullable, .. }
+ | Self::Time { nullable, .. }
+ | Self::Timestamp { nullable, .. }
+ | Self::TimestampLtz { nullable, .. }
+ | Self::Binary { nullable, .. }
+ | Self::Array { nullable, .. }
+ | Self::Map { nullable, .. }
+ | Self::Row { nullable, .. } => nullable,
+ };
+ *nullable = root;
+ self
+ }
+}
+
+impl From<&DataType> for WireDataType {
+ /// Renders the native type, keeping nullability at every level.
+ fn from(data_type: &DataType) -> Self {
+ let nullable = data_type.is_nullable();
+ match data_type {
+ DataType::Boolean(_) => Self::Boolean { nullable },
+ DataType::TinyInt(_) => Self::TinyInt { nullable },
+ DataType::SmallInt(_) => Self::SmallInt { nullable },
+ DataType::Int(_) => Self::Int { nullable },
+ DataType::BigInt(_) => Self::BigInt { nullable },
+ DataType::Float(_) => Self::Float { nullable },
+ DataType::Double(_) => Self::Double { nullable },
+ DataType::Char(value) => Self::Char {
+ nullable,
+ length: value.length(),
+ },
+ DataType::String(_) => Self::String { nullable },
+ DataType::Decimal(value) => Self::Decimal {
+ nullable,
+ precision: value.precision(),
+ scale: value.scale(),
+ },
+ DataType::Date(_) => Self::Date { nullable },
+ DataType::Time(value) => Self::Time {
+ nullable,
+ precision: value.precision(),
+ },
+ DataType::Timestamp(value) => Self::Timestamp {
+ nullable,
+ precision: value.precision(),
+ },
+ DataType::TimestampLTz(value) => Self::TimestampLtz {
+ nullable,
+ precision: value.precision(),
+ },
+ DataType::Bytes(_) => Self::Bytes { nullable },
+ DataType::Binary(value) => Self::Binary {
+ nullable,
+ length: value.length(),
+ },
+ DataType::Array(value) => Self::Array {
+ nullable,
+ element_type: Box::new(Self::from(value.get_element_type())),
+ },
+ DataType::Map(value) => Self::Map {
+ nullable,
+ key_type: Box::new(Self::from(value.key_type())),
+ value_type: Box::new(Self::from(value.value_type())),
+ },
+ DataType::Row(value) => Self::Row {
+ nullable,
+ fields: value.fields().iter().map(WireRowField::from).collect(),
+ },
+ }
+ }
+}
+
+impl From<&DataField> for WireRowField {
+ /// Drops the field ID, which the server assigns and the API never exposes.
+ fn from(field: &DataField) -> Self {
+ Self {
+ name: field.name.clone(),
+ field_type: WireDataType::from(&field.data_type),
+ description: field.description.clone(),
+ }
+ }
+}
+
+impl TryFrom<WireDataType> for DataType {
+ type Error = GatewayError;
+
+ /// Builds the native type using its constructors.
+ fn try_from(data_type: WireDataType) -> Result<Self, Self::Error> {
+ to_native(data_type, 0)
+ }
+}
+
+/// Builds the native tree within the gateway's nesting limit.
+fn to_native(data_type: WireDataType, depth: usize) -> GatewayResult<DataType> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::invalid_argument(format!(
+ "the data type nests deeper than {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ // TODO: Delegate length and ROW field validation once fluss-rs supports it.
+ let converted = match data_type {
+ WireDataType::Boolean { nullable } => {
+ DataType::Boolean(BooleanType::with_nullable(nullable))
+ }
+ WireDataType::TinyInt { nullable } => {
+ DataType::TinyInt(TinyIntType::with_nullable(nullable))
+ }
+ WireDataType::SmallInt { nullable } => {
+ DataType::SmallInt(SmallIntType::with_nullable(nullable))
+ }
+ WireDataType::Int { nullable } => DataType::Int(IntType::with_nullable(nullable)),
+ WireDataType::BigInt { nullable } => DataType::BigInt(BigIntType::with_nullable(nullable)),
+ WireDataType::Float { nullable } => DataType::Float(FloatType::with_nullable(nullable)),
+ WireDataType::Double { nullable } => DataType::Double(DoubleType::with_nullable(nullable)),
+ WireDataType::Char { nullable, length } => {
+ DataType::Char(CharType::with_nullable(length, nullable))
+ }
+ WireDataType::String { nullable } => DataType::String(StringType::with_nullable(nullable)),
+ WireDataType::Decimal {
+ nullable,
+ precision,
+ scale,
+ } => DataType::Decimal(
+ DecimalType::with_nullable(nullable, precision, scale).map_err(invalid_type)?,
+ ),
+ WireDataType::Date { nullable } => DataType::Date(DateType::with_nullable(nullable)),
+ WireDataType::Time {
+ nullable,
+ precision,
+ } => DataType::Time(TimeType::with_nullable(nullable, precision).map_err(invalid_type)?),
+ WireDataType::Timestamp {
+ nullable,
+ precision,
+ } => DataType::Timestamp(
+ TimestampType::with_nullable(nullable, precision).map_err(invalid_type)?,
+ ),
+ WireDataType::TimestampLtz {
+ nullable,
+ precision,
+ } => DataType::TimestampLTz(
+ TimestampLTzType::with_nullable(nullable, precision).map_err(invalid_type)?,
+ ),
+ WireDataType::Bytes { nullable } => DataType::Bytes(BytesType::with_nullable(nullable)),
+ WireDataType::Binary { nullable, length } => {
+ DataType::Binary(BinaryType::with_nullable(nullable, length))
+ }
+ WireDataType::Array {
+ nullable,
+ element_type,
+ } => DataType::Array(ArrayType::with_nullable(
+ nullable,
+ to_native(*element_type, depth + 1)?,
+ )),
+ WireDataType::Map {
+ nullable,
+ key_type,
+ value_type,
+ } => DataType::Map(MapType::with_nullable(
+ nullable,
+ to_native(*key_type, depth + 1)?,
+ to_native(*value_type, depth + 1)?,
+ )),
+ WireDataType::Row { nullable, fields } => DataType::Row(RowType::with_nullable(
+ nullable,
+ fields
+ .into_iter()
+ .map(|field| {
+ Ok(DataField::new(
+ field.name,
+ to_native(field.field_type, depth + 1)?,
+ field.description,
+ ))
+ })
+ .collect::<GatewayResult<Vec<_>>>()?,
+ )),
+ };
+ Ok(converted)
+}
+
+/// A type parameter the native constructor refused, which came from a caller's body.
+fn invalid_type(error: fluss::error::Error) -> GatewayError {
+ GatewayError::invalid_argument(format!("invalid data type: {error}"))
+}
+
+/// A column type without top-level `nullable`; nested nullability is unchanged.
+#[derive(Debug)]
+pub struct ColumnDataType(pub WireDataType);
+
+impl PartialSchema for ColumnDataType {
+ fn schema() -> RefOr<Schema> {
+ let mut schema = WireDataType::schema();
+ if let RefOr::T(Schema::OneOf(one_of)) = &mut schema {
+ one_of.description = Some(
+ "A Fluss column type; top-level nullability is declared by the column.".to_string(),
+ );
+ for item in &mut one_of.items {
+ if let RefOr::T(Schema::Object(object)) = item {
+ object.properties.remove("nullable");
+ }
+ }
+ }
+ schema
+ }
+}
+
+impl ToSchema for ColumnDataType {}
+
+impl<'de> Deserialize<'de> for ColumnDataType {
+ fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+ let mut value = serde_json::Value::deserialize(deserializer)?;
+ if value.get("nullable").is_some() {
+ return Err(serde::de::Error::custom(
+ "a column's nullability is expressed by the column-level `nullable` field; \
+ `nullable` at the top level of `data_type` is not allowed",
+ ));
+ }
+ normalize_type_names(&mut value);
+ serde_json::from_value(value)
+ .map(ColumnDataType)
+ .map_err(serde::de::Error::custom)
+ }
+}
+
+/// Canonicalizes only type tags, including nested types; field names and descriptions stay intact.
+fn normalize_type_names(value: &mut serde_json::Value) {
+ match value {
+ serde_json::Value::Object(object) => {
+ if let Some(serde_json::Value::String(name)) = object.get_mut("type") {
+ name.make_ascii_uppercase();
+ }
+ for value in object.values_mut() {
+ normalize_type_names(value);
+ }
+ }
+ serde_json::Value::Array(array) => {
+ for value in array {
+ normalize_type_names(value);
+ }
+ }
+ _ => {}
+ }
+}
+
+impl Serialize for ColumnDataType {
+ fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+ let mut value = serde_json::to_value(&self.0).map_err(serde::ser::Error::custom)?;
+ if let Some(object) = value.as_object_mut() {
+ object.remove("nullable");
+ }
+ value.serialize(serializer)
+ }
+}
+
+impl From<&DataType> for ColumnDataType {
+ fn from(data_type: &DataType) -> Self {
+ Self(WireDataType::from(data_type))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::error::ErrorKind;
+ use serde_json::json;
+
+ fn parse(value: serde_json::Value) -> Result<DataType, String> {
+ let wire: WireDataType =
+ serde_json::from_value(value).map_err(|error| error.to_string())?;
+ DataType::try_from(wire).map_err(|error| error.message().to_string())
+ }
+
+ fn render(data_type: &DataType) -> serde_json::Value {
+ serde_json::to_value(WireDataType::from(data_type)).expect("the type renders")
+ }
+
+ #[test]
+ fn every_type_round_trips_through_the_wire_shape() {
+ let bodies = [
+ json!({"type": "BOOLEAN"}),
+ json!({"type": "TINYINT", "nullable": false}),
+ json!({"type": "SMALLINT"}),
+ json!({"type": "INTEGER", "nullable": false}),
+ json!({"type": "BIGINT"}),
+ json!({"type": "FLOAT", "nullable": false}),
+ json!({"type": "DOUBLE"}),
+ json!({"type": "CHAR", "nullable": false, "length": 3}),
+ json!({"type": "STRING"}),
+ json!({"type": "DECIMAL", "nullable": false, "precision": 18, "scale": 2}),
+ json!({"type": "DATE"}),
+ json!({"type": "TIME_WITHOUT_TIME_ZONE", "nullable": false, "precision": 9}),
+ json!({"type": "TIMESTAMP_WITHOUT_TIME_ZONE", "precision": 6}),
+ json!({"type": "TIMESTAMP_WITH_LOCAL_TIME_ZONE", "nullable": false, "precision": 9}),
+ json!({"type": "BYTES"}),
+ json!({"type": "BINARY", "nullable": false, "length": 16}),
+ json!({"type": "ARRAY", "element_type": {"type": "STRING", "nullable": false}}),
+ json!({"type": "MAP", "nullable": false, "key_type": {"type": "STRING", "nullable": false}, "value_type": {"type": "INTEGER"}}),
+ json!({"type": "ROW", "fields": [
+ {"name": "inner", "field_type": {"type": "INTEGER", "nullable": false}, "description": "a nested field"},
+ {"name": "plain", "field_type": {"type": "STRING"}}
+ ]}),
+ ];
+
+ for body in bodies {
+ let native = parse(body.clone()).expect("the body parses");
+ assert_eq!(render(&native), body, "{body}");
+ for nullable in [true, false] {
+ let wire = WireDataType::from(&native).with_root_nullable(nullable);
+ let mut expected = body.clone();
+ if nullable {
+ expected.as_object_mut().unwrap().remove("nullable");
+ } else {
+ expected["nullable"] = json!(false);
+ }
+ assert_eq!(serde_json::to_value(wire).unwrap(), expected, "{body}");
+ }
+ }
+ }
+
+ #[test]
+ fn type_names_are_case_insensitive_at_the_rest_boundary() {
+ for (body, expected) in [
+ (json!({"type": "bigint"}), json!({"type": "BIGINT"})),
+ (
+ json!({"type": "aRrAy", "element_type": {
+ "type": "mAp", "key_type": {"type": "sTrIng"}, "value_type": {
+ "type": "rOw", "fields": [{
+ "name": "mixedCase", "description": "Keep this case",
+ "field_type": {"type": "iNtEgEr", "nullable": false}
+ }]
+ }
+ }}),
+ json!({"type": "ARRAY", "element_type": {
+ "type": "MAP", "key_type": {"type": "STRING"}, "value_type": {
+ "type": "ROW", "fields": [{
+ "name": "mixedCase", "description": "Keep this case",
+ "field_type": {"type": "INTEGER", "nullable": false}
+ }]
+ }
+ }}),
+ ),
+ ] {
+ let column: ColumnDataType = serde_json::from_value(body).expect("the type parses");
+ assert_eq!(serde_json::to_value(column).unwrap(), expected);
+ }
+ }
+
+ #[test]
+ fn native_nullability_defaults_are_preserved() {
+ let native = parse(json!({"type": "ARRAY", "element_type": {"type": "INTEGER"}}))
+ .expect("the type parses");
+ assert_eq!(
+ render(&native),
+ json!({"type": "ARRAY", "element_type": {"type": "INTEGER"}})
+ );
+
+ let native = parse(json!({
+ "type": "MAP",
+ "key_type": {"type": "STRING"},
+ "value_type": {"type": "INTEGER"}
+ }))
+ .expect("the type parses");
+ assert_eq!(
+ render(&native),
+ json!({
+ "type": "MAP",
+ "key_type": {"type": "STRING", "nullable": false},
+ "value_type": {"type": "INTEGER"}
+ })
+ );
+ }
+
+ #[test]
+ fn a_type_that_is_not_exactly_described_is_refused() {
+ for body in [
+ json!({}),
+ json!({"type": "NUMBER"}),
+ json!({"type": "INT"}),
+ json!({"type": "DECIMAL", "precision": 10}),
+ json!({"type": "CHAR"}),
+ json!({"type": "BIGINT", "precision": 3}),
+ json!({"type": "BIGINT", "precision": null}),
+ json!({"type": "DECIMAL", "precision": 10, "scale": 2, "length": 4}),
+ json!({"type": "ARRAY"}),
+ json!({"type": "MAP", "key_type": {"type": "STRING"}}),
+ json!({"type": "ROW"}),
+ json!({"type": "STRING", "unexpected": 1}),
+ json!({"type": "CHAR", "length": null}),
+ json!({"type": "CHAR", "length": u64::from(u32::MAX) + 1}),
+ json!({"type": "BINARY", "length": -1}),
+ json!({"type": "DECIMAL", "precision": null, "scale": 0}),
+ json!({"type": "DECIMAL", "precision": 10, "scale": null}),
+ json!({"type": "ARRAY", "element_type": null}),
+ json!({"type": "MAP", "key_type": null, "value_type": {"type": "STRING"}}),
+ json!({"type": "MAP", "key_type": {"type": "STRING"}, "value_type": null}),
+ json!({"type": "ROW", "fields": null}),
+ json!({"type": "ARRAY", "element_type": {"type": "STRING", "nullable": null}}),
+ json!({"type": "ROW", "fields": [{"name": "id", "field_type": {"type": "INTEGER"}, "description": null}]}),
+ json!({"type": "ROW", "fields": [{"name": "id", "field_type": {"type": "INTEGER"}, "field_id": 1}]}),
+ ] {
+ assert!(parse(body.clone()).is_err(), "the body is refused: {body}");
+ }
+ }
+
+ #[test]
+ fn native_constructors_and_nesting_limit_define_validation() {
+ for body in [
+ json!({"type": "DECIMAL", "precision": 0, "scale": 0}),
+ json!({"type": "DECIMAL", "precision": 2, "scale": 3}),
+ json!({"type": "TIME_WITHOUT_TIME_ZONE", "precision": 10}),
+ json!({"type": "TIMESTAMP_WITHOUT_TIME_ZONE", "precision": 10}),
+ json!({"type": "TIMESTAMP_WITH_LOCAL_TIME_ZONE", "precision": 10}),
+ ] {
+ let wire: WireDataType =
+ serde_json::from_value(body.clone()).expect("the shape parses");
+ assert_eq!(
+ DataType::try_from(wire)
+ .expect_err("the value is refused")
+ .kind(),
+ ErrorKind::InvalidArgument,
+ "{body}"
+ );
+ }
+
+ for length in [0, 1, u32::MAX] {
+ for (type_name, native) in [
+ ("CHAR", DataType::Char(CharType::new(length))),
+ ("BINARY", DataType::Binary(BinaryType::new(length as usize))),
+ ] {
+ let body = json!({"type": type_name, "length": length});
+ assert_eq!(parse(body.clone()).unwrap(), native);
+ assert_eq!(render(&native), body);
+ }
+ }
+
+ let fields = ["", "a\nb", "id", "id"]
+ .into_iter()
+ .map(|name| DataField::new(name, DataType::Int(IntType::new()), None))
+ .collect();
+ let native = DataType::Row(RowType::new(fields));
+ assert_eq!(parse(render(&native)).unwrap(), native);
+
+ for depth in [MAX_TYPE_NESTING, MAX_TYPE_NESTING + 1] {
+ let mut wire = WireDataType::Int { nullable: true };
+ for _ in 0..depth {
+ wire = WireDataType::Array {
+ nullable: true,
+ element_type: Box::new(wire),
+ };
+ }
+ let result = DataType::try_from(wire);
+ if depth == MAX_TYPE_NESTING {
+ assert!(result.is_ok(), "the maximum nesting is accepted");
+ } else {
+ let error = result.expect_err("excessive nesting is refused");
+ assert_eq!(error.kind(), ErrorKind::InvalidArgument);
+ assert!(error.message().contains("nests deeper"));
+ }
+ }
+ }
+
+ #[test]
+ fn a_column_type_neither_accepts_nor_emits_a_top_level_nullability() {
+ for nullable in [json!(true), json!(false), json!(null)] {
+ let error = serde_json::from_value::<ColumnDataType>(
+ json!({"type": "INTEGER", "nullable": nullable}),
+ )
+ .expect_err("a top-level nullability is refused");
+ assert!(error.to_string().contains("column-level `nullable`"));
+ }
+
+ let column =
+ ColumnDataType::from(&parse(json!({"type": "INTEGER", "nullable": false})).unwrap());
+ assert_eq!(
+ serde_json::to_value(&column).expect("the column type renders"),
+ json!({"type": "INTEGER"})
+ );
+
+ let nested = ColumnDataType::from(
+ &parse(json!({"type": "ARRAY", "nullable": false, "element_type": {"type": "INTEGER", "nullable": false}}))
+ .unwrap(),
+ );
+ assert_eq!(
+ serde_json::to_value(&nested).expect("the column type renders"),
+ json!({"type": "ARRAY", "element_type": {"type": "INTEGER", "nullable": false}})
+ );
+ }
+}
diff --git a/fluss-gateway/src/protocol/rest/ddl.rs b/fluss-gateway/src/protocol/rest/ddl.rs
new file mode 100644
index 000000000..582323c1d
--- /dev/null
+++ b/fluss-gateway/src/protocol/rest/ddl.rs
@@ -0,0 +1,1426 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Catalog mutation endpoints.
+
+use crate::error::{ErrorEnvelope, GatewayError, GatewayResult};
+use crate::protocol::rest::datatype::ColumnDataType;
+use crate::protocol::rest::metadata::{DatabaseResponse, PartitionResponse};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, parse_json_body, reject_query_parameters, request_id,
+ resolve_cluster,
+};
+use axum::extract::{Path, Request, State};
+use axum::http::{HeaderValue, StatusCode, header};
+use axum::middleware;
+use axum::response::{IntoResponse, Response};
+use fluss::metadata::{
+ AddColumn, AlterConfig, AlterConfigOpType, AlterTableChanges, ColumnPositionType, DataType,
+ JsonSerde, PARTITION_SPEC_SEPARATOR, PartitionSpec, ResolvedPartitionSpec, Schema,
+ TableDescriptor, TableInfo, TablePath,
+};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use utoipa::ToSchema;
+use utoipa_axum::router::OpenApiRouter;
+use utoipa_axum::routes;
+
+/// DDL routes, merged into the main router by [`crate::protocol::rest::build_router`].
+pub fn routes() -> OpenApiRouter<RestState> {
+ OpenApiRouter::new()
+ .routes(routes!(create_database))
+ .routes(routes!(drop_database))
+ .routes(routes!(create_table))
+ .routes(routes!(alter_table))
+ .routes(routes!(drop_table))
+ .routes(routes!(create_partition))
+ .routes(routes!(drop_partition))
+ .route_layer(middleware::from_fn(reject_query_parameters))
+}
+
+/// Body of `POST /v1/clusters/{cluster}/databases`.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct CreateDatabaseBody {
+ pub database: String,
+}
+
+/// One column; top-level nullability belongs here, not inside `data_type`.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct ColumnBody {
+ pub name: String,
+ pub data_type: ColumnDataType,
+ /// Primary-key columns default to false and reject true; other columns default to true.
+ pub nullable: Option<bool>,
+ pub comment: Option<String>,
+}
+
+/// The bucket distribution of a table definition.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct DistributionBody {
+ #[schema(minimum = 1)]
+ pub bucket_count: i32,
+ pub bucket_keys: Vec<String>,
+}
+
+/// Body of `POST /v1/clusters/{cluster}/databases/{database}/tables`.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct CreateTableBody {
+ pub table_name: String,
+ pub columns: Vec<ColumnBody>,
+ /// Primary-key columns; omit for a log table.
+ pub primary_key: Option<Vec<String>>,
+ #[serde(default)]
+ pub partitioned_by: Vec<String>,
+ pub distribution: Option<DistributionBody>,
+ /// Fluss table properties, validated by the server.
+ #[serde(default)]
+ pub configs: HashMap<String, String>,
+ /// Custom table metadata; keys must not use the reserved `table.` prefix.
+ #[serde(default)]
+ pub custom_properties: HashMap<String, String>,
+ pub comment: Option<String>,
+ /// Build and validate the definition locally without creating anything.
+ #[serde(default)]
+ pub validate_only: bool,
+}
+
+/// The result of a successful local `validate_only` dry run.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct ValidateOnlyResponse {
+ pub validate_only: bool,
+ pub database: String,
+ pub table: String,
+ pub column_count: usize,
+ pub primary_key: Vec<String>,
+}
+
+/// One supported table change. Config keys under `table.` address Fluss table properties;
+/// other keys address custom properties.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum TableChangeBody {
+ /// An added column must be nullable: existing rows have no value for it.
+ AddColumn {
+ name: String,
+ data_type: ColumnDataType,
+ comment: Option<String>,
+ },
+ SetConfig {
+ key: String,
+ value: String,
+ },
+ /// Removes the explicitly configured value.
+ ResetConfig {
+ key: String,
+ },
+}
+
+/// Body of `PATCH /v1/clusters/{cluster}/databases/{database}/tables/{table}`.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct AlterTableBody {
+ /// The server currently rejects mixing schema and config changes.
+ pub changes: Vec<TableChangeBody>,
+}
+
+/// Body of `POST .../tables/{table}/partitions`.
+#[derive(Debug, Deserialize, ToSchema)]
+#[serde(deny_unknown_fields)]
+pub struct CreatePartitionBody {
+ pub partition: HashMap<String, String>,
+}
+
+/// Creates one database.
+#[utoipa::path(
+ post,
+ path = "/v1/clusters/{cluster}/databases",
+ operation_id = "createDatabase",
+ tag = "ddl",
+ params(("cluster" = String, Path, description = "Configured cluster ID")),
+ request_body(content = CreateDatabaseBody, content_type = "application/json"),
+ responses(
+ (status = 201, description = "The created database", body = DatabaseResponse,
+ headers(("Location" = String, description = "Created database URL"))),
+ (status = 400, description = "Malformed body or invalid definition", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster", body = ErrorEnvelope),
+ (status = 409, description = "The database already exists", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 415, description = "The body is not JSON", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn create_database(
+ State(state): State<RestState>,
+ Path(cluster): Path<String>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let body: CreateDatabaseBody = parse_json_body(request).await?;
+ let name = body.database;
+ native_name("database name", &name, true)?;
+ backend.create_database(&ctx, &name).await?;
+ let location = database_location(&cluster, &name);
+ created_response(
+ json_response(&DatabaseResponse { database: name })?,
+ &location,
+ )
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Drops a database only if it contains no tables.
+#[utoipa::path(
+ delete,
+ path = "/v1/clusters/{cluster}/databases/{database}",
+ operation_id = "dropDatabase",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ),
+ responses(
+ (status = 204, description = "The database was dropped"),
+ (status = 400, description = "Unsupported query parameter", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster or database", body = ErrorEnvelope),
+ (status = 409, description = "The database still holds tables", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn drop_database(
+ State(state): State<RestState>,
+ Path((cluster, database)): Path<(String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ backend.drop_database(&ctx, &database).await?;
+ Ok(StatusCode::NO_CONTENT.into_response())
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Creates one table, or validates its definition without creating it.
+#[utoipa::path(
+ post,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables",
+ operation_id = "createTable",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ),
+ request_body(content = CreateTableBody, content_type = "application/json"),
+ responses(
+ (status = 200, description = "Dry run: fluss-rs accepted the definition and nothing was created", body = ValidateOnlyResponse),
+ (status = 201, description = "The table was created; GET Location to read its metadata",
+ headers(("Location" = String, description = "Created table URL"))),
+ (status = 400, description = "Malformed body or invalid definition", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster or database", body = ErrorEnvelope),
+ (status = 409, description = "The table already exists", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 415, description = "The body is not JSON", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn create_table(
+ State(state): State<RestState>,
+ Path((cluster, database)): Path<(String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let body: CreateTableBody = parse_json_body(request).await?;
+ let validate_only = body.validate_only;
+ let table = TablePath::new(database, body.table_name.clone());
+ let descriptor = table_descriptor(table.database(), body)?;
+ if validate_only {
+ // TODO: Forward dry runs when Fluss exposes server-side validation through fluss-rs.
+ let schema = descriptor.schema();
+ return json_response(&ValidateOnlyResponse {
+ validate_only: true,
+ database: table.database().to_string(),
+ table: table.table().to_string(),
+ column_count: schema.columns().len(),
+ primary_key: schema
+ .primary_key()
+ .map(|key| key.column_names().to_vec())
+ .unwrap_or_default(),
+ });
+ }
+ backend.create_table(&ctx, &table, &descriptor).await?;
+ created_response((), &table_location(&cluster, &table))
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Applies one group of schema or config changes to a table.
+#[utoipa::path(
+ patch,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}",
+ operation_id = "alterTable",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ),
+ request_body(content = AlterTableBody, content_type = "application/json"),
+ responses(
+ (status = 204, description = "The table was altered; GET the same URL to read its metadata"),
+ (status = 400, description = "Malformed body or invalid change", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, or table", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 415, description = "The body is not JSON", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn alter_table(
+ State(state): State<RestState>,
+ Path((cluster, database, table)): Path<(String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let body: AlterTableBody = parse_json_body(request).await?;
+ let table = TablePath::new(database, table);
+ let changes = table_changes(body)?;
+ backend.alter_table(&ctx, &table, changes).await?;
+ Ok(StatusCode::NO_CONTENT.into_response())
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Drops one table.
+#[utoipa::path(
+ delete,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}",
+ operation_id = "dropTable",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ),
+ responses(
+ (status = 204, description = "The table was dropped"),
+ (status = 400, description = "Unsupported query parameter", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, or table", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn drop_table(
+ State(state): State<RestState>,
+ Path((cluster, database, table)): Path<(String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ backend
+ .drop_table(&ctx, &TablePath::new(database, table))
+ .await?;
+ Ok(StatusCode::NO_CONTENT.into_response())
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Creates one partition of a partitioned table.
+#[utoipa::path(
+ post,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions",
+ operation_id = "createPartition",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ),
+ request_body(content = CreatePartitionBody, content_type = "application/json"),
+ responses(
+ (status = 201, description = "The created partition", body = PartitionResponse,
+ headers(("Location" = String, description = "Created partition URL"))),
+ (status = 400, description = "Malformed body, or a spec that does not match the partition keys", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, or table", body = ErrorEnvelope),
+ (status = 409, description = "The partition already exists or the table's partition limit was reached", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 415, description = "The body is not JSON", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn create_partition(
+ State(state): State<RestState>,
+ Path((cluster, database, table)): Path<(String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let body: CreatePartitionBody = parse_json_body(request).await?;
+ let table = TablePath::new(database, table);
+ let current = backend.describe_table(&ctx, &table).await?;
+ for value in body.partition.values() {
+ native_name("partition value", value, true)?;
+ }
+ let spec = PartitionSpec::new(body.partition.clone());
+ backend.create_partition(&ctx, &table, &spec).await?;
+ // Resolve the Location only after Fluss accepts the unmodified spec.
+ let name = ResolvedPartitionSpec::from_partition_spec(current.partition_keys, &spec)
+ .get_partition_name();
+ created_response(
+ json_response(&PartitionResponse {
+ database: table.database().to_string(),
+ table: table.table().to_string(),
+ partition: body.partition,
+ })?,
+ &partition_location(&cluster, &table, &name),
+ )
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Drops one partition by its Fluss name.
+#[utoipa::path(
+ delete,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions/{partition}",
+ operation_id = "dropPartition",
+ tag = "ddl",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ("partition" = String, Path,
+ description = "Fluss partition name; values follow partition-key order and are joined by `$`"),
+ ),
+ responses(
+ (status = 204, description = "The partition was dropped"),
+ (status = 400, description = "Unsupported query parameter, or a name that does not match the partition keys", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, table, or partition", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn drop_partition(
+ State(state): State<RestState>,
+ Path((cluster, database, table, partition)): Path<(String, String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let table = TablePath::new(database, table);
+ let current = backend.describe_table(&ctx, &table).await?;
+ let spec = partition_of_name(&partition, ¤t)?;
+ backend.drop_partition(&ctx, &table, &spec).await?;
+ Ok(StatusCode::NO_CONTENT.into_response())
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Converts a REST definition using the native builders.
+fn table_descriptor(database: &str, body: CreateTableBody) -> GatewayResult<TableDescriptor> {
+ native_name("database name", database, true)?;
+ native_name("table name", &body.table_name, true)?;
+ if body
+ .custom_properties
+ .keys()
+ .any(|key| key.starts_with("table."))
+ {
+ return Err(GatewayError::invalid_argument(
+ "custom property keys must not use the reserved `table.` prefix",
+ ));
+ }
+
+ let mut schema = Schema::builder();
+ for column in body.columns {
+ if column.nullable == Some(true)
+ && body
+ .primary_key
+ .as_ref()
+ .is_some_and(|keys| keys.contains(&column.name))
+ {
+ return Err(GatewayError::invalid_argument(format!(
+ "primary-key column `{}` must not declare nullable: true",
+ column.name
+ )));
+ }
+ let data_type = column
+ .data_type
+ .0
+ .with_root_nullable(column.nullable.unwrap_or(true));
+ schema = schema.column(column.name, DataType::try_from(data_type)?);
+ if let Some(comment) = column.comment {
+ schema = schema.with_comment(comment);
+ }
+ }
+ if let Some(primary_key) = body.primary_key {
+ schema = schema.primary_key(primary_key);
+ }
+ let schema = schema.build().map_err(|error| {
+ GatewayError::invalid_argument(format!("invalid table schema: {error}"))
+ })?;
+
+ let mut descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .properties(body.configs)
+ .custom_properties(body.custom_properties)
+ .partitioned_by(body.partitioned_by);
+ if let Some(distribution) = body.distribution {
+ descriptor =
+ descriptor.distributed_by(Some(distribution.bucket_count), distribution.bucket_keys);
+ }
+ if let Some(comment) = body.comment {
+ descriptor = descriptor.comment(comment);
+ }
+ descriptor.build().map_err(|error| {
+ GatewayError::invalid_argument(format!("invalid table definition: {error}"))
+ })
+}
+
+/// Builds the native alter-table batch supported by the current Fluss API.
+fn table_changes(body: AlterTableBody) -> GatewayResult<AlterTableChanges> {
+ let mut changes = AlterTableChanges::default();
+ for change in body.changes {
+ match change {
+ TableChangeBody::AddColumn {
+ name,
+ data_type,
+ comment,
+ } => {
+ let native = DataType::try_from(data_type.0.with_root_nullable(true))?;
+ let json = native.serialize_json().map_err(encoding_failure)?;
+ changes.add_columns.push(AddColumn {
+ column_name: name,
+ data_type_json: serde_json::to_vec(&json).map_err(encoding_failure)?,
+ comment,
+ position: ColumnPositionType::Last,
+ });
+ }
+ TableChangeBody::SetConfig { key, value } => {
+ changes.config_changes.push(AlterConfig::new(
+ key,
+ Some(value),
+ AlterConfigOpType::Set,
+ ));
+ }
+ TableChangeBody::ResetConfig { key } => {
+ changes
+ .config_changes
+ .push(AlterConfig::new(key, None, AlterConfigOpType::Delete));
+ }
+ }
+ }
+ Ok(changes)
+}
+
+/// Converts a native partition name from the REST path back into a spec.
+fn partition_of_name(name: &str, current: &TableInfo) -> GatewayResult<PartitionSpec> {
+ let values = name
+ .split(PARTITION_SPEC_SEPARATOR)
+ .map(|value| {
+ native_name("partition value", value, false)?;
+ Ok(value.to_string())
+ })
+ .collect::<GatewayResult<Vec<_>>>()?;
+ ResolvedPartitionSpec::new(current.partition_keys.clone(), values)
+ .map(|resolved| resolved.to_partition_spec())
+ .map_err(|error| GatewayError::invalid_argument(format!("invalid partition: {error}")))
+}
+
+fn native_name(kind: &str, value: &str, reject_internal_prefix: bool) -> GatewayResult<()> {
+ if let Some(reason) = TablePath::detect_invalid_name(value).or_else(|| {
+ reject_internal_prefix
+ .then(|| TablePath::validate_prefix(value))
+ .flatten()
+ }) {
+ return Err(GatewayError::invalid_argument(format!(
+ "invalid {kind} `{value}`: {reason}"
+ )));
+ }
+ Ok(())
+}
+
+fn encoding_failure(error: impl std::fmt::Display) -> GatewayError {
+ log::error!("failed to encode a native column type: {error}");
+ GatewayError::internal("the gateway failed to encode the new column type")
+}
+
+/// A 201 with the created resource's `Location`.
+fn created_response(body: impl IntoResponse, location: &str) -> GatewayResult<Response> {
+ let mut response = (StatusCode::CREATED, body).into_response();
+ let location = HeaderValue::from_str(location).map_err(|error| {
+ GatewayError::internal(format!("failed to render the Location header: {error}"))
+ })?;
+ response.headers_mut().insert(header::LOCATION, location);
+ Ok(response)
+}
+
+fn database_location(cluster: &str, database: &str) -> String {
+ format!(
+ "/v1/clusters/{}/databases/{}",
+ encode_segment(cluster),
+ encode_segment(database)
+ )
+}
+
+fn table_location(cluster: &str, table: &TablePath) -> String {
+ format!(
+ "{}/tables/{}",
+ database_location(cluster, table.database()),
+ encode_segment(table.table())
+ )
+}
+
+fn partition_location(cluster: &str, table: &TablePath, partition: &str) -> String {
+ format!(
+ "{}/partitions/{}",
+ table_location(cluster, table),
+ encode_segment(partition)
+ )
+}
+
+/// Percent-encodes one path segment.
+fn encode_segment(segment: &str) -> String {
+ let mut encoded = String::with_capacity(segment.len());
+ for byte in segment.as_bytes() {
+ match byte {
+ b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
+ encoded.push(*byte as char);
+ }
+ _ => encoded.push_str(&format!("%{byte:02X}")),
+ }
+ }
+ encoded
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::backend::FlussBackend;
+ use crate::backend::fake::{FakeCall, FakeFlussBackend, Operation};
+ use crate::error::{GatewayError, Resource};
+ use crate::protocol::rest::test_support;
+ use axum::body::{Body, Bytes};
+ use axum::http::{Method, Request as HttpRequest, StatusCode, header};
+ use fluss::metadata::{
+ AlterConfigOpType, BigIntType, DataType, DecimalType, PartitionInfo, ResolvedPartitionSpec,
+ Schema, StringType, TableDescriptor, TableInfo, TablePath,
+ };
+ use http_body_util::BodyExt;
+ use serde_json::{Value, json};
+ use std::collections::HashMap;
+ use std::sync::Arc;
+ use tower::ServiceExt;
+
+ fn app(backend: Arc<FakeFlussBackend>) -> axum::Router {
+ let state = test_support::state_with_backend(backend as Arc<dyn FlussBackend>);
+ state.readiness.set_serving();
+ crate::protocol::rest::build_router(state, &test_support::test_options())
+ }
+
+ fn gateway() -> (Arc<FakeFlussBackend>, axum::Router) {
+ let backend = Arc::new(FakeFlussBackend::new());
+ (Arc::clone(&backend), app(backend))
+ }
+
+ async fn send(
+ app: &axum::Router,
+ method: Method,
+ path: &str,
+ body: Option<Value>,
+ ) -> (StatusCode, Option<String>, Value) {
+ let mut builder = HttpRequest::builder().method(method).uri(path);
+ let body = match body {
+ Some(value) => {
+ builder = builder.header(header::CONTENT_TYPE, "application/json");
+ Body::from(serde_json::to_vec(&value).expect("the body serializes"))
+ }
+ None => Body::empty(),
+ };
+ let response = app
+ .clone()
+ .oneshot(builder.body(body).expect("a valid request"))
+ .await
+ .expect("the router answers");
+ let status = response.status();
+ let location = response
+ .headers()
+ .get(header::LOCATION)
+ .map(|value| value.to_str().expect("an ASCII Location").to_string());
+ let bytes = response
+ .into_body()
+ .collect()
+ .await
+ .expect("a body")
+ .to_bytes();
+ let parsed = if bytes.is_empty() {
+ Value::Null
+ } else {
+ Value::Object(serde_json::from_slice(&bytes).expect("a JSON object body"))
+ };
+ (status, location, parsed)
+ }
+
+ async fn post(
+ app: &axum::Router,
+ path: &str,
+ body: Value,
+ ) -> (StatusCode, Option<String>, Value) {
+ send(app, Method::POST, path, Some(body)).await
+ }
+
+ async fn delete(app: &axum::Router, path: &str) -> (StatusCode, Value) {
+ let (status, _, body) = send(app, Method::DELETE, path, None).await;
+ (status, body)
+ }
+
+ fn partitioned_table() -> Value {
+ json!({
+ "table_name": "orders",
+ "columns": [
+ {"name": "id", "data_type": {"type": "BIGINT"}, "nullable": false},
+ {"name": "dt", "data_type": {"type": "STRING"}},
+ {"name": "amount", "data_type": {"type": "DECIMAL", "precision": 18, "scale": 2}, "nullable": true, "comment": "the order total"},
+ ],
+ "primary_key": ["id", "dt"],
+ "partitioned_by": ["dt"],
+ "distribution": {"bucket_count": 4, "bucket_keys": ["id"]},
+ "configs": {"table.log.ttl": "7d"},
+ "custom_properties": {"app.owner": "sales", "app.stage": "draft"},
+ "comment": "the orders table",
+ })
+ }
+
+ #[tokio::test]
+ async fn ddl_requests_are_converted_and_forwarded() {
+ let (backend, app) = gateway();
+
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases",
+ json!({"database": "sales"}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::CREATED);
+ assert_eq!(
+ location.as_deref(),
+ Some("/v1/clusters/default/databases/sales")
+ );
+ assert_eq!(body, json!({"database": "sales"}));
+
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::with_nullable(false)))
+ .column("dt", DataType::String(StringType::with_nullable(false)))
+ .column(
+ "amount",
+ DataType::Decimal(DecimalType::with_nullable(true, 18, 2).unwrap()),
+ )
+ .with_comment("the order total")
+ .primary_key(["id", "dt"])
+ .build()
+ .expect("a fixture schema");
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ partitioned_table(),
+ )
+ .await;
+ assert_eq!(status, StatusCode::CREATED);
+ assert_eq!(
+ location.as_deref(),
+ Some("/v1/clusters/default/databases/sales/tables/orders")
+ );
+ assert_eq!(body, Value::Null);
+
+ let (status, _, body) = send(
+ &app,
+ Method::PATCH,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ Some(json!({
+ "changes": [
+ {"kind": "add_column", "name": "note", "data_type": {"type": "STRING"}}
+ ]
+ })),
+ )
+ .await;
+ assert_eq!(status, StatusCode::NO_CONTENT);
+ assert_eq!(body, Value::Null);
+
+ let (status, _, body) = send(
+ &app,
+ Method::PATCH,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ Some(json!({
+ "changes": [
+ {"kind": "set_config", "key": "table.log.ttl", "value": "30d"},
+ {"kind": "reset_config", "key": "table.datalake.enabled"},
+ {"kind": "set_config", "key": "app.owner", "value": "finance"},
+ {"kind": "reset_config", "key": "app.stage"}
+ ]
+ })),
+ )
+ .await;
+ assert_eq!(status, StatusCode::NO_CONTENT);
+ assert_eq!(body, Value::Null);
+
+ backend.define_table(TableInfo::of(
+ TablePath::new("sales", "orders"),
+ 1,
+ 1,
+ TableDescriptor::builder()
+ .schema(schema.clone())
+ .partitioned_by(vec!["dt"])
+ .distributed_by(Some(4), vec!["id".to_string()])
+ .build()
+ .expect("a fixture descriptor"),
+ 0,
+ 0,
+ ));
+
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
+ json!({"partition": {"dt": "2026-08-25"}}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::CREATED);
+ assert_eq!(
+ location.as_deref(),
+ Some("/v1/clusters/default/databases/sales/tables/orders/partitions/2026-08-25")
+ );
+ assert_eq!(
+ body,
+ json!({
+ "database": "sales",
+ "table": "orders",
+ "partition": {"dt": "2026-08-25"},
+ })
+ );
+
+ for path in [
+ "/v1/clusters/default/databases/sales/tables/orders/partitions/2026-08-25",
+ "/v1/clusters/default/databases/sales/tables/orders",
+ "/v1/clusters/default/databases/sales",
+ ] {
+ let (status, body) = delete(&app, path).await;
+ assert_eq!(status, StatusCode::NO_CONTENT, "{path}");
+ assert_eq!(body, Value::Null, "{path}");
+ }
+
+ let calls = backend.calls();
+ assert_eq!(calls.len(), 8);
+ assert!(matches!(&calls[0], FakeCall::CreateDatabase(name) if name == "sales"));
+ let FakeCall::CreateTable(table, descriptor) = &calls[1] else {
+ panic!("expected create table, got {:?}", calls[1]);
+ };
+ assert_eq!(table, &TablePath::new("sales", "orders"));
+ assert_eq!(descriptor.schema(), &schema);
+ assert_eq!(descriptor.partition_keys(), ["dt"]);
+ assert_eq!(
+ descriptor.table_distribution().unwrap().bucket_count(),
+ Some(4)
+ );
+ assert_eq!(descriptor.bucket_keys(), ["id"]);
+ assert_eq!(descriptor.comment(), Some("the orders table"));
+ assert_eq!(
+ descriptor.properties(),
+ &HashMap::from([("table.log.ttl".to_string(), "7d".to_string())])
+ );
+ assert_eq!(
+ serde_json::to_value(descriptor.custom_properties()).unwrap(),
+ partitioned_table()["custom_properties"]
+ );
+
+ let FakeCall::AlterTable(_, add_column) = &calls[2] else {
+ panic!("expected add column, got {:?}", calls[2]);
+ };
+ assert_eq!(add_column.add_columns.len(), 1);
+ assert_eq!(add_column.add_columns[0].column_name, "note");
+ assert_eq!(
+ serde_json::from_slice::<Value>(&add_column.add_columns[0].data_type_json).unwrap(),
+ json!({"type": "STRING"})
+ );
+
+ let FakeCall::AlterTable(_, configs) = &calls[3] else {
+ panic!("expected config changes, got {:?}", calls[3]);
+ };
+ assert_eq!(
+ configs
+ .config_changes
+ .iter()
+ .map(|change| (
+ change.config_key.as_str(),
+ change.config_value.as_deref(),
+ change.op_type,
+ ))
+ .collect::<Vec<_>>(),
+ [
+ ("table.log.ttl", Some("30d"), AlterConfigOpType::Set),
+ ("table.datalake.enabled", None, AlterConfigOpType::Delete),
+ ("app.owner", Some("finance"), AlterConfigOpType::Set),
+ ("app.stage", None, AlterConfigOpType::Delete),
+ ]
+ );
+
+ for call in [&calls[4], &calls[5]] {
+ let spec = match call {
+ FakeCall::CreatePartition(table, spec) | FakeCall::DropPartition(table, spec) => {
+ assert_eq!(table, &TablePath::new("sales", "orders"));
+ spec
+ }
+ other => panic!("expected a partition call, got {other:?}"),
+ };
+ assert_eq!(spec.get_spec_map().get("dt").unwrap(), "2026-08-25");
+ }
+ assert!(
+ matches!(&calls[6], FakeCall::DropTable(table) if table == &TablePath::new("sales", "orders"))
+ );
+ assert!(matches!(&calls[7], FakeCall::DropDatabase(name) if name == "sales"));
+ }
+
+ #[tokio::test]
+ async fn table_definitions_are_checked_before_creation() {
+ let (backend, app) = gateway();
+ backend.define_database("sales");
+
+ for validate_only in [false, true] {
+ for (index, name) in [(0, "id"), (1, "dt")] {
+ let mut definition = partitioned_table();
+ definition["columns"][index]["nullable"] = json!(true);
+ definition["validate_only"] = json!(validate_only);
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ definition,
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
+ assert_eq!(location, None);
+ assert_eq!(body["error"]["code"], "invalid_argument");
+ assert_eq!(
+ body["error"]["message"],
+ format!("primary-key column `{name}` must not declare nullable: true")
+ );
+ }
+ }
+ assert!(backend.calls().is_empty());
+
+ let mut definition = partitioned_table();
+ definition["custom_properties"] = json!({"table.custom": "value"});
+ let (status, _, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ definition,
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
+ assert_eq!(
+ body["error"]["message"],
+ "custom property keys must not use the reserved `table.` prefix"
+ );
+ assert!(backend.calls().is_empty());
+
+ let mut definition = partitioned_table();
+ definition["validate_only"] = json!(true);
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ definition,
+ )
+ .await;
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(location, None);
+ assert_eq!(
+ body,
+ json!({
+ "validate_only": true,
+ "database": "sales",
+ "table": "orders",
+ "column_count": 3,
+ "primary_key": ["id", "dt"],
+ })
+ );
+
+ let (status, _, _) = send(
+ &app,
+ Method::GET,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ None,
+ )
+ .await;
+ assert_eq!(status, StatusCode::NOT_FOUND);
+
+ for (field, value) in [
+ ("primary_key", json!(["absent"])),
+ ("primary_key", json!([])),
+ (
+ "distribution",
+ json!({"bucket_count": 2, "bucket_keys": ["amount"]}),
+ ),
+ (
+ "columns",
+ json!([
+ {"name": "id", "data_type": {"type": "BIGINT"}},
+ {"name": "id", "data_type": {"type": "STRING"}}
+ ]),
+ ),
+ ] {
+ let mut invalid = partitioned_table();
+ invalid["validate_only"] = json!(true);
+ invalid[field] = value;
+ let (status, _, body) =
+ post(&app, "/v1/clusters/default/databases/sales/tables", invalid).await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
+ assert_eq!(body["error"]["code"], "invalid_argument");
+ assert!(
+ body["error"]["message"]
+ .as_str()
+ .unwrap()
+ .contains("invalid table")
+ );
+ }
+
+ let mut log_table = partitioned_table();
+ log_table["validate_only"] = json!(true);
+ log_table.as_object_mut().unwrap().remove("primary_key");
+ log_table
+ .as_object_mut()
+ .unwrap()
+ .remove("custom_properties");
+ let (status, _, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ log_table,
+ )
+ .await;
+ assert_eq!(status, StatusCode::OK, "{body}");
+ assert_eq!(body["primary_key"], json!([]));
+ assert!(backend.calls().is_empty());
+ }
+
+ #[tokio::test]
+ async fn a_body_that_is_not_exactly_the_contract_is_refused() {
+ let (backend, app) = gateway();
+ backend.define_database("sales");
+
+ let cases = [
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "BIGINT"}, "nullable": false}], "primary_keys": ["id"]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "BIGINT", "nullable": false}}]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "NUMBER"}}]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "BIGINT", "precision": 3}}]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "BIGINT", "precision": null}}]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "ARRAY", "element_type": {"type": "STRING", "nullable": null}}}]}),
+ json!({"table_name": "t", "columns": [{"name": "id", "data_type": {"type": "BIGINT"}}], "custom_properties": {"app.owner": 42}}),
+ json!({"columns": []}),
+ ];
+ for body in cases {
+ let (status, _, answered) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables",
+ body.clone(),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
+ assert_eq!(answered["error"]["code"], "invalid_argument", "{body}");
+ }
+
+ let response = app
+ .oneshot(
+ HttpRequest::builder()
+ .method(Method::POST)
+ .uri("/v1/clusters/default/databases")
+ .body(Body::from("{}"))
+ .expect("a valid request"),
+ )
+ .await
+ .expect("the router answers");
+ assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
+ }
+
+ #[tokio::test]
+ async fn a_conflict_and_a_missing_resource_keep_their_own_status() {
+ let (backend, app) = gateway();
+ backend.fail_once(
+ Operation::DropDatabase,
+ GatewayError::failed_precondition("database is not empty")
+ .with_resource(Resource::Database),
+ );
+ for (operation, path, definition, resource, code) in [
+ (
+ Operation::CreateDatabase,
+ "/v1/clusters/default/databases",
+ json!({"database": "sales"}),
+ Resource::Database,
+ "database_already_exists",
+ ),
+ (
+ Operation::CreateTable,
+ "/v1/clusters/default/databases/sales/tables",
+ partitioned_table(),
+ Resource::Table,
+ "table_already_exists",
+ ),
+ ] {
+ backend.fail_once(
+ operation,
+ GatewayError::already_exists("already exists").with_resource(resource),
+ );
+ let (status, _, body) = post(&app, path, definition).await;
+ assert_eq!(status, StatusCode::CONFLICT, "{operation:?}");
+ assert_eq!(body["error"]["code"], code);
+ }
+
+ let (status, body) = delete(&app, "/v1/clusters/default/databases/sales").await;
+ assert_eq!(status, StatusCode::CONFLICT);
+ assert_eq!(body["error"]["code"], "database_not_empty");
+ let (status, _) = delete(&app, "/v1/clusters/default/databases/sales").await;
+ assert_eq!(status, StatusCode::NO_CONTENT);
+
+ for (operation, path, resource) in [
+ (
+ Operation::DropDatabase,
+ "/v1/clusters/default/databases/absent",
+ Resource::Database,
+ ),
+ (
+ Operation::DropTable,
+ "/v1/clusters/default/databases/sales/tables/absent",
+ Resource::Table,
+ ),
+ ] {
+ backend.fail_once(
+ operation,
+ GatewayError::not_found("missing").with_resource(resource),
+ );
+ let (status, body) = delete(&app, path).await;
+ assert_eq!(status, StatusCode::NOT_FOUND, "{path}");
+ assert!(
+ body["error"]["code"]
+ .as_str()
+ .is_some_and(|code| code.ends_with("_not_found")),
+ "{path}: {body}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn partition_conversion_preserves_specs_and_backend_errors() {
+ let (backend, app) = gateway();
+ let definition = serde_json::from_value(json!({
+ "table_name": "orders",
+ "columns": [
+ {"name": "region", "data_type": {"type": "STRING"}},
+ {"name": "dt", "data_type": {"type": "STRING"}}
+ ],
+ "partitioned_by": ["region", "dt"],
+ "distribution": {"bucket_count": 1, "bucket_keys": []}
+ }))
+ .unwrap();
+ backend.define_table(TableInfo::of(
+ TablePath::new("sales", "orders"),
+ 1,
+ 1,
+ super::table_descriptor("sales", definition).unwrap(),
+ 0,
+ 0,
+ ));
+
+ for name in ["eu", "eu$2026-08-25$extra"] {
+ let (status, body) = delete(
+ &app,
+ &format!("/v1/clusters/default/databases/sales/tables/orders/partitions/{name}"),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST);
+ assert_eq!(body["error"]["code"], "invalid_argument");
+ }
+ assert!(backend.calls().is_empty());
+
+ let (status, location, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
+ json!({"partition": {"dt": "2026-08-25", "region": "eu"}}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::CREATED);
+ assert_eq!(
+ location.as_deref(),
+ Some("/v1/clusters/default/databases/sales/tables/orders/partitions/eu%242026-08-25")
+ );
+ assert_eq!(
+ body["partition"],
+ json!({"region": "eu", "dt": "2026-08-25"})
+ );
+
+ let table = TablePath::new("sales", "orders");
+ let spec = ResolvedPartitionSpec::new(
+ Arc::from(["region".to_string(), "dt".to_string()]),
+ vec!["eu".to_string(), "2026-08-25".to_string()],
+ )
+ .unwrap();
+ backend.define_partition(&table, PartitionInfo::new(1, spec));
+ let (status, _, listed) = send(
+ &app,
+ Method::GET,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
+ None,
+ )
+ .await;
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(
+ listed["partitions"],
+ json!([{"name": "eu$2026-08-25", "partition": body["partition"]}])
+ );
+ let name = listed["partitions"][0]["name"].as_str().unwrap();
+ let path = super::partition_location("default", &table, name);
+ assert_eq!(location.as_deref(), Some(path.as_str()));
+ assert_eq!(delete(&app, &path).await.0, StatusCode::NO_CONTENT);
+ let calls = backend.calls();
+ let Some(FakeCall::DropPartition(dropped_table, dropped_spec)) = calls.last() else {
+ panic!("expected drop partition, got {calls:?}");
+ };
+ assert_eq!(dropped_table, &table);
+ assert_eq!(
+ serde_json::to_value(dropped_spec.get_spec_map()).unwrap(),
+ body["partition"]
+ );
+
+ for partition in [
+ json!({}),
+ json!({"region": "eu"}),
+ json!({"dt": "2026-08-25", "region": "eu", "extra": "preserved"}),
+ ] {
+ backend.fail_once(
+ Operation::CreatePartition,
+ GatewayError::invalid_argument("backend rejected partition spec"),
+ );
+ let (status, _, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
+ json!({"partition": partition}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST);
+ assert_eq!(body["error"]["message"], "backend rejected partition spec");
+ let calls = backend.calls();
+ let Some(FakeCall::CreatePartition(table, spec)) = calls.last() else {
+ panic!("expected create partition, got {calls:?}");
+ };
+ assert_eq!(table, &TablePath::new("sales", "orders"));
+ assert_eq!(
+ serde_json::to_value(spec.get_spec_map()).unwrap(),
+ partition
+ );
+ }
+
+ let (status, _, body) = post(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
+ json!({"partition": {"dt": "2026$08"}}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
+
+ backend.fail_once(
+ Operation::DropPartition,
+ GatewayError::not_found("partition does not exist").with_resource(Resource::Partition),
+ );
+ let (status, body) = delete(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions/eu%242026-08-25",
+ )
+ .await;
+ assert_eq!(status, StatusCode::NOT_FOUND);
+ assert_eq!(body["error"]["code"], "partition_not_found");
+ }
+
+ #[tokio::test]
+ async fn alterations_are_forwarded_and_backend_errors_are_preserved() {
+ let (backend, app) = gateway();
+
+ for changes in [
+ json!([]),
+ json!([
+ {"kind": "set_config", "key": "table.log.ttl", "value": "30d"},
+ {"kind": "add_column", "name": "note", "data_type": {"type": "STRING"}},
+ ]),
+ ] {
+ backend.fail_once(
+ Operation::AlterTable,
+ GatewayError::invalid_argument("backend rejected alteration"),
+ );
+ let (status, _, body) = send(
+ &app,
+ Method::PATCH,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ Some(json!({"changes": changes})),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST);
+ assert_eq!(body["error"]["code"], "invalid_argument");
+ assert_eq!(body["error"]["message"], "backend rejected alteration");
+ }
+
+ let calls = backend.calls();
+ assert_eq!(calls.len(), 2);
+ for (i, call) in calls.iter().enumerate() {
+ let FakeCall::AlterTable(table, changes) = call else {
+ panic!("expected alter table, got {call:?}");
+ };
+ assert_eq!(table, &TablePath::new("sales", "orders"));
+ assert_eq!(changes.add_columns.len(), i);
+ assert_eq!(changes.config_changes.len(), i);
+ }
+ }
+
+ #[tokio::test]
+ async fn a_mutation_is_checked_before_its_body() {
+ let (backend, app) = gateway();
+
+ for (method, resource) in [
+ (Method::POST, "/databases"),
+ (Method::DELETE, "/databases/sales"),
+ (Method::POST, "/databases/sales/tables"),
+ (Method::PATCH, "/databases/sales/tables/orders"),
+ (Method::DELETE, "/databases/sales/tables/orders"),
+ (Method::POST, "/databases/sales/tables/orders/partitions"),
+ (
+ Method::DELETE,
+ "/databases/sales/tables/orders/partitions/eu",
+ ),
+ ] {
+ for (cluster, query, status, code) in [
+ ("other", "", StatusCode::NOT_FOUND, "cluster_not_found"),
+ (
+ "other",
+ "?dry_run=true",
+ StatusCode::BAD_REQUEST,
+ "invalid_argument",
+ ),
+ (
+ "default",
+ "?dry_run=true",
+ StatusCode::BAD_REQUEST,
+ "invalid_argument",
+ ),
+ ("%FF", "", StatusCode::BAD_REQUEST, "invalid_argument"),
+ ] {
+ let path = format!("/v1/clusters/{cluster}{resource}{query}");
+ let body = Body::from_stream(futures_util::stream::poll_fn(
+ |_| -> std::task::Poll<Option<Result<Bytes, std::io::Error>>> {
+ panic!("rejected request body must not be read");
+ },
+ ));
+ let response = app
+ .clone()
+ .oneshot(
+ HttpRequest::builder()
+ .method(method.clone())
+ .uri(&path)
+ .body(body)
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(response.status(), status, "{method} {path}");
+ let request_id = response.headers()["x-request-id"]
+ .to_str()
+ .unwrap()
+ .to_string();
+ let bytes = response.into_body().collect().await.unwrap().to_bytes();
+ let body: Value = serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(body["error"]["code"], code, "{method} {path}");
+ assert_eq!(body["error"]["request_id"], request_id, "{method} {path}");
+ }
+ }
+ assert!(backend.calls().is_empty());
+
+ let (status, _, body) = post(
+ &app,
+ "/v1/clusters/default/databases",
+ json!({"database": "x".repeat(2048)}),
+ )
+ .await;
+ assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE);
+ assert_eq!(body["error"]["code"], "limit_exceeded");
+
+ assert!(backend.calls().is_empty());
+ }
+
+ #[test]
+ fn a_location_percent_encodes_its_segments() {
+ assert_eq!(
+ super::database_location("default", "sales db"),
+ "/v1/clusters/default/databases/sales%20db"
+ );
+ assert_eq!(
+ super::partition_location("default", &TablePath::new("sales", "orders"), "2026/08"),
+ "/v1/clusters/default/databases/sales/tables/orders/partitions/2026%2F08"
+ );
+ }
+}
diff --git a/fluss-gateway/src/protocol/rest/health.rs b/fluss-gateway/src/protocol/rest/health.rs
index c0e284dce..8b780d325 100644
--- a/fluss-gateway/src/protocol/rest/health.rs
+++ b/fluss-gateway/src/protocol/rest/health.rs
@@ -17,7 +17,7 @@
//! Health endpoints.
//!
-//! `GET /health` returns the FIP-49 `{status, uptime_ms}` shape and answers from the event loop
+//! `GET /health` returns `{status, uptime_ms}` and answers from the event loop
//! without a backend RPC; deeper diagnostics live in the Prometheus metrics, not in this payload.
//! `GET /ready` is the readiness counterpart: 200 only while the gateway accepts traffic.
@@ -37,7 +37,7 @@ pub fn routes() -> OpenApiRouter<RestState> {
.routes(routes!(ready))
}
-/// Response of `GET /health` (FIP-49): liveness plus process uptime.
+/// Response of `GET /health`: liveness plus process uptime.
#[derive(Debug, Serialize, ToSchema)]
pub struct HealthResponse {
pub status: &'static str,
@@ -45,7 +45,7 @@ pub struct HealthResponse {
pub uptime_ms: u64,
}
-/// The FIP-49 health summary: `{status, uptime_ms}`, always 200 while the process answers.
+/// Returns process liveness and uptime without a backend RPC.
#[utoipa::path(
get,
path = "/health",
@@ -130,7 +130,7 @@ mod tests {
serde_json::from_slice(&bytes).expect("json body")
}
- /// `/health` answers the FIP-49 `{status, uptime_ms}` shape and nothing else.
+ /// `/health` answers `{status, uptime_ms}` and nothing else.
#[tokio::test]
async fn health_answers_status_and_uptime_only() {
let response = get(app(), "/health").await;
@@ -141,7 +141,7 @@ mod tests {
assert_eq!(
json.as_object().expect("object").len(),
2,
- "no diagnostic fields beyond the FIP shape: {json}"
+ "no diagnostic fields beyond status and uptime_ms: {json}"
);
}
diff --git a/fluss-gateway/src/protocol/rest/metadata.rs b/fluss-gateway/src/protocol/rest/metadata.rs
index 26df5051b..c52f586f7 100644
--- a/fluss-gateway/src/protocol/rest/metadata.rs
+++ b/fluss-gateway/src/protocol/rest/metadata.rs
@@ -15,25 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-//! Metadata discovery: listing the databases of a cluster and the tables of a database.
-//!
-//! Both handlers do the same four things and nothing else: check that the cluster is one this gateway
-//! serves (404), validate the page parameters (400), call the backend, and cut the page out of the
-//! answer. Every other status — 429, 500, 503, 504 — is emitted by shared admission or middleware,
-//! or propagated from [`crate::backend::FlussBackend`], so this module knows nothing about
-//! connections, identity modes, or capacity.
+//! Catalog metadata endpoints.
+use crate::backend::FlussBackend;
use crate::backend::context::RequestContext;
-use crate::backend::types::ClusterId;
-use crate::backend::{FlussBackend, unknown_cluster};
use crate::error::{ErrorEnvelope, GatewayResult};
+use crate::protocol::rest::datatype::ColumnDataType;
use crate::protocol::rest::pagination::{Collection, Page};
use crate::protocol::rest::{
- RestState, error_response, json_response, request_context, request_id,
+ RestState, error_response, json_response, reject_query_parameters, request_id, resolve_cluster,
};
use axum::extract::{Path, Request, State};
+use axum::middleware;
use axum::response::Response;
+use fluss::metadata::{PartitionInfo, TableInfo, TablePath};
use serde::Serialize;
+use std::collections::HashMap;
use std::sync::Arc;
use utoipa::ToSchema;
use utoipa_axum::router::OpenApiRouter;
@@ -45,6 +42,7 @@ pub struct DatabasesResponse {
pub databases: Vec<String>,
/// Present only while more entries follow.
#[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
pub next_page_token: Option<String>,
}
@@ -54,14 +52,145 @@ pub struct TablesResponse {
pub tables: Vec<String>,
/// Present only while more entries follow.
#[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub next_page_token: Option<String>,
+}
+
+/// Response of a database creation.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct DatabaseResponse {
+ pub database: String,
+}
+
+/// One column of a table schema.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct ColumnResponse {
+ pub name: String,
+ pub data_type: ColumnDataType,
+ pub nullable: bool,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub comment: Option<String>,
+}
+
+/// The bucket distribution of a table.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct DistributionResponse {
+ #[schema(minimum = 1)]
+ pub bucket_count: i32,
+ pub bucket_keys: Vec<String>,
+}
+
+/// Response of `GET /v1/clusters/{cluster}/databases/{database}/tables/{table}`.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct TableResponse {
+ pub database: String,
+ pub table: String,
+ pub columns: Vec<ColumnResponse>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub primary_key: Option<Vec<String>>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub partitioned_by: Option<Vec<String>>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub distribution: Option<DistributionResponse>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub comment: Option<String>,
+ /// Fluss table properties.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub configs: Option<HashMap<String, String>>,
+ /// Custom table metadata.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
+ pub custom_properties: Option<HashMap<String, String>>,
+}
+
+impl From<&TableInfo> for TableResponse {
+ fn from(info: &TableInfo) -> Self {
+ Self {
+ database: info.table_path.database().to_string(),
+ table: info.table_path.table().to_string(),
+ columns: info
+ .schema
+ .columns()
+ .iter()
+ .map(|column| {
+ let data_type = column.data_type();
+ ColumnResponse {
+ name: column.name().to_string(),
+ data_type: ColumnDataType::from(data_type),
+ nullable: data_type.is_nullable(),
+ comment: column.comment().map(str::to_string),
+ }
+ })
+ .collect(),
+ primary_key: (!info.primary_keys.is_empty()).then(|| info.primary_keys.clone()),
+ partitioned_by: (!info.partition_keys.is_empty()).then(|| info.partition_keys.to_vec()),
+ distribution: (info.num_buckets > 0).then(|| DistributionResponse {
+ bucket_count: info.num_buckets,
+ bucket_keys: info.bucket_keys.clone(),
+ }),
+ comment: info.comment.clone(),
+ configs: (!info.properties.is_empty()).then(|| info.properties.clone()),
+ custom_properties: (!info.custom_properties.is_empty())
+ .then(|| info.custom_properties.clone()),
+ }
+ }
+}
+
+/// Response of a partition creation.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct PartitionResponse {
+ pub database: String,
+ pub table: String,
+ pub partition: HashMap<String, String>,
+}
+
+/// One partition in a listing.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct PartitionEntry {
+ /// Native partition name; percent-encode it as a path segment when deleting.
+ pub name: String,
+ pub partition: HashMap<String, String>,
+}
+
+impl From<&PartitionInfo> for PartitionEntry {
+ fn from(info: &PartitionInfo) -> Self {
+ let spec = info.get_resolved_partition_spec();
+ Self {
+ name: info.get_partition_name(),
+ partition: spec
+ .get_partition_keys()
+ .iter()
+ .cloned()
+ .zip(spec.get_partition_values().iter().cloned())
+ .collect(),
+ }
+ }
+}
+
+/// Response of `GET /v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions`.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct PartitionsResponse {
+ pub partitions: Vec<PartitionEntry>,
+ /// Present only while more entries follow.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[schema(nullable = false)]
pub next_page_token: Option<String>,
}
/// Metadata routes, merged into the main router by [`crate::protocol::rest::build_router`].
pub fn routes() -> OpenApiRouter<RestState> {
OpenApiRouter::new()
+ .routes(routes!(describe_table))
+ .route_layer(middleware::from_fn(reject_query_parameters))
.routes(routes!(list_databases))
.routes(routes!(list_tables))
+ .routes(routes!(list_partitions))
}
/// Lists the databases of one configured cluster.
@@ -82,8 +211,10 @@ pub fn routes() -> OpenApiRouter<RestState> {
(status = 200, description = "Databases in lexical order", body = DatabasesResponse),
(status = 400, description = "Invalid page parameter or page token", body = ErrorEnvelope),
(status = 404, description = "Unknown cluster", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
(status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
(status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
(status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
(status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
)
@@ -94,7 +225,7 @@ pub(crate) async fn list_databases(
request: Request,
) -> Response {
let request_id = request_id(&request);
- let prepared = prepare(&state, &request, &cluster, Collection::Databases, None);
+ let prepared = prepare_page(&state, &request, &cluster, Collection::Databases, None);
let result = async {
let (backend, page, ctx) = prepared?;
let (databases, next_page_token) = page.apply(backend.list_databases(&ctx).await?);
@@ -126,8 +257,10 @@ pub(crate) async fn list_databases(
(status = 200, description = "Tables in lexical order", body = TablesResponse),
(status = 400, description = "Invalid page parameter or page token", body = ErrorEnvelope),
(status = 404, description = "Unknown cluster or database", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
(status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
(status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
(status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
(status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
)
@@ -138,7 +271,7 @@ pub(crate) async fn list_tables(
request: Request,
) -> Response {
let request_id = request_id(&request);
- let prepared = prepare(
+ let prepared = prepare_page(
&state,
&request,
&cluster,
@@ -157,53 +290,140 @@ pub(crate) async fn list_tables(
result.unwrap_or_else(|error| error_response(&error, &request_id))
}
-/// Validates the cluster (404) and the page request (400), then builds the backend context.
-///
-/// Everything that reads the HTTP request happens here, before the backend is called: a malformed page
-/// token is therefore always a 400, never the 404 of a parent resource that happens not to exist. The
-/// cluster is checked first, because a path resource that does not exist outranks a bad query
-/// parameter.
-fn prepare(
+/// Describes one table: its schema, keys, distribution, partitioning, and properties.
+#[utoipa::path(
+ get,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}",
+ operation_id = "describeTable",
+ tag = "metadata",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ),
+ responses(
+ (status = 200, description = "The table metadata", body = TableResponse),
+ (status = 400, description = "Unsupported query parameter", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, or table", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn describe_table(
+ State(state): State<RestState>,
+ Path((cluster, database, table)): Path<(String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let prepared = resolve_cluster(&state, &request, &cluster);
+ let result = async {
+ let (backend, ctx) = prepared?;
+ let description = backend
+ .describe_table(&ctx, &TablePath::new(database, table))
+ .await?;
+ json_response(&TableResponse::from(&description))
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Lists the partitions of one partitioned table.
+#[utoipa::path(
+ get,
+ path = "/v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions",
+ operation_id = "listPartitions",
+ tag = "metadata",
+ params(
+ ("cluster" = String, Path, description = "Configured cluster ID"),
+ ("database" = String, Path, description = "Database name"),
+ ("table" = String, Path, description = "Table name"),
+ ("max_results" = Option<usize>, Query,
+ description = "Maximum entries to return. Defaults to 100, capped at 1000.",
+ minimum = 1, maximum = 1000),
+ ("page_token" = Option<String>, Query,
+ description = "Opaque token from the `next_page_token` of the previous response."),
+ ),
+ responses(
+ (status = 200, description = "Partitions in partition-name order", body = PartitionsResponse),
+ (status = 400, description = "Invalid page parameter or page token", body = ErrorEnvelope),
+ (status = 404, description = "Unknown cluster, database, or table", body = ErrorEnvelope),
+ (status = 413, description = "Request body above the configured limit", body = ErrorEnvelope),
+ (status = 429, description = "Metadata concurrency limit exceeded", body = ErrorEnvelope),
+ (status = 500, description = "Fluss backend failure", body = ErrorEnvelope),
+ (status = 501, description = "Fluss does not support the operation or API version", body = ErrorEnvelope),
+ (status = 503, description = "Fluss is unavailable, or the gateway is starting or shutting down", body = ErrorEnvelope),
+ (status = 504, description = "Request deadline exceeded", body = ErrorEnvelope),
+ )
+)]
+pub(crate) async fn list_partitions(
+ State(state): State<RestState>,
+ Path((cluster, database, table)): Path<(String, String, String)>,
+ request: Request,
+) -> Response {
+ let request_id = request_id(&request);
+ let table = TablePath::new(database, table);
+ // The page token is scoped to the qualified table, so a token minted for one table cannot page
+ // another.
+ let prepared = prepare_page(
+ &state,
+ &request,
+ &cluster,
+ Collection::Partitions,
+ Some(&table.to_string()),
+ );
+ let result = async {
+ let (backend, page, ctx) = prepared?;
+ let (partitions, next_page_token) = page
+ .apply_by(backend.list_partitions(&ctx, &table).await?, |partition| {
+ partition.get_partition_name().into()
+ });
+ json_response(&PartitionsResponse {
+ partitions: partitions.iter().map(PartitionEntry::from).collect(),
+ next_page_token,
+ })
+ }
+ .await;
+ result.unwrap_or_else(|error| error_response(&error, &request_id))
+}
+
+/// Validates the cluster and page request before calling the backend.
+fn prepare_page(
state: &RestState,
request: &Request,
cluster: &str,
collection: Collection,
scope: Option<&str>,
) -> GatewayResult<(Arc<dyn FlussBackend>, Page, RequestContext)> {
- if !state.backend.has_cluster(cluster) {
- return Err(unknown_cluster(cluster));
- }
+ let (backend, ctx) = resolve_cluster(state, request, cluster)?;
let page = Page::parse(request.uri(), cluster, collection, scope)?;
- let cluster = ClusterId::try_from(cluster).expect("the backend serves this cluster ID");
- let ctx = request_context(cluster, request);
- Ok((state.backend.clone(), page, ctx))
+ Ok((backend, page, ctx))
}
#[cfg(test)]
mod tests {
use crate::backend::FlussBackend;
- use crate::backend::fake::FakeFlussBackend;
+ use crate::backend::fake::{FakeFlussBackend, Operation};
use crate::error::GatewayError;
use crate::protocol::rest::test_support;
use axum::body::Body;
- use axum::http::{Method, Request as HttpRequest, StatusCode};
+ use axum::http::{Request as HttpRequest, StatusCode};
+ use fluss::metadata::{
+ BigIntType, DataType, DecimalType, PartitionInfo, ResolvedPartitionSpec, Schema,
+ StringType, TableDescriptor, TableInfo, TablePath,
+ };
use http_body_util::BodyExt;
+ use std::collections::HashMap;
use std::sync::Arc;
- use std::time::Duration;
use tower::ServiceExt;
- /// A serving router over one configured cluster backed by the fixture.
fn app(backend: Arc<FakeFlussBackend>) -> axum::Router {
- app_with(backend, test_support::test_options())
- }
-
- fn app_with(
- backend: Arc<FakeFlussBackend>,
- options: crate::protocol::rest::RestOptions,
- ) -> axum::Router {
let state = test_support::state_with_backend(backend as Arc<dyn FlussBackend>);
state.readiness.set_serving();
- crate::protocol::rest::build_router(state, &options)
+ crate::protocol::rest::build_router(state, &test_support::test_options())
}
fn catalog() -> Arc<FakeFlussBackend> {
@@ -229,7 +449,6 @@ mod tests {
(status, serde_json::from_slice(&bytes).unwrap())
}
- /// Both collections answer in lexical order and omit the token on a complete page.
#[tokio::test]
async fn the_collections_answer_in_order_without_a_token() {
let app = app(catalog());
@@ -238,17 +457,19 @@ mod tests {
assert_eq!(status, StatusCode::OK);
assert_eq!(body, serde_json::json!({"databases": ["ops", "sales"]}));
- let (status, body) = get(&app, "/v1/clusters/default/databases/sales/tables").await;
+ let (status, body) = get(
+ &app,
+ "/v1/clusters/default/databases/sales/tables?max_results=2",
+ )
+ .await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body, serde_json::json!({"tables": ["customers", "orders"]}));
- // An empty database is an empty list, not a 404.
let (status, body) = get(&app, "/v1/clusters/default/databases/ops/tables").await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body, serde_json::json!({"tables": []}));
}
- /// A page hands out the token that continues it, and the token is scoped to its endpoint.
#[tokio::test]
async fn a_partial_page_carries_a_token_scoped_to_its_endpoint() {
let app = app(catalog());
@@ -268,7 +489,6 @@ mod tests {
assert_eq!(status, StatusCode::OK);
assert_eq!(body, serde_json::json!({"databases": ["sales"]}));
- // The same token on the tables endpoint is a bad request, never a reinterpretation.
let (status, body) = get(
&app,
&format!("/v1/clusters/default/databases/sales/tables?page_token={token}"),
@@ -278,14 +498,14 @@ mod tests {
assert_eq!(body["error"]["code"], "invalid_argument");
}
- /// An unconfigured or malformed cluster is `cluster_not_found`, and a page parameter is validated
- /// before the parent resource is looked up so a bad token cannot answer 404.
#[tokio::test]
- async fn unknown_clusters_and_bad_parameters_keep_their_own_status() {
- let app = app(catalog());
+ async fn metadata_errors_keep_their_own_status() {
+ let backend = catalog();
+ let app = app(Arc::clone(&backend));
for path in [
"/v1/clusters/other/databases",
+ "/v1/clusters/other/databases?max_results=0",
"/v1/clusters/Not%20A%20Cluster/databases",
"/v1/clusters/other/databases/sales/tables",
] {
@@ -296,9 +516,12 @@ mod tests {
for path in [
"/v1/clusters/default/databases?max_results=0",
+ "/v1/clusters/default/databases?max_results=1&max_results=2",
+ "/v1/clusters/%FF/databases",
"/v1/clusters/default/databases?page_token=nope!",
- // The database does not exist, but the token is what is wrong.
"/v1/clusters/default/databases/missing/tables?max_results=99999",
+ "/v1/clusters/default/databases/sales/tables/orders?page_token=x",
+ "/v1/clusters/other/databases/sales/tables/orders?page_token=x",
] {
let (status, body) = get(&app, path).await;
assert_eq!(status, StatusCode::BAD_REQUEST, "{path}");
@@ -308,63 +531,207 @@ mod tests {
let (status, body) = get(&app, "/v1/clusters/default/databases/missing/tables").await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(body["error"]["code"], "database_not_found");
- }
- /// Every status beyond the adapter's own 400 and 404 is the backend error, mapped mechanically.
- #[tokio::test]
- async fn backend_failures_are_mapped_without_the_adapter_deciding() {
- for (failure, status, code) in [
+ for (operation, path) in [
+ (Operation::ListDatabases, "/v1/clusters/default/databases"),
(
- GatewayError::unavailable("Fluss is unavailable"),
- StatusCode::SERVICE_UNAVAILABLE,
- "unavailable",
+ Operation::ListTables,
+ "/v1/clusters/default/databases/sales/tables",
),
(
- GatewayError::backend("Fluss denied the gateway"),
- StatusCode::INTERNAL_SERVER_ERROR,
- "backend",
+ Operation::DescribeTable,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ ),
+ (
+ Operation::ListPartitions,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions",
),
] {
- let backend = catalog();
- backend.fail_with(failure);
- let (answered, body) = get(&app(backend), "/v1/clusters/default/databases").await;
- assert_eq!(answered, status);
- assert_eq!(body["error"]["code"], code);
+ backend.fail_once(operation, GatewayError::unavailable("backend unavailable"));
+ let (status, body) = get(&app, &format!("{path}?unsupported=true")).await;
+ assert_eq!(status, StatusCode::BAD_REQUEST, "{operation:?}");
+ assert_eq!(body["error"]["code"], "invalid_argument");
+ let (status, body) = get(&app, path).await;
+ assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{operation:?}");
+ assert_eq!(body["error"]["code"], "unavailable");
+ assert_eq!(body["error"]["message"], "backend unavailable");
+ assert_eq!(get(&app, path).await.0, StatusCode::OK, "{operation:?}");
}
}
- /// A backend that outruns the request budget answers 504 from the handler's own deadline.
#[tokio::test]
- async fn a_slow_backend_answers_the_request_timeout() {
- let backend = catalog();
- backend.set_latency(Duration::from_secs(30));
- let options = crate::protocol::rest::RestOptions {
- request_timeout: Duration::from_millis(200),
- ..test_support::test_options()
- };
+ async fn a_table_is_described_in_the_shape_that_recreates_it() {
+ let backend = Arc::new(FakeFlussBackend::new());
+ let mut table = described_table();
+ backend.define_table(table.clone());
+ let app = app(Arc::clone(&backend));
+
+ let (status, body) = get(&app, "/v1/clusters/default/databases/sales/tables/orders").await;
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(
+ body,
+ serde_json::json!({
+ "database": "sales",
+ "table": "orders",
+ "columns": [
+ {"name": "id", "data_type": {"type": "BIGINT"}, "nullable": false},
+ {
+ "name": "amount",
+ "data_type": {"type": "DECIMAL", "precision": 18, "scale": 2},
+ "nullable": true,
+ "comment": "the order total",
+ },
+ {"name": "dt", "data_type": {"type": "STRING"}, "nullable": false},
+ ],
+ "primary_key": ["id", "dt"],
+ "partitioned_by": ["dt"],
+ "distribution": {"bucket_count": 4, "bucket_keys": ["id"]},
+ "comment": "the orders table",
+ "configs": {"table.log.ttl": "7d"},
+ "custom_properties": {"app.owner": "sales", "table.log.ttl": "custom value"},
+ })
+ );
+ table.properties.clear();
+ backend.define_table(table);
+ let (status, custom_only) =
+ get(&app, "/v1/clusters/default/databases/sales/tables/orders").await;
+ assert_eq!(status, StatusCode::OK);
+ assert!(custom_only.get("configs").is_none(), "{custom_only}");
+ assert_eq!(custom_only["custom_properties"], body["custom_properties"]);
+ }
+
+ #[tokio::test]
+ async fn a_log_table_omits_the_primary_key_fields() {
+ let backend = catalog();
let (status, body) = get(
- &app_with(backend, options),
- "/v1/clusters/default/databases",
+ &app(backend),
+ "/v1/clusters/default/databases/sales/tables/orders",
)
.await;
- assert_eq!(status, StatusCode::GATEWAY_TIMEOUT);
- assert_eq!(body["error"]["code"], "timeout");
+ assert_eq!(status, StatusCode::OK);
+ assert!(body.get("primary_key").is_none(), "{body}");
+ assert!(body.get("partitioned_by").is_none(), "{body}");
+ assert!(body.get("configs").is_none(), "{body}");
+ assert!(body.get("custom_properties").is_none(), "{body}");
}
#[tokio::test]
- async fn the_collections_are_read_only() {
+ async fn describing_what_does_not_exist_is_a_not_found() {
let app = app(catalog());
- let response = app
- .oneshot(
- HttpRequest::builder()
- .method(Method::DELETE)
- .uri("/v1/clusters/default/databases")
- .body(Body::empty())
- .unwrap(),
+
+ for (path, code) in [
+ (
+ "/v1/clusters/default/databases/absent/tables/orders",
+ "database_not_found",
+ ),
+ (
+ "/v1/clusters/default/databases/sales/tables/absent",
+ "table_not_found",
+ ),
+ (
+ "/v1/clusters/other/databases/sales/tables/orders",
+ "cluster_not_found",
+ ),
+ ] {
+ let (status, body) = get(&app, path).await;
+ assert_eq!(status, StatusCode::NOT_FOUND, "{path}");
+ assert_eq!(body["error"]["code"], code, "{path}");
+ }
+ }
+
+ #[tokio::test]
+ async fn partitions_page_in_name_order_under_a_table_scoped_token() {
+ let backend = Arc::new(FakeFlussBackend::new());
+ let table = TablePath::new("sales", "orders");
+ backend.define_table(described_table());
+ for name in ["2026-08-26", "2026-08-24", "2026-08-25"] {
+ let spec =
+ ResolvedPartitionSpec::new(Arc::from(["dt".to_string()]), vec![name.to_string()])
+ .expect("a fixture partition");
+ backend.define_partition(&table, PartitionInfo::new(1, spec));
+ }
+ backend.define_database("sales");
+ let mut updated = described_table();
+ updated.comment = Some("updated table".to_string());
+ backend.define_table(updated);
+ let app = app(backend);
+
+ let (status, body) = get(&app, "/v1/clusters/default/databases/sales/tables/orders").await;
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(body["comment"], "updated table");
+
+ let (status, body) = get(
+ &app,
+ "/v1/clusters/default/databases/sales/tables/orders/partitions?max_results=2",
+ )
+ .await;
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(
+ body["partitions"],
+ serde_json::json!([
+ {"name": "2026-08-24", "partition": {"dt": "2026-08-24"}},
+ {"name": "2026-08-25", "partition": {"dt": "2026-08-25"}},
+ ])
+ );
+
+ let token = body["next_page_token"].as_str().expect("a page token");
+ let (status, body) = get(
+ &app,
+ &format!(
+ "/v1/clusters/default/databases/sales/tables/orders/partitions?page_token={token}"
+ ),
+ )
+ .await;
+ assert_eq!(status, StatusCode::OK);
+ assert!(body.get("next_page_token").is_none(), "{body}");
+ assert_eq!(
+ body["partitions"],
+ serde_json::json!([{"name": "2026-08-26", "partition": {"dt": "2026-08-26"}}])
+ );
+
+ let (status, _) = get(
+ &app,
+ &format!("/v1/clusters/default/databases?page_token={token}"),
+ )
+ .await;
+ assert_eq!(status, StatusCode::BAD_REQUEST);
+ }
+
+ fn described_table() -> TableInfo {
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::with_nullable(false)))
+ .column(
+ "amount",
+ DataType::Decimal(DecimalType::with_nullable(true, 18, 2).unwrap()),
)
- .await
- .unwrap();
- assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
+ .with_comment("the order total")
+ .column("dt", DataType::String(StringType::with_nullable(false)))
+ .primary_key(["id", "dt"])
+ .build()
+ .expect("the described schema is valid");
+ let descriptor = TableDescriptor::builder()
+ .schema(schema)
+ .partitioned_by(vec!["dt".to_string()])
+ .distributed_by(Some(4), vec!["id".to_string()])
+ .comment("the orders table")
+ .properties(HashMap::from([(
+ "table.log.ttl".to_string(),
+ "7d".to_string(),
+ )]))
+ .custom_properties(HashMap::from([
+ ("app.owner".to_string(), "sales".to_string()),
+ ("table.log.ttl".to_string(), "custom value".to_string()),
+ ]))
+ .build()
+ .expect("the described descriptor is valid");
+ TableInfo::of(
+ TablePath::new("sales", "orders"),
+ 77,
+ 1,
+ descriptor,
+ 1_700_000_000_000,
+ 1_700_000_000_000,
+ )
}
}
diff --git a/fluss-gateway/src/protocol/rest/mod.rs b/fluss-gateway/src/protocol/rest/mod.rs
index 0a92e5674..259f59b2b 100644
--- a/fluss-gateway/src/protocol/rest/mod.rs
+++ b/fluss-gateway/src/protocol/rest/mod.rs
@@ -26,23 +26,25 @@
//! [`RequestContext`] carries into the backend.
pub mod clusters;
+pub mod datatype;
+pub mod ddl;
pub mod health;
pub mod metadata;
pub mod openapi;
pub mod pagination;
-use crate::backend::FlussBackend;
use crate::backend::context::RequestContext;
use crate::backend::types::ClusterId;
+use crate::backend::{FlussBackend, unknown_cluster};
#[cfg(test)]
use crate::config::REST_RESPONSE_GRACE;
use crate::config::{RestServerConfig, rest_handler_timeout};
-use crate::error::{ErrorEnvelope, ErrorKind, GatewayError, panic_message};
+use crate::error::{ErrorEnvelope, ErrorKind, GatewayError, GatewayResult, panic_message};
use crate::lifecycle::Readiness;
use crate::observability;
use axum::Router;
use axum::body::{Body, Bytes};
-use axum::extract::{DefaultBodyLimit, MatchedPath, Request};
+use axum::extract::{DefaultBodyLimit, FromRequest, MatchedPath, Request};
use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, Uri, header};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
@@ -141,7 +143,7 @@ struct ShapedResponse;
/// Renders the error envelope with the status its kind maps to, marks the response as already shaped, and adds
/// `Retry-After` where the taxonomy calls for it.
-pub fn error_response(error: &GatewayError, request_id: &RequestId) -> Response {
+fn error_response(error: &GatewayError, request_id: &RequestId) -> Response {
let status = StatusCode::from_u16(error.kind().http_status())
.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
let mut response =
@@ -157,12 +159,12 @@ pub fn error_response(error: &GatewayError, request_id: &RequestId) -> Response
}
/// Serializes `value` as a 200 JSON response. Fails only when serialization fails, which is reported as internal.
-pub fn json_response<T: Serialize>(value: &T) -> Result<Response, GatewayError> {
+fn json_response<T: Serialize>(value: &T) -> Result<Response, GatewayError> {
json_response_with_status(StatusCode::OK, value)
}
/// Serializes `value` as a JSON response with the given status. Serialization failures are reported as internal.
-pub(crate) fn json_response_with_status<T: Serialize>(
+fn json_response_with_status<T: Serialize>(
status: StatusCode,
value: &T,
) -> Result<Response, GatewayError> {
@@ -177,17 +179,22 @@ pub(crate) fn json_response_with_status<T: Serialize>(
Ok(response)
}
-/// Deserializes a JSON request body, requiring a JSON `Content-Type`.
-pub fn parse_json_body<T: DeserializeOwned>(
- headers: &HeaderMap,
- body: &Bytes,
-) -> Result<T, GatewayError> {
- validate_json_content_type(headers)?;
- serde_json::from_slice(body)
+/// Reads a size-limited request body and deserializes it as JSON.
+async fn parse_json_body<T: DeserializeOwned>(mut request: Request) -> Result<T, GatewayError> {
+ let headers = std::mem::take(request.headers_mut());
+ let body = Bytes::from_request(request, &()).await.map_err(|error| {
+ if error.status() == StatusCode::PAYLOAD_TOO_LARGE {
+ GatewayError::limit_exceeded("request body exceeds the configured limit")
+ } else {
+ GatewayError::invalid_argument(format!("unreadable request body: {error}"))
+ }
+ })?;
+ validate_json_content_type(&headers)?;
+ serde_json::from_slice(&body)
.map_err(|error| GatewayError::invalid_argument(format!("invalid JSON body: {error}")))
}
-pub(crate) fn validate_json_content_type(headers: &HeaderMap) -> Result<(), GatewayError> {
+fn validate_json_content_type(headers: &HeaderMap) -> Result<(), GatewayError> {
let Some(value) = headers.get(header::CONTENT_TYPE) else {
return Err(GatewayError::new(
crate::error::ErrorKind::UnsupportedMediaType,
@@ -217,7 +224,7 @@ pub(crate) fn validate_json_content_type(headers: &HeaderMap) -> Result<(), Gate
}
/// The request ID the outermost middleware assigned, for a handler rendering its own envelope.
-pub(crate) fn request_id(request: &Request) -> RequestId {
+fn request_id(request: &Request) -> RequestId {
request
.extensions()
.get::<RequestId>()
@@ -225,11 +232,24 @@ pub(crate) fn request_id(request: &Request) -> RequestId {
.unwrap_or_default()
}
+/// Resolves a configured cluster and builds its backend request context.
+fn resolve_cluster(
+ state: &RestState,
+ request: &Request,
+ cluster: &str,
+) -> GatewayResult<(Arc<dyn FlussBackend>, RequestContext)> {
+ if !state.backend.has_cluster(cluster) {
+ return Err(unknown_cluster(cluster));
+ }
+ let cluster = ClusterId::try_from(cluster).expect("the backend serves this cluster ID");
+ Ok((state.backend.clone(), request_context(cluster, request)))
+}
+
/// Builds the backend context of one cluster-scoped request from the middleware-assigned metadata.
///
/// The caller is anonymous until the authenticator lands; the deadline and the cancellation signal are
/// the ones the middleware also enforces, so a backend call cannot outlive its request.
-pub(crate) fn request_context(cluster_id: ClusterId, request: &Request) -> RequestContext {
+fn request_context(cluster_id: ClusterId, request: &Request) -> RequestContext {
let deadline = request
.extensions()
.get::<RequestDeadline>()
@@ -250,13 +270,6 @@ pub(crate) fn request_context(cluster_id: ClusterId, request: &Request) -> Reque
)
}
-/// Marks a response as final so the error-normalising middleware does not rewrite its body. Use it for handler
-/// responses that already carry their own envelope.
-pub fn shaped(mut response: Response) -> Response {
- response.extensions_mut().insert(ShapedResponse);
- response
-}
-
/// Assembles the REST frontend from the REST configuration and the shared process services.
///
/// This is the frontend's entry point for the lifecycle: it owns how the handler state is built and
@@ -291,6 +304,7 @@ pub fn build_router(state: RestState, options: &RestOptions) -> Router {
.split_for_parts();
let (metadata_router, metadata_api) = OpenApiRouter::new()
.merge(metadata::routes())
+ .merge(ddl::routes())
.split_for_parts();
let (open_router, open_api) = OpenApiRouter::new()
.merge(health::routes())
@@ -357,27 +371,16 @@ fn apply_acceptance_guard(router: Router, readiness: Arc<Readiness>) -> Router {
let readiness = readiness.clone();
async move {
if let Err(error) = readiness.ensure_accepting() {
- let request_id = request
- .extensions()
- .get::<RequestId>()
- .cloned()
- .unwrap_or_default();
- return error_response(&error, &request_id);
+ return error_response(&error, &request_id(&request));
}
next.run(request).await
}
}))
}
-/// Applies the cross-cutting middleware stack to an already-routed app.
-///
-/// Exposed separately so tests can wrap purpose-built routers with the production middleware.
-/// The body-limit layer is a streaming-body backstop. Requests with a declared length are rejected earlier with an
-/// envelope.
-///
-/// Order (outermost first): request-id assignment and error normalisation, then access logging, then the
-/// body size and deadline limits.
-pub fn apply_middleware(router: Router, options: &RestOptions) -> Router {
+/// Wraps a test router with the production middleware.
+#[cfg(test)]
+fn apply_middleware(router: Router, options: &RestOptions) -> Router {
apply_common_middleware(apply_data_limits(router, options), None)
}
@@ -416,11 +419,7 @@ fn apply_data_limits(router: Router, options: &RestOptions) -> Router {
let max_body_bytes = options.max_body_bytes;
let limits = move |mut request: Request, next: Next| async move {
- let request_id = request
- .extensions()
- .get::<RequestId>()
- .cloned()
- .unwrap_or_default();
+ let request_id = request_id(&request);
request
.extensions_mut()
.insert(deadline_from_now(request_timeout));
@@ -487,11 +486,7 @@ fn apply_common_middleware(router: Router, backend: Option<Arc<dyn FlussBackend>
/// response, where a dropped connection would have been recorded nowhere. The payload goes to the
/// log, never into the response, since it can carry internal detail.
async fn catch_panic(request: Request, next: Next) -> Response {
- let request_id = request
- .extensions()
- .get::<RequestId>()
- .cloned()
- .unwrap_or_default();
+ let request_id = request_id(&request);
match AssertUnwindSafe(next.run(request)).catch_unwind().await {
Ok(response) => response,
Err(payload) => {
@@ -526,11 +521,7 @@ fn request_log(
.to_string();
let cluster =
cluster_label(backend.as_deref(), &route, request.uri().path()).to_string();
- let request_id = request
- .extensions()
- .get::<RequestId>()
- .cloned()
- .unwrap_or_default();
+ let request_id = request_id(&request);
let response = next.run(request).await;
let elapsed = started.elapsed();
let status = response.status();
@@ -636,11 +627,7 @@ fn normalize_error(response: Response, request_id: &RequestId) -> Response {
}
async fn unknown_route(method: Method, uri: Uri, request: Request) -> Response {
- let request_id = request
- .extensions()
- .get::<RequestId>()
- .cloned()
- .unwrap_or_default();
+ let request_id = request_id(&request);
error_response(
&GatewayError::not_found(format!("no route for {method} {}", uri.path())),
&request_id,
@@ -690,7 +677,7 @@ pub mod test_support {
#[cfg(test)]
mod tests {
use super::*;
- use axum::routing::{get, post};
+ use axum::routing::get;
use http_body_util::BodyExt;
use serde::Deserialize;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -712,33 +699,48 @@ mod tests {
serde_json::from_slice(&bytes).expect("json body")
}
- #[test]
- fn shared_json_parser_enforces_media_type_and_serde_strictness() {
- let body = Bytes::from_static(br#"{"value": 7}"#);
- assert!(parse_json_body::<BodyFixture>(&HeaderMap::new(), &body).is_err());
-
- let mut headers = HeaderMap::new();
- headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("text/plain"));
- assert!(parse_json_body::<BodyFixture>(&headers, &body).is_err());
-
- headers.insert(
- header::CONTENT_TYPE,
- HeaderValue::from_static("application/vnd.fluss+json; charset=utf-8"),
- );
- assert_eq!(
- parse_json_body::<BodyFixture>(&headers, &body).unwrap(),
- BodyFixture { value: 7 }
- );
- for bad in [
- br#"{"value": 7, "unknown": true}"#.as_slice(),
- br#"{"value": 7, "value": 8}"#.as_slice(),
- br#"{"value":"#.as_slice(),
+ #[tokio::test]
+ async fn shared_json_parser_enforces_media_type_and_serde_strictness() {
+ for (content_type, expected) in [
+ (None, Err(ErrorKind::UnsupportedMediaType)),
+ (Some("text/plain"), Err(ErrorKind::UnsupportedMediaType)),
+ (Some("application/json"), Ok(BodyFixture { value: 7 })),
+ (
+ Some("application/vnd.fluss+json; charset=utf-8"),
+ Ok(BodyFixture { value: 7 }),
+ ),
] {
- assert!(
- parse_json_body::<BodyFixture>(&headers, &Bytes::copy_from_slice(bad)).is_err()
- );
+ let mut request = Request::builder();
+ if let Some(content_type) = content_type {
+ request = request.header(header::CONTENT_TYPE, content_type);
+ }
+ let request = request.body(Body::from(r#"{"value": 7}"#)).unwrap();
+ let result = parse_json_body::<BodyFixture>(request)
+ .await
+ .map_err(|error| error.kind());
+ assert_eq!(result, expected, "{content_type:?}");
+ }
+ for body in [
+ r#"{"value": 7, "unknown": true}"#,
+ r#"{"value": 7, "value": 8}"#,
+ r#"{"value":"#,
+ "",
+ ] {
+ let request = Request::builder()
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(Body::from(body))
+ .unwrap();
+ let error = parse_json_body::<BodyFixture>(request).await.unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::InvalidArgument, "{body}");
}
- assert!(parse_json_body::<BodyFixture>(&headers, &Bytes::new()).is_err());
+
+ let stream =
+ futures_util::stream::iter([Err::<Bytes, _>(std::io::Error::other("read failed"))]);
+ let error = parse_json_body::<BodyFixture>(Request::new(Body::from_stream(stream)))
+ .await
+ .unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::InvalidArgument);
+ assert!(error.message().starts_with("unreadable request body:"));
}
#[tokio::test]
@@ -771,26 +773,38 @@ mod tests {
#[tokio::test]
async fn oversized_body_yields_413_envelope() {
- let app = apply_middleware(
- Router::new().route("/echo", post(|| async { "ok" })),
- &test_support::test_options(),
- );
- let response = app
- .oneshot(
- Request::builder()
- .method(Method::POST)
- .uri("/echo")
- .header(header::CONTENT_LENGTH, "1048576")
- .body(Body::empty())
- .unwrap(),
- )
- .await
- .unwrap();
+ let state = test_support::test_state();
+ state.readiness.set_serving();
+ let app = build_router(state, &test_support::test_options());
+ for (method, path) in [
+ (Method::POST, "/v1/clusters/default/databases"),
+ (
+ Method::GET,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ ),
+ (
+ Method::DELETE,
+ "/v1/clusters/default/databases/sales/tables/orders",
+ ),
+ ] {
+ let response = app
+ .clone()
+ .oneshot(
+ Request::builder()
+ .method(method.clone())
+ .uri(path)
+ .header(header::CONTENT_LENGTH, "1048576")
+ .body(Body::empty())
+ .unwrap(),
+ )
+ .await
+ .unwrap();
- assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
- let json = body_json(response).await;
- assert_eq!(json["error"]["code"], "limit_exceeded");
- assert!(json["error"]["request_id"].as_str().is_some());
+ assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE, "{method}");
+ let json = body_json(response).await;
+ assert_eq!(json["error"]["code"], "limit_exceeded", "{method}");
+ assert!(json["error"]["request_id"].as_str().is_some(), "{method}");
+ }
}
#[tokio::test]
diff --git a/fluss-gateway/src/protocol/rest/openapi.rs b/fluss-gateway/src/protocol/rest/openapi.rs
index cfb223521..cd4011cfa 100644
--- a/fluss-gateway/src/protocol/rest/openapi.rs
+++ b/fluss-gateway/src/protocol/rest/openapi.rs
@@ -24,17 +24,18 @@
//! the taxonomy, so the contract cannot drift from the implementation either.
use crate::error::{ErrorCode, ErrorEnvelope};
+use crate::protocol::rest::datatype::{ColumnDataType, WireDataType, WireRowField};
use crate::protocol::rest::{RestState, json_response};
use axum::extract::State;
use axum::response::Response;
use serde_json::Value;
-use utoipa::{OpenApi, openapi::OpenApiBuilder};
+use utoipa::{OpenApi, ToSchema, openapi::OpenApiBuilder};
use utoipa_axum::router::OpenApiRouter;
use utoipa_axum::routes;
-/// Seeds the generated document with the shared error schemas, which no single handler owns.
+/// Registers shared schemas not collected from one handler.
#[derive(OpenApi)]
-#[openapi(components(schemas(ErrorCode, ErrorEnvelope)))]
+#[openapi(components(schemas(ErrorCode, ErrorEnvelope, WireDataType, WireRowField)))]
struct SharedSchemas;
/// OpenAPI routes, merged into the main router by [`crate::protocol::rest::build_router`].
@@ -42,7 +43,7 @@ pub fn routes() -> OpenApiRouter<RestState> {
OpenApiRouter::with_openapi(SharedSchemas::openapi()).routes(routes!(serve))
}
-/// Applies the gateway's own metadata to the router-generated document.
+/// Applies gateway metadata and schema constraints to the router-generated document.
///
/// Called once by [`crate::protocol::rest::build_router`].
pub(crate) fn finalize(api: utoipa::openapi::OpenApi) -> Value {
@@ -69,7 +70,17 @@ pub(crate) fn finalize(api: utoipa::openapi::OpenApi) -> Value {
// authentication capability adds securitySchemes and per-operation requirements.
.security(Some(Vec::new()))
.build();
- serde_json::to_value(api).expect("generated OpenAPI is serializable")
+ let mut document = serde_json::to_value(api).expect("generated OpenAPI is serializable");
+ // utoipa does not propagate deny_unknown_fields to internally tagged enum variants.
+ for name in [ColumnDataType::name(), WireDataType::name()] {
+ for variant in document["components"]["schemas"][name.as_ref()]["oneOf"]
+ .as_array_mut()
+ .expect("data type variants are generated")
+ {
+ variant["additionalProperties"] = Value::Bool(false);
+ }
+ }
+ document
}
/// Serves the generated OpenAPI 3.1 document as JSON.
@@ -123,7 +134,7 @@ mod tests {
serde_json::from_slice(&bytes).unwrap()
}
- /// The checked-in `openapi.yaml` next to this crate's `Cargo.toml` (FIP-49).
+ /// The checked-in `openapi.yaml` next to this crate's `Cargo.toml`.
fn checked_in_path() -> std::path::PathBuf {
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("openapi.yaml")
}
@@ -138,7 +149,7 @@ mod tests {
}
/// The checked-in document always matches the served one, so the published specification
- /// cannot drift from the implementation (FIP-49 schema-validation contract).
+ /// cannot drift from the implementation.
#[tokio::test]
async fn the_checked_in_document_matches_the_served_one() {
let checked_in = std::fs::read_to_string(checked_in_path())
@@ -192,6 +203,148 @@ mod tests {
"the envelope code refers to the generated vocabulary: {}",
document["components"]["schemas"]["ErrorBody"]
);
+
+ let schemas = &document["components"]["schemas"];
+ assert_eq!(
+ schemas["CreateDatabaseBody"]["required"],
+ serde_json::json!(["database"])
+ );
+ assert_eq!(
+ schemas["CreateTableBody"]["required"],
+ serde_json::json!(["table_name", "columns"])
+ );
+ assert!(
+ document["paths"]["/v1/clusters/{cluster}/databases/{database}"]
+ .get("get")
+ .is_none(),
+ "the API does not expose describeDatabase"
+ );
+ assert_eq!(
+ schemas["TableResponse"]["required"],
+ serde_json::json!(["database", "table", "columns"])
+ );
+ for (path, method, status) in [
+ (
+ "/v1/clusters/{cluster}/databases/{database}/tables",
+ "post",
+ "201",
+ ),
+ (
+ "/v1/clusters/{cluster}/databases/{database}/tables/{table}",
+ "patch",
+ "204",
+ ),
+ ] {
+ let response = &document["paths"][path][method]["responses"][status];
+ assert!(response.is_object(), "{method} {path}: {status}");
+ assert!(response.get("content").is_none(), "{method} {path}");
+ }
+ let table =
+ &document["paths"]["/v1/clusters/{cluster}/databases/{database}/tables/{table}"];
+ assert!(table["patch"]["responses"].get("200").is_none());
+ assert_eq!(
+ schemas["PartitionEntry"]["required"],
+ serde_json::json!(["name", "partition"])
+ );
+ assert_eq!(
+ schemas["PartitionsResponse"]["properties"]["partitions"]["items"]["$ref"],
+ "#/components/schemas/PartitionEntry"
+ );
+ for field in [
+ "primary_key",
+ "partitioned_by",
+ "distribution",
+ "configs",
+ "custom_properties",
+ "comment",
+ ] {
+ assert!(
+ !schemas["TableResponse"]["properties"][field]
+ .to_string()
+ .contains("\"null\""),
+ "an absent table field is omitted, not nullable: {field}"
+ );
+ }
+ for schema in ["CreateTableBody", "TableResponse"] {
+ for field in ["configs", "custom_properties"] {
+ let property = &schemas[schema]["properties"][field];
+ assert_eq!(property["type"], "object", "{schema}.{field}");
+ assert_eq!(
+ property["additionalProperties"]["type"], "string",
+ "{schema}.{field}"
+ );
+ }
+ }
+ for (name, has_nullable) in [("ColumnDataType", false), ("WireDataType", true)] {
+ for variant in schemas[name]["oneOf"].as_array().expect("type variants") {
+ assert_eq!(
+ variant["properties"].get("nullable").is_some(),
+ has_nullable,
+ "{name}: {variant}"
+ );
+ assert_eq!(
+ variant["additionalProperties"], false,
+ "type variants must reject undeclared fields: {name}: {variant}"
+ );
+ }
+ }
+
+ let mut pending = vec![&document];
+ while let Some(value) = pending.pop() {
+ match value {
+ Value::Object(object) => {
+ if let Some(reference) = object.get("$ref").and_then(Value::as_str)
+ && let Some(pointer) = reference.strip_prefix('#')
+ {
+ assert!(
+ document.pointer(pointer).is_some(),
+ "unresolved reference: {reference}"
+ );
+ }
+ pending.extend(object.values());
+ }
+ Value::Array(array) => pending.extend(array),
+ _ => {}
+ }
+ }
+
+ for (path, item) in document["paths"]
+ .as_object()
+ .expect("paths object")
+ .iter()
+ .filter(|(path, _)| path.contains("/databases"))
+ {
+ for method in ["get", "post", "patch", "delete"] {
+ let Some(operation) = item.get(method) else {
+ continue;
+ };
+ for status in ["400", "404", "413", "429", "500", "501", "503", "504"] {
+ assert!(
+ operation["responses"].get(status).is_some(),
+ "{method} {path} must declare {status}"
+ );
+ }
+ assert!(
+ operation["responses"].get("403").is_none(),
+ "{method} {path} must not declare caller authorization failures"
+ );
+ }
+ }
+
+ for (path, method) in [
+ ("/v1/clusters/{cluster}/databases", "post"),
+ ("/v1/clusters/{cluster}/databases/{database}/tables", "post"),
+ (
+ "/v1/clusters/{cluster}/databases/{database}/tables/{table}/partitions",
+ "post",
+ ),
+ ] {
+ assert!(
+ document["paths"][path][method]["responses"]["201"]["headers"]["Location"]
+ .is_object(),
+ "{method} {path} must declare its Location header"
+ );
+ }
}
/// The document route follows the same strict query policy as every other endpoint.
diff --git a/fluss-gateway/src/protocol/rest/pagination.rs b/fluss-gateway/src/protocol/rest/pagination.rs
index 2e7de790f..d390fd598 100644
--- a/fluss-gateway/src/protocol/rest/pagination.rs
+++ b/fluss-gateway/src/protocol/rest/pagination.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Keyset pagination for the FIP-49 list endpoints.
+//! Keyset pagination for catalog list endpoints.
//!
//! The page token is self-describing: it carries the cluster and collection it belongs to, its scope,
//! and the last name of the page it follows, base64url-encoded. Nothing is stored server-side, so any
@@ -36,9 +36,9 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
-/// Page size when the request does not ask for one (FIP-49).
+/// Default page size.
const DEFAULT_MAX_RESULTS: usize = 100;
-/// Largest page the gateway serves (FIP-49). Not configurable: it bounds one response body.
+/// Maximum page size, bounding one response body.
const MAX_MAX_RESULTS: usize = 1000;
/// Current token layout. A token from another version is rejected, never reinterpreted.
const TOKEN_VERSION: u32 = 1;
@@ -48,6 +48,7 @@ const TOKEN_VERSION: u32 = 1;
pub enum Collection {
Databases,
Tables,
+ Partitions,
}
impl Collection {
@@ -55,6 +56,7 @@ impl Collection {
match self {
Self::Databases => "databases",
Self::Tables => "tables",
+ Self::Partitions => "partitions",
}
}
}
@@ -90,7 +92,7 @@ impl Page {
max_results = value
.parse()
.ok()
- .filter(is_supported_page_size)
+ .filter(|value| (1..=MAX_MAX_RESULTS).contains(value))
.ok_or_else(|| {
GatewayError::invalid_argument(format!(
"max_results must be between 1 and {MAX_MAX_RESULTS}"
@@ -120,27 +122,39 @@ impl Page {
///
/// The gateway sorts by UTF-8 byte order, because Fluss does not promise an order and keyset
/// pagination needs one.
- pub fn apply(&self, mut names: Vec<String>) -> (Vec<String>, Option<String>) {
- names.sort_unstable();
+ pub fn apply(&self, names: Vec<String>) -> (Vec<String>, Option<String>) {
+ self.apply_by(names, |name| Cow::Borrowed(name))
+ }
+
+ /// Paginates entries by a borrowed or computed name.
+ pub fn apply_by<T>(
+ &self,
+ mut entries: Vec<T>,
+ key: impl Fn(&T) -> Cow<'_, str>,
+ ) -> (Vec<T>, Option<String>) {
+ entries.sort_unstable_by(|left, right| key(left).cmp(&key(right)));
let start = match &self.after {
- Some(after) => names.partition_point(|name| name.as_str() <= after.as_str()),
+ Some(after) => entries.partition_point(|entry| key(entry).as_ref() <= after.as_str()),
None => 0,
};
- let remaining = &names[start..];
- let has_more = remaining.len() > self.max_results;
- let page: Vec<String> = remaining.iter().take(self.max_results).cloned().collect();
- let next = page
- .last()
- .filter(|_| has_more)
- .map(|last| encode_token(&self.cluster, self.collection, self.scope.as_deref(), last));
+ let has_more = entries.len() - start > self.max_results;
+ let page: Vec<T> = entries
+ .into_iter()
+ .skip(start)
+ .take(self.max_results)
+ .collect();
+ let next = page.last().filter(|_| has_more).map(|last| {
+ encode_token(
+ &self.cluster,
+ self.collection,
+ self.scope.as_deref(),
+ &key(last),
+ )
+ });
(page, next)
}
}
-fn is_supported_page_size(value: &usize) -> bool {
- (1..=MAX_MAX_RESULTS).contains(value)
-}
-
/// The token payload. Compact field names keep the encoded token short.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
@@ -317,7 +331,7 @@ mod tests {
}
}
- /// The FIP-49 page-size contract, and the parameters the endpoints do not define.
+ /// Page-size bounds and unsupported query parameters.
#[test]
fn the_page_size_is_bounded_and_unknown_parameters_are_refused() {
let parse = |query: &str| Page::parse(&uri(query), "default", Collection::Databases, None);
diff --git a/fluss-gateway/tests/e2e_cluster.rs b/fluss-gateway/tests/e2e_cluster.rs
index fc5653fe8..e07f9c4cf 100644
--- a/fluss-gateway/tests/e2e_cluster.rs
+++ b/fluss-gateway/tests/e2e_cluster.rs
@@ -15,11 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! End-to-end metadata API tests against a real Dockerized Fluss cluster.
-//!
-//! The test crosses every production boundary: a real HTTP client calls the compiled Gateway binary,
-//! which reaches the Docker cluster through the native `fluss-rs` backend. Run it with
-//! `just test-e2e`.
+//! End-to-end Gateway tests against a Dockerized Fluss cluster (`just test-e2e`).
mod support;
@@ -34,6 +30,7 @@ use support::{Api, ChildGuard, await_http_ok, binary, free_port};
const DATABASE: &str = "gateway_e2e";
const TABLE: &str = "events";
+const JOURNEY_DATABASE: &str = "gateway_e2e_journey";
const SERVICE_ACCOUNT: &str = "admin";
const SERVICE_SECRET: &str = "admin-secret";
@@ -187,7 +184,11 @@ fn write_gateway_config(
path
}
-async fn assert_metadata_apis(bootstrap_servers: &str, security_protocol: &str) {
+async fn assert_metadata_apis(
+ bootstrap_servers: &str,
+ security_protocol: &str,
+ catalog_journey: bool,
+) {
let gateway_port = free_port();
let directory = tempfile::tempdir().expect("temporary Gateway config directory");
let config = write_gateway_config(
@@ -228,12 +229,184 @@ async fn assert_metadata_apis(bootstrap_servers: &str, security_protocol: &str)
.await,
serde_json::json!({"tables": [TABLE]})
);
+ if catalog_journey {
+ assert_catalog_journey(&api).await;
+ }
gateway.send_sigterm();
let status = gateway.wait_for_exit(Duration::from_secs(35)).await;
assert_eq!(status.code(), Some(0), "Gateway drains cleanly");
}
+/// Drives the catalog life cycle using only the REST API.
+async fn assert_catalog_journey(api: &Api) {
+ let databases = "/v1/clusters/default/databases";
+ let database = format!("{databases}/{JOURNEY_DATABASE}");
+ let tables = format!("{database}/tables");
+ let table = format!("{tables}/orders");
+ let partitions = format!("{table}/partitions");
+
+ let (location, created) = api
+ .post_created(
+ databases,
+ &serde_json::json!({"database": JOURNEY_DATABASE}),
+ )
+ .await;
+ assert_eq!(location, database);
+ assert_eq!(created, serde_json::json!({"database": JOURNEY_DATABASE}));
+
+ // A dry run reaches Fluss for nothing, so the table is still absent afterwards.
+ let definition = serde_json::json!({
+ "table_name": "orders",
+ "columns": [
+ {"name": "id", "data_type": {"type": "BIGINT"}, "nullable": false},
+ {"name": "dt", "data_type": {"type": "STRING"}, "nullable": false},
+ {"name": "amount", "data_type": {"type": "DECIMAL", "precision": 18, "scale": 2}},
+ ],
+ "primary_key": ["id", "dt"],
+ "partitioned_by": ["dt"],
+ "distribution": {"bucket_count": 2, "bucket_keys": ["id"]},
+ "configs": {"table.log.ttl": "7d"},
+ "custom_properties": {"app.owner": "sales", "app.stage": "draft"},
+ "comment": "the E2E journey table",
+ });
+ let mut dry_run = definition.clone();
+ dry_run["validate_only"] = serde_json::json!(true);
+ let (status, _, summary) = api.post(&tables, &dry_run).await;
+ assert_eq!(status, 200, "the dry run is accepted: {summary}");
+ assert_eq!(summary["validate_only"], true);
+ assert_eq!(
+ api.get(&table).await.status(),
+ 404,
+ "the dry run created nothing"
+ );
+
+ for key in ["nope.key", "table.nope"] {
+ let mut invalid = definition.clone();
+ invalid["configs"][key] = serde_json::json!("1");
+ let (status, _, error) = api.post(&tables, &invalid).await;
+ assert_eq!(status, 400, "{key}: {error}");
+ assert_eq!(error["error"]["code"], "invalid_argument");
+ }
+
+ let (location, body) = api.post_created(&tables, &definition).await;
+ assert_eq!(location, table);
+ assert_eq!(body, serde_json::Value::Null);
+ let created = api.get_ok(&table).await;
+ assert_eq!(created["distribution"]["bucket_count"], 2);
+ assert_eq!(created["configs"]["table.log.ttl"], "7d");
+ assert_eq!(
+ created["custom_properties"],
+ definition["custom_properties"]
+ );
+ // The declared type survives the round trip through Fluss's own schema serialization.
+ assert_eq!(
+ created["columns"][2]["data_type"],
+ serde_json::json!({"type": "DECIMAL", "precision": 18, "scale": 2})
+ );
+ let (status, _, error) = api
+ .patch(
+ &table,
+ &serde_json::json!({
+ "changes": [{"kind": "set_config", "key": "table.nope", "value": "1"}]
+ }),
+ )
+ .await;
+ assert_eq!(status, 400, "{error}");
+ assert_eq!(error["error"]["code"], "invalid_argument");
+ assert_eq!(api.get_ok(&table).await, created);
+
+ // Fluss currently requires schema and config changes in separate requests.
+ let (status, _, body) = api
+ .patch(
+ &table,
+ &serde_json::json!({
+ "changes": [
+ {"kind": "add_column", "name": "note", "data_type": {"type": "STRING"}}
+ ]
+ }),
+ )
+ .await;
+ assert_eq!(status, 204, "the alteration is applied: {body}");
+ assert_eq!(body, serde_json::Value::Null);
+ let altered = api.get_ok(&table).await;
+ assert_eq!(altered["columns"][3]["name"], "note");
+
+ for (changes, expected) in [
+ (
+ serde_json::json!([
+ {"kind": "set_config", "key": "app.owner", "value": "finance"},
+ {"kind": "reset_config", "key": "app.stage"}
+ ]),
+ Some(serde_json::json!({"app.owner": "finance"})),
+ ),
+ (
+ serde_json::json!([{"kind": "reset_config", "key": "app.owner"}]),
+ None,
+ ),
+ ] {
+ let (status, _, body) = api
+ .patch(&table, &serde_json::json!({"changes": changes}))
+ .await;
+ assert_eq!(status, 204, "{body}");
+ assert_eq!(body, serde_json::Value::Null);
+ let altered = api.get_ok(&table).await;
+ assert_eq!(altered["configs"], created["configs"]);
+ assert_eq!(altered.get("custom_properties"), expected.as_ref());
+ }
+
+ let (status, _, body) = api
+ .patch(
+ &table,
+ &serde_json::json!({
+ "changes": [
+ {"kind": "set_config", "key": "table.log.ttl", "value": "30d"}
+ ]
+ }),
+ )
+ .await;
+ assert_eq!(status, 204, "the alteration is applied: {body}");
+ assert_eq!(body, serde_json::Value::Null);
+ let altered = api.get_ok(&table).await;
+ assert_eq!(altered["configs"]["table.log.ttl"], "30d");
+
+ let (location, partition) = api
+ .post_created(
+ &partitions,
+ &serde_json::json!({"partition": {"dt": "2026-08-25"}}),
+ )
+ .await;
+ assert_eq!(location, format!("{partitions}/2026-08-25"));
+ assert_eq!(
+ partition["partition"],
+ serde_json::json!({"dt": "2026-08-25"})
+ );
+ let listed = api.get_ok(&partitions).await;
+ assert_eq!(
+ listed,
+ serde_json::json!({
+ "partitions": [{"name": "2026-08-25", "partition": {"dt": "2026-08-25"}}]
+ })
+ );
+
+ // Drop without cascade requires leaf-to-root cleanup.
+ assert_eq!(
+ api.delete(&database).await,
+ 409,
+ "the database is not empty"
+ );
+ let name = listed["partitions"][0]["name"].as_str().unwrap();
+ assert_eq!(api.delete(&format!("{partitions}/{name}")).await, 204);
+ assert_eq!(api.delete(&table).await, 204);
+ assert_eq!(api.delete(&database).await, 204);
+ assert!(
+ !api.get_ok(databases).await["databases"]
+ .as_array()
+ .expect("database list")
+ .contains(&serde_json::json!(JOURNEY_DATABASE))
+ );
+}
+
#[tokio::test]
async fn metadata_apis_support_plaintext_and_sasl_fluss_clusters() {
let cluster = tokio::task::spawn_blocking(|| FlussCluster::start(free_cluster_port_pair()))
@@ -270,8 +443,8 @@ async fn metadata_apis_support_plaintext_and_sasl_fluss_clusters() {
.await
.expect("create the E2E table");
- assert_metadata_apis(&cluster.plaintext_bootstrap_servers, "plaintext").await;
- assert_metadata_apis(&cluster.sasl_bootstrap_servers, "sasl").await;
+ assert_metadata_apis(&cluster.plaintext_bootstrap_servers, "plaintext", false).await;
+ assert_metadata_apis(&cluster.sasl_bootstrap_servers, "sasl", true).await;
connection
.close(Duration::from_secs(10))
diff --git a/fluss-gateway/tests/http_api.rs b/fluss-gateway/tests/http_api.rs
index 6f07c9f9f..6ef32f71a 100644
--- a/fluss-gateway/tests/http_api.rs
+++ b/fluss-gateway/tests/http_api.rs
@@ -70,7 +70,7 @@ async fn an_unknown_route_returns_the_shared_error_envelope() {
assert_eq!(
body["error"].as_object().expect("error object").len(),
3,
- "the FIP-49 envelope carries code, message, and the correlating request id: {body}"
+ "the error envelope carries code, message, and the correlating request id: {body}"
);
gateway.shutdown().await.expect("clean shutdown");
diff --git a/fluss-gateway/tests/support/mod.rs b/fluss-gateway/tests/support/mod.rs
index 44b8c2b33..bee609f5a 100644
--- a/fluss-gateway/tests/support/mod.rs
+++ b/fluss-gateway/tests/support/mod.rs
@@ -67,6 +67,58 @@ impl Api {
assert_eq!(response.status(), 200, "GET {path}");
response.json().await.expect("JSON body")
}
+
+ pub async fn post(&self, path: &str, body: &Value) -> (u16, Option<String>, Value) {
+ self.send(self.client.post(self.url(path)).json(body), path)
+ .await
+ }
+
+ pub async fn patch(&self, path: &str, body: &Value) -> (u16, Option<String>, Value) {
+ self.send(self.client.patch(self.url(path)).json(body), path)
+ .await
+ }
+
+ pub async fn delete(&self, path: &str) -> u16 {
+ self.send(self.client.delete(self.url(path)), path).await.0
+ }
+
+ pub async fn post_created(&self, path: &str, body: &Value) -> (String, Value) {
+ let (status, location, answered) = self.post(path, body).await;
+ assert_eq!(status, 201, "POST {path}: {answered}");
+ (
+ location.unwrap_or_else(|| panic!("POST {path} answers a Location header")),
+ answered,
+ )
+ }
+
+ /// Reports an empty body as JSON null in test results.
+ async fn send(
+ &self,
+ request: reqwest::RequestBuilder,
+ path: &str,
+ ) -> (u16, Option<String>, Value) {
+ let response = request.send().await.expect("request");
+ let status = response.status().as_u16();
+ let location = response
+ .headers()
+ .get(reqwest::header::LOCATION)
+ .map(|value| {
+ value
+ .to_str()
+ .unwrap_or_else(|_| panic!("{path} answers an ASCII Location"))
+ .to_string()
+ });
+ let bytes = response.bytes().await.expect("response body");
+ let body = if bytes.is_empty() {
+ Value::Null
+ } else {
+ Value::Object(
+ serde_json::from_slice(&bytes)
+ .unwrap_or_else(|error| panic!("{path} answers a JSON object: {error}")),
+ )
+ };
+ (status, location, body)
+ }
}
/// Starts an in-process gateway over `lifecycle::start` with an ephemeral port and no metrics listener.