This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new ac28d65 feat(rest-server): add FileSystemCatalog-backed REST catalog
server (#371)
ac28d65 is described below
commit ac28d65898f7a04fe97d2c1161d1acf350685973
Author: chaoyang <[email protected]>
AuthorDate: Tue Jun 23 15:32:14 2026 +0800
feat(rest-server): add FileSystemCatalog-backed REST catalog server (#371)
---
.github/workflows/ci.yml | 10 +
Cargo.toml | 2 +-
crates/paimon-rest-server/Cargo.toml | 50 +++
crates/paimon-rest-server/README.md | 81 +++++
crates/paimon-rest-server/src/lib.rs | 566 +++++++++++++++++++++++++++++++++
crates/paimon-rest-server/src/main.rs | 63 ++++
crates/paimon-rest-server/tests/e2e.rs | 431 +++++++++++++++++++++++++
7 files changed, 1202 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 46c0a90..cffd549 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -90,6 +90,16 @@ jobs:
RUST_LOG: DEBUG
RUST_BACKTRACE: full
+ # The e2e tests round-trip through a real FileSystemCatalog, which has
+ # known Windows path-compatibility issues (the `paimon` filesystem
+ # catalog tests are likewise gated with `#[cfg(not(windows))]`).
+ - name: Test paimon-rest-server
+ if: runner.os != 'Windows'
+ run: cargo test -p paimon-rest-server --all-targets
+ env:
+ RUST_LOG: DEBUG
+ RUST_BACKTRACE: full
+
integration:
runs-on: ubuntu-latest
steps:
diff --git a/Cargo.toml b/Cargo.toml
index a669a2a..c72bf55 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,7 @@
[workspace]
resolver = "2"
-members = ["crates/paimon", "crates/integration_tests", "bindings/c",
"bindings/python", "crates/integrations/datafusion"]
+members = ["crates/paimon", "crates/paimon-rest-server",
"crates/integration_tests", "bindings/c", "bindings/python",
"crates/integrations/datafusion"]
[workspace.package]
version = "0.3.0"
diff --git a/crates/paimon-rest-server/Cargo.toml
b/crates/paimon-rest-server/Cargo.toml
new file mode 100644
index 0000000..238eacc
--- /dev/null
+++ b/crates/paimon-rest-server/Cargo.toml
@@ -0,0 +1,50 @@
+# 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.
+
+[package]
+name = "paimon-rest-server"
+version.workspace = true
+edition.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+rust-version.workspace = true
+description = "A FileSystemCatalog-backed Paimon REST catalog server for local
end-to-end testing."
+# Testing/dev tool: not published to crates.io.
+publish = false
+
+[lib]
+name = "paimon_rest_server"
+path = "src/lib.rs"
+
+[[bin]]
+name = "paimon-rest-server"
+path = "src/main.rs"
+
+[dependencies]
+paimon = { workspace = true }
+axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
+tokio = { version = "1.39.2", features = ["rt-multi-thread", "macros", "net",
"signal", "time"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1.0.120"
+async-trait = "0.1.81"
+
+[dev-dependencies]
+tempfile = "3"
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+futures = "0.3"
diff --git a/crates/paimon-rest-server/README.md
b/crates/paimon-rest-server/README.md
new file mode 100644
index 0000000..2b8a1ea
--- /dev/null
+++ b/crates/paimon-rest-server/README.md
@@ -0,0 +1,81 @@
+<!--
+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.
+-->
+
+# paimon-rest-server
+
+A Paimon REST catalog server backed by a real [`FileSystemCatalog`], for
+**local end-to-end testing** of the REST catalog client. It is a dev/testing
+tool and is **not published** to crates.io.
+
+Unlike the in-memory mock used in `paimon`'s own unit tests, this server maps
+the Paimon REST protocol onto a real `FileSystemCatalog`, so the client-side
+`RESTCatalog` can be exercised against actual on-disk metadata:
+
+- config + database/table metadata CRUD;
+- append write + commit (the commit endpoint persists the posted snapshot via
+ `SnapshotManager`) + read back;
+- column-level `alter table`.
+
+Because both the server and the client point at the **same** local warehouse,
+the client writes data files directly while the server persists the snapshot
+metadata it receives on the commit endpoint. The wire format mirrors Java
+Paimon, so the same warehouse can be round-tripped with a Java reader/writer.
+
+## Run the standalone server
+
+```bash
+REST_WAREHOUSE=/tmp/paimon-warehouse REST_HOST=127.0.0.1 REST_PORT=8080 \
+ cargo run -p paimon-rest-server
+```
+
+Environment variables: `REST_WAREHOUSE` (default `/tmp/paimon-warehouse`),
+`REST_HOST` (default `127.0.0.1`), `REST_PORT` (default `8080`),
+`REST_PREFIX` (default empty).
+
+## Use as a test fixture
+
+```rust,no_run
+# async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+let server = paimon_rest_server::FsRestCatalogServer::start("/tmp/paimon-wh",
"").await?;
+// Point a RESTCatalog client at `server.url()`.
+# Ok(()) }
+```
+
+See `tests/e2e.rs` for the full metadata / write-commit-read / alter-table
+round trips.
+
+## Endpoints
+
+Served under the configured prefix (`/v1/...` by default):
+
+| Method | Path | Operation |
+| --- | --- | --- |
+| GET | `/v1/config` | server config |
+| GET / POST | `/databases` | list / create database |
+| GET / POST / DELETE | `/databases/{db}` | get / alter (no-op) / drop
database |
+| GET / POST | `/databases/{db}/tables` | list / create table |
+| GET / POST / DELETE | `/databases/{db}/tables/{table}` | get / alter / drop
table |
+| POST | `/tables/rename` | rename table |
+| POST | `/databases/{db}/tables/{table}/commit` | commit a snapshot |
+| GET | `/databases/{db}/tables/{table}/partitions` | list partitions |
+
+The data-token endpoint returns `501`; it is never called when
+`data-token.enabled=false` (the default).
+
+[`FileSystemCatalog`]: https://docs.rs/paimon
diff --git a/crates/paimon-rest-server/src/lib.rs
b/crates/paimon-rest-server/src/lib.rs
new file mode 100644
index 0000000..434f38d
--- /dev/null
+++ b/crates/paimon-rest-server/src/lib.rs
@@ -0,0 +1,566 @@
+// 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.
+
+//! A real Paimon REST catalog server backed by [`FileSystemCatalog`].
+//!
+//! Unlike the in-memory mock used by `paimon`'s own tests, this server maps
the
+//! Paimon REST protocol onto a real [`FileSystemCatalog`], so the client-side
+//! [`RESTCatalog`](paimon::catalog::RESTCatalog) can be exercised end to end:
+//! metadata CRUD, plus append write + commit + read back (the commit endpoint
+//! persists the posted snapshot via [`SnapshotManager`]).
+//!
+//! Because both the server and the client point at the *same* local warehouse,
+//! the client writes data files directly while the server only persists the
+//! snapshot metadata it receives on the commit endpoint.
+//!
+//! # Example
+//! ```no_run
+//! # async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+//! let server =
paimon_rest_server::FsRestCatalogServer::start("/tmp/paimon-wh", "").await?;
+//! println!("listening on {}", server.url());
+//! # Ok(()) }
+//! ```
+
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use axum::{
+ extract::{Extension, FromRequestParts, Json, MatchedPath, Query},
+ http::request::Parts,
+ http::StatusCode,
+ response::{IntoResponse, Response},
+ routing::{get, post},
+ serve, Router,
+};
+use serde::Deserialize;
+use serde_json::json;
+
+use paimon::api::{
+ AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
ConfigResponse, CreateTableRequest,
+ ErrorResponse, GetDatabaseResponse, GetTableResponse,
ListDatabasesResponse,
+ ListPartitionsResponse, ListTablesResponse, RESTUtil, RenameTableRequest,
ResourcePaths,
+};
+use paimon::catalog::{list_partitions_from_file_system, Catalog, Identifier};
+use paimon::common::{CatalogOptions, Options};
+use paimon::spec::{Schema, Snapshot};
+use paimon::table::SnapshotManager;
+use paimon::{Error, FileSystemCatalog};
+
+/// Convenience boxed error type for server construction (covers both
+/// [`paimon::Error`] from catalog setup and [`std::io::Error`] from binding).
+pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+/// Shared server state handed to every request handler.
+struct AppState {
+ catalog: FileSystemCatalog,
+ config: ConfigResponse,
+}
+
+/// A running FileSystemCatalog-backed REST catalog server.
+///
+/// The server runs on a background Tokio task and is aborted on drop, so a
+/// test can simply let it go out of scope when finished.
+pub struct FsRestCatalogServer {
+ addr: SocketAddr,
+ handle: tokio::task::JoinHandle<()>,
+}
+
+impl FsRestCatalogServer {
+ /// Start a server on an OS-assigned port (`127.0.0.1:0`).
+ ///
+ /// `warehouse` is the local warehouse root; `prefix` is the REST resource
+ /// prefix (pass `""` for none, mirroring `/v1/...`).
+ pub async fn start(warehouse: impl Into<String>, prefix: &str) ->
Result<Self, BoxError> {
+ Self::start_on(warehouse, prefix, "127.0.0.1", 0).await
+ }
+
+ /// Start a server bound to an explicit host and port.
+ pub async fn start_on(
+ warehouse: impl Into<String>,
+ prefix: &str,
+ host: &str,
+ port: u16,
+ ) -> Result<Self, BoxError> {
+ let warehouse = warehouse.into();
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse.clone());
+ let catalog = FileSystemCatalog::new(options)?;
+
+ // Advertise the prefix (and warehouse) to clients via GET /v1/config.
+ let mut defaults = HashMap::new();
+ if !prefix.is_empty() {
+ defaults.insert(CatalogOptions::PREFIX.to_string(),
prefix.to_string());
+ }
+ defaults.insert(CatalogOptions::WAREHOUSE.to_string(), warehouse);
+ let config = ConfigResponse::new(defaults);
+
+ let state = Arc::new(AppState { catalog, config });
+ let app = build_router(prefix, state);
+
+ let listener = tokio::net::TcpListener::bind((host, port)).await?;
+ let addr = listener.local_addr()?;
+ let handle = tokio::spawn(async move {
+ if let Err(e) = serve(listener, app.into_make_service()).await {
+ eprintln!("paimon-rest-server error: {e}");
+ }
+ });
+
+ Ok(Self { addr, handle })
+ }
+
+ /// The bound socket address.
+ pub fn addr(&self) -> SocketAddr {
+ self.addr
+ }
+
+ /// The base URL clients should connect to (e.g. `http://127.0.0.1:54321`).
+ pub fn url(&self) -> String {
+ format!("http://{}";, self.addr)
+ }
+}
+
+impl Drop for FsRestCatalogServer {
+ fn drop(&mut self) {
+ self.handle.abort();
+ }
+}
+
+/// Build the axum router, wiring every endpoint under the given prefix.
+fn build_router(prefix: &str, state: Arc<AppState>) -> Router {
+ let paths = ResourcePaths::new(prefix);
+ let base = paths.base_path();
+
+ Router::new()
+ // Config endpoint is always at the fixed /v1/config path.
+ .route("/v1/config", get(get_config))
+ .route(
+ &format!("{base}/databases"),
+ get(list_databases).post(create_database),
+ )
+ .route(
+ &format!("{base}/databases/:db"),
+ get(get_database).post(alter_database).delete(drop_database),
+ )
+ .route(
+ &format!("{base}/databases/:db/tables"),
+ get(list_tables).post(create_table),
+ )
+ .route(
+ &format!("{base}/databases/:db/tables/:table"),
+ get(get_table).post(alter_table).delete(drop_table),
+ )
+ .route(&format!("{base}/tables/rename"), post(rename_table))
+ .route(
+ &format!("{base}/databases/:db/tables/:table/commit"),
+ post(commit),
+ )
+ .route(
+ &format!("{base}/databases/:db/tables/:table/partitions"),
+ get(list_partitions),
+ )
+ // Token endpoint is never hit when `data-token.enabled=false`
(default),
+ // but we serve a stub so a misconfigured client gets a clear 501.
+ .route(
+ &format!("{base}/databases/:db/tables/:table/token"),
+ get(table_token_stub),
+ )
+ .layer(Extension(state))
+}
+
+// ============================================================================
+// Error mapping: paimon::Error -> (HTTP status, ErrorResponse)
+//
+// The client reconstructs the original error solely from the HTTP status code
+// (see `crates/paimon/src/api/rest_error.rs`), so the code below MUST line the
+// status codes up with `RestError::from_error_response`.
+// ============================================================================
+
+fn error_response(e: Error) -> Response {
+ let (status, resource_type, resource_name) = match &e {
+ Error::DatabaseNotExist { database } => (
+ StatusCode::NOT_FOUND,
+ Some("database".to_string()),
+ Some(database.clone()),
+ ),
+ Error::TableNotExist { full_name } => (
+ StatusCode::NOT_FOUND,
+ Some("table".to_string()),
+ Some(full_name.clone()),
+ ),
+ Error::DatabaseAlreadyExist { database } => (
+ StatusCode::CONFLICT,
+ Some("database".to_string()),
+ Some(database.clone()),
+ ),
+ Error::TableAlreadyExist { full_name } => (
+ StatusCode::CONFLICT,
+ Some("table".to_string()),
+ Some(full_name.clone()),
+ ),
+ Error::DatabaseNotEmpty { database } => (
+ StatusCode::CONFLICT,
+ Some("database".to_string()),
+ Some(database.clone()),
+ ),
+ Error::ColumnNotExist { full_name, column } => (
+ StatusCode::BAD_REQUEST,
+ Some(format!("column:{full_name}")),
+ Some(column.clone()),
+ ),
+ Error::ColumnAlreadyExist { full_name, column } => (
+ StatusCode::BAD_REQUEST,
+ Some(format!("column:{full_name}")),
+ Some(column.clone()),
+ ),
+ Error::IdentifierInvalid { .. } | Error::ConfigInvalid { .. } => {
+ (StatusCode::BAD_REQUEST, None, None)
+ }
+ Error::Unsupported { .. } => (StatusCode::NOT_IMPLEMENTED, None, None),
+ _ => (StatusCode::INTERNAL_SERVER_ERROR, None, None),
+ };
+
+ let body = ErrorResponse::new(
+ resource_type,
+ resource_name,
+ Some(e.to_string()),
+ Some(status.as_u16() as i32),
+ );
+ (status, Json(body)).into_response()
+}
+
+/// A 2xx response with an empty JSON body, for operations whose REST contract
+/// returns nothing meaningful (the client deserializes these as `Value`).
+fn ok_empty() -> Response {
+ (StatusCode::OK, Json(json!({}))).into_response()
+}
+
+// ============================================================================
+// Path parameter decoding
+//
+// The client (`ResourcePaths`) builds path segments with
`RESTUtil::encode_string`
+// (`application/x-www-form-urlencoded`), so e.g. a space becomes `+`. Axum's
own
+// `Path`/`RawPathParams` extractors percent-decode `%xx` but leave `+`
untouched,
+// which makes catalog names containing spaces unaddressable through
`RESTCatalog`.
+//
+// We therefore decode the *raw* (still percent-encoded) URI segments with the
+// same `RESTUtil` codec, mirroring Java's `RESTCatalogServer`, which calls
+// `RESTUtil.decodeString` on the raw segments. Decoding the raw segment
(rather
+// than post-processing Axum's already percent-decoded value) is the only way
to
+// recover names correctly for all inputs — a literal `+` (encoded as `%2B`)
and
+// a real space (encoded as `+`) are indistinguishable once `%xx` is decoded.
+// ============================================================================
+
+/// Path parameters captured from the matched route, decoded with the REST
codec.
+struct RestPath(HashMap<String, String>);
+
+impl RestPath {
+ /// The decoded value of a captured parameter (empty string if absent).
+ fn get(&self, key: &str) -> String {
+ self.0.get(key).cloned().unwrap_or_default()
+ }
+}
+
+#[async_trait::async_trait]
+impl<S: Send + Sync> FromRequestParts<S> for RestPath {
+ type Rejection = std::convert::Infallible;
+
+ async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self,
Self::Rejection> {
+ // The route pattern (e.g. `/v1/databases/:db/tables/:table`) is
recorded
+ // by Axum in the request extensions once a route matches.
+ let pattern = parts
+ .extensions
+ .get::<MatchedPath>()
+ .map(|m| m.as_str().to_string());
+ // `parts.uri.path()` is the original, still-percent-encoded request
path.
+ let raw_path = parts.uri.path().to_string();
+
+ let mut params = HashMap::new();
+ if let Some(pattern) = pattern {
+ for (pat_seg, raw_seg) in
pattern.split('/').zip(raw_path.split('/')) {
+ if let Some(name) = pat_seg.strip_prefix(':') {
+ params.insert(name.to_string(),
RESTUtil::decode_string(raw_seg));
+ }
+ }
+ }
+ Ok(RestPath(params))
+ }
+}
+
+// ============================================================================
+// Handlers
+// ============================================================================
+
+async fn get_config(
+ Query(_params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<AppState>>,
+) -> Response {
+ (StatusCode::OK, Json(state.config.clone())).into_response()
+}
+
+async fn list_databases(Extension(state): Extension<Arc<AppState>>) ->
Response {
+ match state.catalog.list_databases().await {
+ Ok(mut dbs) => {
+ dbs.sort();
+ (StatusCode::OK, Json(ListDatabasesResponse::new(dbs,
None))).into_response()
+ }
+ Err(e) => error_response(e),
+ }
+}
+
+async fn create_database(
+ Extension(state): Extension<Arc<AppState>>,
+ Json(payload): Json<paimon::api::CreateDatabaseRequest>,
+) -> Response {
+ match state
+ .catalog
+ .create_database(&payload.name, false, payload.options)
+ .await
+ {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn get_database(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -> Response {
+ let db = path.get("db");
+ match state.catalog.get_database(&db).await {
+ Ok(database) => {
+ let response = GetDatabaseResponse::new(
+ Some(database.name.clone()),
+ Some(database.name),
+ None,
+ database.options,
+ empty_audit(),
+ );
+ (StatusCode::OK, Json(response)).into_response()
+ }
+ Err(e) => error_response(e),
+ }
+}
+
+/// Alter database: `FileSystemCatalog` does not persist database properties,
+/// and the `Catalog` trait has no `alter_database`, so no client path reaches
+/// this. We only validate that the database exists and return OK; the request
+/// is intentionally a no-op.
+async fn alter_database(
+ path: RestPath,
+ Extension(state): Extension<Arc<AppState>>,
+ Json(_request): Json<AlterDatabaseRequest>,
+) -> Response {
+ let db = path.get("db");
+ match state.catalog.get_database(&db).await {
+ Ok(_) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn drop_database(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -> Response {
+ let db = path.get("db");
+ // The client (`RESTCatalog::drop_database`) already enforces the
non-cascade
+ // "database must be empty" check before issuing the DELETE, so the server
+ // force-drops with cascade=true.
+ match state.catalog.drop_database(&db, false, true).await {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn list_tables(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -> Response {
+ let db = path.get("db");
+ match state.catalog.list_tables(&db).await {
+ Ok(mut tables) => {
+ tables.sort();
+ (
+ StatusCode::OK,
+ Json(ListTablesResponse::new(Some(tables), None)),
+ )
+ .into_response()
+ }
+ Err(e) => error_response(e),
+ }
+}
+
+async fn create_table(
+ path: RestPath,
+ Extension(state): Extension<Arc<AppState>>,
+ Json(request): Json<CreateTableRequest>,
+) -> Response {
+ // Trust the path's database; take the table name from the request body.
+ let identifier = Identifier::new(path.get("db"),
request.identifier.object().to_string());
+ match state
+ .catalog
+ .create_table(&identifier, request.schema, false)
+ .await
+ {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn get_table(path: RestPath, Extension(state): Extension<Arc<AppState>>)
-> Response {
+ let table = path.get("table");
+ let identifier = Identifier::new(path.get("db"), table.clone());
+ let resolved = match state.catalog.get_table(&identifier).await {
+ Ok(t) => t,
+ Err(e) => return error_response(e),
+ };
+
+ let table_schema = resolved.schema();
+ // Convert the stored `TableSchema` into the DDL `Schema` the response
+ // carries. `Schema` is a field subset of `TableSchema` (both camelCase),
+ // and serde ignores the extra keys, preserving field ids exactly.
+ let schema: Schema =
+ match
serde_json::to_value(table_schema).and_then(serde_json::from_value::<Schema>) {
+ Ok(s) => s,
+ Err(e) => {
+ return error_response(Error::DataInvalid {
+ message: format!("Failed to convert table schema: {e}"),
+ source: Some(Box::new(e)),
+ })
+ }
+ };
+
+ let response = GetTableResponse::new(
+ // FileSystemCatalog has no UUID concept; the full name is a stable id
+ // that satisfies the client's RESTEnv requirement.
+ Some(identifier.full_name()),
+ Some(table),
+ Some(resolved.location().to_string()),
+ Some(false),
+ Some(table_schema.id()),
+ Some(schema),
+ empty_audit(),
+ );
+ (StatusCode::OK, Json(response)).into_response()
+}
+
+async fn drop_table(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -> Response {
+ let identifier = Identifier::new(path.get("db"), path.get("table"));
+ match state.catalog.drop_table(&identifier, false).await {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn alter_table(
+ path: RestPath,
+ Extension(state): Extension<Arc<AppState>>,
+ Json(request): Json<AlterTableRequest>,
+) -> Response {
+ let identifier = Identifier::new(path.get("db"), path.get("table"));
+ match state
+ .catalog
+ .alter_table(&identifier, request.changes, false)
+ .await
+ {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn rename_table(
+ Extension(state): Extension<Arc<AppState>>,
+ Json(request): Json<RenameTableRequest>,
+) -> Response {
+ match state
+ .catalog
+ .rename_table(&request.source, &request.destination, false)
+ .await
+ {
+ Ok(()) => ok_empty(),
+ Err(e) => error_response(e),
+ }
+}
+
+/// Request body posted by the client's `RESTSnapshotCommit` (see
+/// `crates/paimon/src/api/rest_api.rs::commit_snapshot`).
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct CommitRequest {
+ #[serde(default)]
+ table_uuid: Option<String>,
+ snapshot: Snapshot,
+ #[serde(default)]
+ statistics: serde_json::Value,
+}
+
+async fn commit(
+ path: RestPath,
+ Extension(state): Extension<Arc<AppState>>,
+ Json(request): Json<CommitRequest>,
+) -> Response {
+ let _ = (request.table_uuid, request.statistics);
+ let identifier = Identifier::new(path.get("db"), path.get("table"));
+
+ // Resolve the table's FileIO and on-disk location, then persist the posted
+ // snapshot exactly like the filesystem catalog's own commit path does.
+ let resolved = match state.catalog.get_table(&identifier).await {
+ Ok(t) => t,
+ Err(e) => return error_response(e),
+ };
+ let manager = SnapshotManager::new(resolved.file_io().clone(),
resolved.location().to_string());
+ match manager.commit_snapshot(&request.snapshot).await {
+ Ok(success) => (StatusCode::OK, Json(json!({ "success": success
}))).into_response(),
+ Err(e) => error_response(e),
+ }
+}
+
+/// List a table's partitions, computed from the latest snapshot on disk.
+///
+/// Mirrors `Catalog::list_partitions`: resolve the table, then derive
partition
+/// aggregates from the filesystem via [`list_partitions_from_file_system`].
The
+/// pagination params (`maxResults`/`pageToken`) are accepted but ignored — the
+/// whole set is returned in one page (`nextPageToken = null`).
+async fn list_partitions(
+ path: RestPath,
+ Query(_params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<AppState>>,
+) -> Response {
+ let identifier = Identifier::new(path.get("db"), path.get("table"));
+ let resolved = match state.catalog.get_table(&identifier).await {
+ Ok(t) => t,
+ Err(e) => return error_response(e),
+ };
+ match list_partitions_from_file_system(&resolved).await {
+ Ok(partitions) => (
+ StatusCode::OK,
+ Json(ListPartitionsResponse::new(Some(partitions), None)),
+ )
+ .into_response(),
+ Err(e) => error_response(e),
+ }
+}
+
+async fn table_token_stub() -> Response {
+ let body = ErrorResponse::new(
+ None,
+ None,
+ Some(
+ "Data token is not supported by paimon-rest-server; \
+ set data-token.enabled=false (the default)."
+ .to_string(),
+ ),
+ Some(StatusCode::NOT_IMPLEMENTED.as_u16() as i32),
+ );
+ (StatusCode::NOT_IMPLEMENTED, Json(body)).into_response()
+}
+
+fn empty_audit() -> AuditRESTResponse {
+ AuditRESTResponse::new(None, None, None, None, None)
+}
diff --git a/crates/paimon-rest-server/src/main.rs
b/crates/paimon-rest-server/src/main.rs
new file mode 100644
index 0000000..8a4cbfe
--- /dev/null
+++ b/crates/paimon-rest-server/src/main.rs
@@ -0,0 +1,63 @@
+// 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.
+
+//! Standalone Paimon REST catalog server backed by a local filesystem
warehouse.
+//!
+//! # Usage
+//! ```bash
+//! REST_WAREHOUSE=/tmp/paimon-warehouse REST_HOST=127.0.0.1 REST_PORT=8080 \
+//! cargo run -p paimon-rest-server
+//! ```
+//!
+//! Then point a client (e.g. the `rest_local_smoke` example, or Java) at it:
+//! ```bash
+//! REST_URI=http://localhost:8080 REST_WAREHOUSE=/tmp/paimon-warehouse \
+//! cargo run -p paimon --example rest_local_smoke
+//! ```
+
+use paimon_rest_server::{BoxError, FsRestCatalogServer};
+
+fn env_or(key: &str, default: &str) -> String {
+ std::env::var(key).unwrap_or_else(|_| default.to_string())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), BoxError> {
+ let warehouse = env_or("REST_WAREHOUSE", "/tmp/paimon-warehouse");
+ let host = env_or("REST_HOST", "127.0.0.1");
+ let port: u16 = env_or("REST_PORT", "8080")
+ .parse()
+ .map_err(|e| format!("invalid REST_PORT: {e}"))?;
+ let prefix = env_or("REST_PREFIX", "");
+
+ // Ensure the warehouse directory exists so FileIO can list it.
+ std::fs::create_dir_all(&warehouse)?;
+
+ let server = FsRestCatalogServer::start_on(warehouse.clone(), &prefix,
&host, port).await?;
+
+ println!("Paimon REST catalog server (filesystem-backed)");
+ println!(" warehouse : {warehouse}");
+ println!(" listening : {}", server.url());
+ if !prefix.is_empty() {
+ println!(" prefix : {prefix}");
+ }
+ println!("Press Ctrl-C to stop.");
+
+ tokio::signal::ctrl_c().await?;
+ println!("\nShutting down.");
+ Ok(())
+}
diff --git a/crates/paimon-rest-server/tests/e2e.rs
b/crates/paimon-rest-server/tests/e2e.rs
new file mode 100644
index 0000000..6edf274
--- /dev/null
+++ b/crates/paimon-rest-server/tests/e2e.rs
@@ -0,0 +1,431 @@
+// 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.
+
+//! End-to-end tests for the FileSystemCatalog-backed REST catalog server.
+//!
+//! Each test spins up a real [`FsRestCatalogServer`] over a temporary
warehouse
+//! and drives it through the client-side [`RESTCatalog`], covering metadata
CRUD
+//! and a full append write + commit + read-back round trip.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{Int32Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use futures::TryStreamExt;
+use tempfile::TempDir;
+
+use paimon::catalog::{Catalog, Identifier, RESTCatalog};
+use paimon::common::{CatalogOptions, Options};
+use paimon::spec::{BigIntType, DataType, IntType, Schema, SchemaChange,
VarCharType};
+
+use paimon_rest_server::FsRestCatalogServer;
+
+/// Holds the temp warehouse and the running server so they outlive the
catalog.
+struct TestContext {
+ _warehouse: TempDir,
+ _server: FsRestCatalogServer,
+ catalog: RESTCatalog,
+}
+
+async fn setup() -> TestContext {
+ let warehouse = TempDir::new().expect("create temp warehouse");
+ let warehouse_path = warehouse.path().to_str().unwrap().to_string();
+
+ let server = FsRestCatalogServer::start(warehouse_path.clone(), "")
+ .await
+ .expect("start server");
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::METASTORE, "rest");
+ options.set(CatalogOptions::URI, server.url());
+ options.set(CatalogOptions::WAREHOUSE, &warehouse_path);
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "dummy-token");
+
+ let catalog = RESTCatalog::new(options, true)
+ .await
+ .expect("create RESTCatalog");
+
+ TestContext {
+ _warehouse: warehouse,
+ _server: server,
+ catalog,
+ }
+}
+
+fn append_only_schema() -> Schema {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .build()
+ .expect("build schema")
+}
+
+fn sample_batch() -> RecordBatch {
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["alice", "bob", "carol"])),
+ ],
+ )
+ .expect("build batch")
+}
+
+fn partitioned_schema() -> Schema {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("region", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["region"])
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .build()
+ .expect("build schema")
+}
+
+fn partitioned_batch() -> RecordBatch {
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("region", ArrowDataType::Utf8, true),
+ ]));
+ RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["us", "eu", "us"])),
+ ],
+ )
+ .expect("build batch")
+}
+
+// ==================== Database metadata ====================
+
+#[tokio::test]
+async fn test_database_crud() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+
+ // Initially empty.
+ assert!(cat.list_databases().await.unwrap().is_empty());
+
+ // Create + list + get.
+ cat.create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let dbs = cat.list_databases().await.unwrap();
+ assert_eq!(dbs, vec!["db1".to_string()]);
+ let db = cat.get_database("db1").await.unwrap();
+ assert_eq!(db.name, "db1");
+
+ // Duplicate without ignore_if_exists -> error; with ignore -> ok.
+ assert!(cat
+ .create_database("db1", false, HashMap::new())
+ .await
+ .is_err());
+ cat.create_database("db1", true, HashMap::new())
+ .await
+ .unwrap();
+
+ // Get missing -> error.
+ assert!(cat.get_database("nope").await.is_err());
+
+ // Drop + verify; ignore_if_not_exists semantics.
+ cat.drop_database("db1", false, false).await.unwrap();
+ assert!(cat.list_databases().await.unwrap().is_empty());
+ assert!(cat.drop_database("db1", false, false).await.is_err());
+ cat.drop_database("db1", true, false).await.unwrap();
+}
+
+// ==================== Table metadata ====================
+
+#[tokio::test]
+async fn test_table_crud_and_rename() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+ cat.create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+
+ let users = Identifier::new("db", "users");
+ cat.create_table(&users, append_only_schema(), false)
+ .await
+ .unwrap();
+ assert_eq!(
+ cat.list_tables("db").await.unwrap(),
+ vec!["users".to_string()]
+ );
+
+ // Duplicate create errors unless ignored.
+ assert!(cat
+ .create_table(&users, append_only_schema(), false)
+ .await
+ .is_err());
+ cat.create_table(&users, append_only_schema(), true)
+ .await
+ .unwrap();
+
+ // get_table returns a usable table with the right schema/location.
+ let table = cat.get_table(&users).await.unwrap();
+ assert_eq!(table.schema().fields().len(), 2);
+ assert!(table.location().ends_with("db.db/users"));
+
+ // Rename round trip.
+ let renamed = Identifier::new("db", "users_renamed");
+ cat.rename_table(&users, &renamed, false).await.unwrap();
+ assert_eq!(
+ cat.list_tables("db").await.unwrap(),
+ vec!["users_renamed".to_string()]
+ );
+ assert!(cat.get_table(&users).await.is_err());
+ cat.rename_table(&renamed, &users, false).await.unwrap();
+
+ // Drop + missing semantics.
+ cat.drop_table(&users, false).await.unwrap();
+ assert!(cat.list_tables("db").await.unwrap().is_empty());
+ assert!(cat.drop_table(&users, false).await.is_err());
+ cat.drop_table(&users, true).await.unwrap();
+}
+
+#[tokio::test]
+async fn test_get_table_missing() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+ cat.create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+ assert!(cat
+ .get_table(&Identifier::new("db", "ghost"))
+ .await
+ .is_err());
+}
+
+// ==================== Full write + commit + read ====================
+
+#[tokio::test]
+async fn test_write_commit_read_roundtrip() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+
+ cat.create_database("smoke_db", false, HashMap::new())
+ .await
+ .unwrap();
+ let ident = Identifier::new("smoke_db", "users");
+ cat.create_table(&ident, append_only_schema(), false)
+ .await
+ .unwrap();
+
+ // Append write + commit through RESTSnapshotCommit (-> server commit
endpoint).
+ let table = cat.get_table(&ident).await.unwrap();
+ let write_builder = table.new_write_builder();
+ let mut writer = write_builder.new_write().unwrap();
+ writer.write_arrow_batch(&sample_batch()).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(!messages.is_empty(), "expected at least one commit message");
+ write_builder.new_commit().commit(messages).await.unwrap();
+
+ // Read back.
+ let table = cat.get_table(&ident).await.unwrap();
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let read = read_builder.new_read().unwrap();
+ let mut stream = read.to_arrow(plan.splits()).unwrap();
+
+ let mut total = 0usize;
+ while let Some(batch) = stream.try_next().await.unwrap() {
+ total += batch.num_rows();
+ }
+ assert_eq!(total, 3, "expected 3 rows read back, got {total}");
+}
+
+// ==================== alter table over REST ====================
+
+#[tokio::test]
+async fn test_alter_table_columns() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+
+ cat.create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+ let ident = Identifier::new("db", "events");
+ cat.create_table(&ident, append_only_schema(), false)
+ .await
+ .unwrap();
+
+ // Apply a batch of column changes through the REST alter_table path.
+ cat.alter_table(
+ &ident,
+ vec![
+ SchemaChange::add_column("age".to_string(),
DataType::Int(IntType::new())),
+ SchemaChange::rename_column("name".to_string(),
"full_name".to_string()),
+ SchemaChange::update_column_comment("id".to_string(), "the
id".to_string()),
+ SchemaChange::update_column_type(
+ "age".to_string(),
+ DataType::BigInt(BigIntType::new()),
+ ),
+ ],
+ false,
+ )
+ .await
+ .unwrap();
+
+ // The server persisted a new schema version; get_table reflects it.
+ let table = cat.get_table(&ident).await.unwrap();
+ let schema = table.schema();
+ let names: Vec<&str> = schema.fields().iter().map(|f| f.name()).collect();
+ assert_eq!(names, vec!["id", "full_name", "age"]);
+
+ let id_field = schema.fields().iter().find(|f| f.name() == "id").unwrap();
+ assert_eq!(id_field.description(), Some("the id"));
+ let age_field = schema.fields().iter().find(|f| f.name() ==
"age").unwrap();
+ assert!(matches!(age_field.data_type(), DataType::BigInt(_)));
+
+ // alter on a missing table: ignored vs. error.
+ let missing = Identifier::new("db", "nope");
+ cat.alter_table(
+ &missing,
+ vec![SchemaChange::update_column_comment(
+ "id".to_string(),
+ "x".to_string(),
+ )],
+ true,
+ )
+ .await
+ .unwrap();
+ assert!(cat
+ .alter_table(
+ &missing,
+ vec![SchemaChange::update_column_comment(
+ "id".to_string(),
+ "x".to_string(),
+ )],
+ false,
+ )
+ .await
+ .is_err());
+}
+
+// ==================== list partitions over REST ====================
+
+#[tokio::test]
+async fn test_list_partitions() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+
+ cat.create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+ let ident = Identifier::new("db", "events");
+ cat.create_table(&ident, partitioned_schema(), false)
+ .await
+ .unwrap();
+
+ // Write rows spanning two partitions (region=us has 2 rows, region=eu 1),
+ // then commit through the server's commit endpoint.
+ let table = cat.get_table(&ident).await.unwrap();
+ let write_builder = table.new_write_builder();
+ let mut writer = write_builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&partitioned_batch())
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(!messages.is_empty(), "expected at least one commit message");
+ write_builder.new_commit().commit(messages).await.unwrap();
+
+ // The partitions endpoint must serve the two partitions (no 404 fallback);
+ // the client receives them directly from the server.
+ let partitions = cat.list_partitions(&ident).await.unwrap();
+ assert_eq!(partitions.len(), 2, "expected two partitions");
+
+ let mut regions: Vec<String> = partitions
+ .iter()
+ .map(|p| p.spec.get("region").cloned().unwrap_or_default())
+ .collect();
+ regions.sort();
+ assert_eq!(regions, vec!["eu".to_string(), "us".to_string()]);
+
+ let total: i64 = partitions.iter().map(|p| p.record_count).sum();
+ assert_eq!(total, 3, "partition record counts should sum to all rows");
+
+ let us = partitions
+ .iter()
+ .find(|p| p.spec.get("region").map(String::as_str) == Some("us"))
+ .expect("us partition present");
+ assert_eq!(us.record_count, 2, "region=us holds two rows");
+}
+
+// ==================== Path codec round trip ====================
+
+/// Database/table names with characters the REST path codec encodes specially
+/// must survive the client encode -> server decode round trip.
+///
+/// The client builds path segments with `RESTUtil::encode_string`
+/// (`application/x-www-form-urlencoded`), so a space becomes `+` and a literal
+/// `+` becomes `%2B`. The server must decode them back to the exact original
+/// names; otherwise these databases/tables are unaddressable through
+/// `RESTCatalog`. A literal `+` additionally proves the server decodes the raw
+/// segment (not Axum's already percent-decoded value), since `+` and a space
+/// would otherwise be indistinguishable.
+#[tokio::test]
+async fn test_special_char_names() {
+ let ctx = setup().await;
+ let cat = &ctx.catalog;
+
+ // Two databases: one with a space, one with a literal `+`.
+ let space_db = "sales db";
+ let plus_db = "a+b";
+ cat.create_database(space_db, false, HashMap::new())
+ .await
+ .unwrap();
+ cat.create_database(plus_db, false, HashMap::new())
+ .await
+ .unwrap();
+
+ let mut dbs = cat.list_databases().await.unwrap();
+ dbs.sort();
+ assert_eq!(dbs, vec![plus_db.to_string(), space_db.to_string()]);
+
+ // get_database must address each one by its exact name.
+ assert_eq!(cat.get_database(space_db).await.unwrap().name, space_db);
+ assert_eq!(cat.get_database(plus_db).await.unwrap().name, plus_db);
+
+ // A table whose name also contains a space, under the space database.
+ let ident = Identifier::new(space_db, "my table");
+ cat.create_table(&ident, append_only_schema(), false)
+ .await
+ .unwrap();
+ assert_eq!(
+ cat.list_tables(space_db).await.unwrap(),
+ vec!["my table".to_string()]
+ );
+ let table = cat.get_table(&ident).await.unwrap();
+ assert_eq!(table.schema().fields().len(), 2);
+
+ // Drop the table and both databases by their exact names.
+ cat.drop_table(&ident, false).await.unwrap();
+ cat.drop_database(space_db, false, false).await.unwrap();
+ cat.drop_database(plus_db, false, false).await.unwrap();
+ assert!(cat.list_databases().await.unwrap().is_empty());
+}
