Skip to content

Commit 4cb5e43

Browse files
luoyuxialuoyuxia
committed
feat: introduce catalog api
1 parent 3296926 commit 4cb5e43

File tree

5 files changed

+256
-2
lines changed

5 files changed

+256
-2
lines changed

‎crates/paimon/src/catalog/mod.rs‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
//! Catalog API for Apache Paimon.
19+
//!
20+
//! Design aligns with [Paimon Java Catalog](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java)
21+
//! and follows API patterns from Apache Iceberg Rust.
22+
23+
use std::collections::HashMap;
24+
use std::fmt;
25+
26+
use serde::{Deserialize, Serialize};
27+
28+
/// Splitter for system table names (e.g. `table$snapshots`).
29+
#[allow(dead_code)]
30+
pub const SYSTEM_TABLE_SPLITTER: &str = "$";
31+
/// Prefix for branch in object name (e.g. `table$branch_foo`).
32+
#[allow(dead_code)]
33+
pub const SYSTEM_BRANCH_PREFIX: &str = "branch_";
34+
/// Default main branch name.
35+
#[allow(dead_code)]
36+
pub const DEFAULT_MAIN_BRANCH: &str = "main";
37+
/// Database value when the database is not known; [`Identifier::full_name`] returns only the object.
38+
pub const UNKNOWN_DATABASE: &str = "unknown";
39+
40+
// ======================= Identifier ===============================
41+
42+
/// Identifies a catalog object (e.g. a table) by database and object name.
43+
///
44+
/// Corresponds to [org.apache.paimon.catalog.Identifier](https://github.com/apache/paimon/blob/release-1.3/paimon-api/src/main/java/org/apache/paimon/catalog/Identifier.java).
45+
/// The object name may be a table name or a qualified name like `table$branch_foo` or
46+
/// `table$snapshots` for system tables.
47+
#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
48+
#[serde(rename_all = "camelCase")]
49+
pub struct Identifier {
50+
/// Database name.
51+
database: String,
52+
/// Object name (table name, or table$branch$system for system tables).
53+
object: String,
54+
}
55+
56+
impl Identifier {
57+
/// Create an identifier from database and object name.
58+
pub fn new(database: impl Into<String>, object: impl Into<String>) -> Self {
59+
Self {
60+
database: database.into(),
61+
object: object.into(),
62+
}
63+
}
64+
65+
/// Database name.
66+
pub fn database(&self) -> &str {
67+
&self.database
68+
}
69+
70+
/// Full object name (table name, or with branch/system suffix).
71+
pub fn object(&self) -> &str {
72+
&self.object
73+
}
74+
75+
/// Full name: when database is [`UNKNOWN_DATABASE`], returns only the object;
76+
/// otherwise returns `database.object`.
77+
pub fn full_name(&self) -> String {
78+
if self.database == UNKNOWN_DATABASE {
79+
self.object.clone()
80+
} else {
81+
format!("{}.{}", self.database, self.object)
82+
}
83+
}
84+
}
85+
86+
impl fmt::Display for Identifier {
87+
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88+
write!(f, "{}", self.full_name())
89+
}
90+
}
91+
92+
impl fmt::Debug for Identifier {
93+
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94+
f.debug_struct("Identifier")
95+
.field("database", &self.database)
96+
.field("object", &self.object)
97+
.finish()
98+
}
99+
}
100+
101+
// ======================= Catalog trait ===============================
102+
103+
use async_trait::async_trait;
104+
105+
use crate::spec::{Schema, SchemaChange};
106+
use crate::table::Table;
107+
use crate::Result;
108+
109+
/// Catalog API for reading and writing metadata (databases, tables) in Paimon.
110+
///
111+
/// Corresponds to [org.apache.paimon.catalog.Catalog](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java).
112+
#[async_trait]
113+
pub trait Catalog: Send + Sync {
114+
// ======================= database methods ===============================
115+
116+
/// List names of all databases in this catalog.
117+
///
118+
/// # Errors
119+
/// Implementations may return other errors (e.g. I/O or backend-specific).
120+
async fn list_databases(&self) -> Result<Vec<String>>;
121+
122+
/// Create a database.
123+
///
124+
/// * `ignore_if_exists` - if true, do nothing when the database already exists;
125+
/// if false, return [`crate::Error::DatabaseAlreadyExist`].
126+
///
127+
/// # Errors
128+
/// * [`crate::Error::DatabaseAlreadyExist`] - database already exists when `ignore_if_exists` is false.
129+
async fn create_database(
130+
&self,
131+
name: &str,
132+
ignore_if_exists: bool,
133+
properties: HashMap<String, String>,
134+
) -> Result<()>;
135+
136+
/// Drop a database.
137+
///
138+
/// * `ignore_if_not_exists` - if true, do nothing when the database does not exist.
139+
/// * `cascade` - if true, delete all tables in the database then delete the database;
140+
/// if false, return [`crate::Error::DatabaseNotEmpty`] when not empty.
141+
///
142+
/// # Errors
143+
/// * [`crate::Error::DatabaseNotExist`] - database does not exist when `ignore_if_not_exists` is false.
144+
/// * [`crate::Error::DatabaseNotEmpty`] - database is not empty when `cascade` is false.
145+
async fn drop_database(
146+
&self,
147+
name: &str,
148+
ignore_if_not_exists: bool,
149+
cascade: bool,
150+
) -> Result<()>;
151+
152+
// ======================= table methods ===============================
153+
154+
/// Get table metadata for the given identifier.
155+
///
156+
/// # Errors
157+
/// * [`crate::Error::DatabaseNotExist`] - database in identifier does not exist.
158+
/// * [`crate::Error::TableNotExist`] - table does not exist.
159+
async fn get_table(&self, identifier: &Identifier) -> Result<Table>;
160+
161+
/// List table names in a database. System tables are not listed.
162+
///
163+
/// # Errors
164+
/// * [`crate::Error::DatabaseNotExist`] - database does not exist.
165+
async fn list_tables(&self, database_name: &str) -> Result<Vec<String>>;
166+
167+
/// Create a table.
168+
///
169+
/// * `ignore_if_exists` - if true, do nothing when the table already exists;
170+
/// if false, return [`crate::Error::TableAlreadyExist`].
171+
///
172+
/// # Errors
173+
/// * [`crate::Error::DatabaseNotExist`] - database in identifier does not exist.
174+
/// * [`crate::Error::TableAlreadyExist`] - table already exists when `ignore_if_exists` is false.
175+
async fn create_table(
176+
&self,
177+
identifier: &Identifier,
178+
creation: Schema,
179+
ignore_if_exists: bool,
180+
) -> Result<()>;
181+
182+
/// Drop a table. System tables cannot be dropped.
183+
///
184+
/// # Errors
185+
/// * [`crate::Error::TableNotExist`] - table does not exist when `ignore_if_not_exists` is false.
186+
async fn drop_table(&self, identifier: &Identifier, ignore_if_not_exists: bool) -> Result<()>;
187+
188+
/// Rename a table.
189+
///
190+
/// # Errors
191+
/// * [`crate::Error::TableNotExist`] - source table does not exist when `ignore_if_not_exists` is false.
192+
/// * [`crate::Error::TableAlreadyExist`] - target table already exists.
193+
async fn rename_table(
194+
&self,
195+
from: &Identifier,
196+
to: &Identifier,
197+
ignore_if_not_exists: bool,
198+
) -> Result<()>;
199+
200+
/// Apply schema changes to a table.
201+
///
202+
/// # Errors
203+
/// * [`crate::Error::TableNotExist`] - table does not exist when `ignore_if_not_exists` is false.
204+
/// * [`crate::Error::ColumnAlreadyExist`] - adding a column that already exists.
205+
/// * [`crate::Error::ColumnNotExist`] - altering or dropping a column that does not exist.
206+
async fn alter_table(
207+
&self,
208+
identifier: &Identifier,
209+
changes: Vec<SchemaChange>,
210+
ignore_if_not_exists: bool,
211+
) -> Result<()>;
212+
}

