You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "beryllw (via GitHub)" <gi...@apache.org> on 2026/08/25 16:19:01 UTC
[PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
beryllw opened a new pull request, #4107:
URL: https://github.com/apache/fluss/pull/4107
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4082
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #4107:
URL: https://github.com/apache/fluss/pull/4107#issuecomment-5422635150
@fresh-borzoni @leonardBang @naivedogger cc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #4107:
URL: https://github.com/apache/fluss/pull/4107#discussion_r3864330538
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1349 @@
+// 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, TableResponse, resolve_cluster,
+};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, json_response_with_status, parse_json_body,
+ request_id,
+};
+use axum::extract::{Path, Request, State};
+use axum::http::{HeaderValue, StatusCode, header};
+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))
+}
+
+/// 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,
+ /// Defaults to true; fluss-rs makes primary-key columns non-nullable.
+ 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>,
+ #[serde(default)]
+ pub configs: 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.
+#[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,
+ },
+ 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(&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 created table", body = TableResponse,
+ 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 {
Review Comment:
Thanks for highlighting this. For now, `validate_only` is limited to checks provided by fluss-rust; it does not perform server-side validation or guarantee that creation will succeed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4107:
URL: https://github.com/apache/fluss/pull/4107#discussion_r3862519415
##########
fluss-gateway/src/backend/errors.rs:
##########
@@ -89,8 +107,19 @@ fn map_api_error(what: &str, api_error: FlussError) -> Option<GatewayError> {
"Fluss rejected the name 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::InvalidConfigException
Review Comment:
Four errors collapse into "Fluss rejected the definition" and none of these arms log.
The server had said "The option 'x' is not supported to alter yet." and I got a 400 with nothing in the log at debug.
How does an operator find out why? Should these log the native error like the transport arms do?
##########
fluss-gateway/src/backend/errors.rs:
##########
@@ -80,6 +70,34 @@ 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),
+ // Asking for a partition of a table that declares no partition keys, or supplying a spec that
+ // does not match the declared keys, is a malformed request rather than a missing resource.
+ FlussError::TableNotPartitionedException | FlussError::PartitionSpecInvalidException => {
Review Comment:
GET .../partitions on a non-partitioned table says "Fluss rejected the partition spec" and the request has no spec.
Shall we reword for the list case?
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1350 @@
+// 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, TableResponse};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, json_response_with_status, 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,
+ /// Defaults to true; fluss-rs makes primary-key columns non-nullable.
+ 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>,
+ #[serde(default)]
+ pub configs: 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.
+#[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,
+ },
+ 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(&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 created table", body = TableResponse,
+ 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(),
+ });
+ }
+ let created = backend.create_table(&ctx, &table, &descriptor).await?;
+ created_response(
+ &TableResponse::from(&created),
+ &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 = 200, description = "The altered table", body = TableResponse),
+ (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)?;
+ let altered = backend.alter_table(&ctx, &table, changes).await?;
+ json_response(&TableResponse::from(&altered))
+ }
+ .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(
+ &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)?;
+
+ let mut schema = Schema::builder();
+ for column in body.columns {
+ 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)
Review Comment:
Create puts everything into properties, so configs: {"nope.key":"1"} is 400. But PATCH set_config nope.key=1 returns 200 and the server keeps it in custom_properties. Why the difference? And describe only reads properties, so how does a caller ever read that value back? I set custom.thing=v, got 200, describe showed nothing.
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1350 @@
+// 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, TableResponse};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, json_response_with_status, 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,
+ /// Defaults to true; fluss-rs makes primary-key columns non-nullable.
+ 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>,
+ #[serde(default)]
+ pub configs: 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.
+#[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,
+ },
+ 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(&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 created table", body = TableResponse,
+ 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(),
+ });
+ }
+ let created = backend.create_table(&ctx, &table, &descriptor).await?;
+ created_response(
+ &TableResponse::from(&created),
+ &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 = 200, description = "The altered table", body = TableResponse),
+ (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)?;
+ let altered = backend.alter_table(&ctx, &table, changes).await?;
+ json_response(&TableResponse::from(&altered))
+ }
+ .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(
+ &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 `$`"),
Review Comment:
CreatePartitionBody.partition and both partition responses are named maps, but drop takes the values joined by $ in partition-key order and if we transpose them - it's 404 partition_not_found.
How is a client meant to know the order? Can drop take a spec too?
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1350 @@
+// 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, TableResponse};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, json_response_with_status, 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,
+ /// Defaults to true; fluss-rs makes primary-key columns non-nullable.
+ 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>,
+ #[serde(default)]
+ pub configs: 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.
+#[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,
+ },
+ 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(&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 created table", body = TableResponse,
+ 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(),
+ });
+ }
+ let created = backend.create_table(&ctx, &table, &descriptor).await?;
+ created_response(
+ &TableResponse::from(&created),
+ &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 = 200, description = "The altered table", body = TableResponse),
+ (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)?;
+ let altered = backend.alter_table(&ctx, &table, changes).await?;
+ json_response(&TableResponse::from(&altered))
+ }
+ .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(
+ &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)?;
+
+ let mut schema = Schema::builder();
+ for column in body.columns {
+ let data_type = column
+ .data_type
+ .0
+ .with_root_nullable(column.nullable.unwrap_or(true));
Review Comment:
A column with nullable: true named in primary_key returns 201 with nullable: false.
The FIP has that as 400, so should we reject it rather than change it quietly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4107:
URL: https://github.com/apache/fluss/pull/4107#discussion_r3862660463
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1349 @@
+// 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, TableResponse, resolve_cluster,
+};
+use crate::protocol::rest::{
+ RestState, error_response, json_response, json_response_with_status, parse_json_body,
+ request_id,
+};
+use axum::extract::{Path, Request, State};
+use axum::http::{HeaderValue, StatusCode, header};
+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))
+}
+
+/// 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,
+ /// Defaults to true; fluss-rs makes primary-key columns non-nullable.
+ 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>,
+ #[serde(default)]
+ pub configs: 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.
+#[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,
+ },
+ 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(&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 created table", body = TableResponse,
+ 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 {
Review Comment:
`validate_only` only exercises the fluss-rs builders, so nested ROW fields with blank or duplicate names return 200 here but are rejected by Fluss Server during actual creation.
##########
fluss-gateway/src/backend/client.rs:
##########
@@ -146,9 +127,102 @@ 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<TableInfo> {
+ self.admin_call(ctx, "create the table", |admin| async move {
+ admin.create_table(table, descriptor, false).await?;
+ admin.get_table_info(table).await
Review Comment:
`get_table_info` runs after the mutation has already succeeded, but its failure currently overwrites that success. A committed create can therefore return 503/504, and a retry then receives `already_exists`; `alter_table` has the same issue.
Could we avoid reporting a post-commit read failure as a mutation failure?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #4107:
URL: https://github.com/apache/fluss/pull/4107#discussion_r3864324451
##########
fluss-gateway/src/backend/client.rs:
##########
@@ -146,9 +127,102 @@ 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<TableInfo> {
+ self.admin_call(ctx, "create the table", |admin| async move {
+ admin.create_table(table, descriptor, false).await?;
+ admin.get_table_info(table).await
Review Comment:
Thanks for pointing this out. I’ve removed the post-mutation reads: create now returns `201` with a `Location` header, and alter returns `204`; metadata can be fetched separately via GET.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4107:
URL: https://github.com/apache/fluss/pull/4107#discussion_r3866360383
##########
fluss-gateway/src/backend/errors.rs:
##########
@@ -89,8 +107,19 @@ fn map_api_error(what: &str, api_error: FlussError) -> Option<GatewayError> {
"Fluss rejected the name 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::InvalidConfigException
Review Comment:
Logging looks good now, thanks. The other half is still there though: those five codes still share "Fluss rejected the definition", so a bad replication factor, too many buckets and an unalterable option all read the same.
Can each get its own message?
##########
fluss-gateway/src/protocol/rest/ddl.rs:
##########
@@ -0,0 +1,1402 @@
+// 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; avoid the `table.` prefix used by Fluss configs.
Review Comment:
The doc says avoid the table. prefix, but nothing stops it and once it's there, set_config table.custom and reset_config table.custom both 400, since table. keys go to Fluss properties.
Should we reject the prefix here instead of asking?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [gateway] Add catalog and DDL backend operations with REST APIs [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #4107:
URL: https://github.com/apache/fluss/pull/4107
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org