‎crates/paimon/src/error.rs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,24 @@ pub enum Error {
6666
display("Paimon hitting invalid file index format: {}", message)
6767
)]
6868
FileIndexFormatInvalid { message: String },
69+
70+
// ======================= catalog errors ===============================
71+
#[snafu(display("Database {} already exists.", database))]
72+
DatabaseAlreadyExist { database: String },
73+
#[snafu(display("Database {} does not exist.", database))]
74+
DatabaseNotExist { database: String },
75+
#[snafu(display("Database {} is not empty.", database))]
76+
DatabaseNotEmpty { database: String },
77+
#[snafu(display("Table {} already exists.", full_name))]
78+
TableAlreadyExist { full_name: String },
79+
#[snafu(display("Table {} does not exist.", full_name))]
80+
TableNotExist { full_name: String },
81+
#[snafu(display("Column {} already exists in table {}.", column, full_name))]
82+
ColumnAlreadyExist { full_name: String, column: String },
83+
#[snafu(display("Column {} does not exist in table {}.", column, full_name))]
84+
ColumnNotExist { full_name: String, column: String },
85+
#[snafu(display("Invalid identifier: {}", message))]
86+
IdentifierInvalid { message: String },
6987
}
7088

7189
impl From<opendal::Error> for Error {

‎crates/paimon/src/lib.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ mod error;
1919
pub use error::Error;
2020
pub use error::Result;
2121

22+
pub mod catalog;
2223
pub mod file_index;
2324
pub mod io;
2425
pub mod spec;
26+
mod table;

‎crates/paimon/src/spec/schema.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,7 @@ pub const PARTITION_OPTION: &str = "partition";
111111

112112
/// Schema of a table (logical DDL schema).
113113
///
114-
/// Corresponds to [org.apache.paimon.schema.Schema](https://github.com/apache/paimon/blob/1.3/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java).
114+
/// Corresponds to [org.apache.paimon.schema.Schema](https://github.com/apache/paimon/blob/release-1.3/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java).
115115
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116116
#[serde(rename_all = "camelCase")]
117117
pub struct Schema {
@@ -124,7 +124,7 @@ pub struct Schema {
124124

125125
impl Schema {
126126
/// Build a schema with validation. Normalizes partition/primary keys from options if present.
127-
pub fn new(
127+
fn new(
128128
fields: Vec<DataField>,
129129
partition_keys: Vec<String>,
130130
primary_keys: Vec<String>,

‎crates/paimon/src/table.rs‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
//! Table API for Apache Paimon
19+
20+
/// Table represents a table in the catalog.
21+
#[derive(Debug, Clone)]
22+
pub struct Table {}

0 commit comments

Comments
 (0)