risingwave_connector/connector_common/iceberg/
mock_catalog.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16
17use async_trait::async_trait;
18use iceberg::io::FileIO;
19use iceberg::spec::{
20    NestedField, PrimitiveType, Schema, TableMetadataBuilder, Transform, Type,
21    UnboundPartitionField, UnboundPartitionSpec,
22};
23use iceberg::table::Table;
24use iceberg::{
25    Catalog as CatalogV2, Namespace, NamespaceIdent, Runtime, TableCommit, TableCreation,
26    TableIdent,
27};
28
29/// A mock catalog for iceberg used for plan test.
30#[derive(Debug)]
31pub struct MockCatalog;
32
33impl MockCatalog {
34    const RANGE_TABLE: &'static str = "range_table";
35    const SPARSE_TABLE: &'static str = "sparse_table";
36}
37
38impl MockCatalog {
39    fn build_table(name: &str, schema: Schema, partition_spec: UnboundPartitionSpec) -> Table {
40        let file_io = FileIO::new_with_memory();
41        let table_creation = TableCreation {
42            name: "ignore".to_owned(),
43            location: Some("1".to_owned()),
44            schema,
45            partition_spec: Some(partition_spec),
46            sort_order: None,
47            properties: HashMap::new(),
48            format_version: iceberg::spec::FormatVersion::V2,
49        };
50        Table::builder()
51            .identifier(TableIdent::new(
52                NamespaceIdent::new("mock_namespace".to_owned()),
53                name.to_owned(),
54            ))
55            .file_io(file_io)
56            .runtime(Runtime::try_current().unwrap())
57            .metadata(
58                TableMetadataBuilder::from_table_creation(table_creation)
59                    .unwrap()
60                    .build()
61                    .unwrap()
62                    .metadata,
63            )
64            .build()
65            .unwrap()
66    }
67
68    fn sparse_table() -> Table {
69        Self::build_table(
70            Self::SPARSE_TABLE,
71            Schema::builder()
72                .with_fields(vec![
73                    NestedField::new(1, "v1", Type::Primitive(PrimitiveType::Int), true).into(),
74                    NestedField::new(2, "v2", Type::Primitive(PrimitiveType::Long), true).into(),
75                    NestedField::new(3, "v3", Type::Primitive(PrimitiveType::String), true).into(),
76                    NestedField::new(4, "v4", Type::Primitive(PrimitiveType::Time), true).into(),
77                ])
78                .build()
79                .unwrap(),
80            UnboundPartitionSpec::builder()
81                .with_spec_id(1)
82                .add_partition_fields(vec![
83                    UnboundPartitionField {
84                        source_id: 1,
85                        field_id: Some(5),
86                        name: "f1".to_owned(),
87                        transform: Transform::Identity,
88                    },
89                    UnboundPartitionField {
90                        source_id: 2,
91                        field_id: Some(6),
92                        name: "f2".to_owned(),
93                        transform: Transform::Bucket(1),
94                    },
95                    UnboundPartitionField {
96                        source_id: 3,
97                        field_id: Some(7),
98                        name: "f3".to_owned(),
99                        transform: Transform::Truncate(1),
100                    },
101                    UnboundPartitionField {
102                        source_id: 4,
103                        field_id: Some(8),
104                        name: "f4".to_owned(),
105                        transform: Transform::Void,
106                    },
107                ])
108                .unwrap()
109                .build(),
110        )
111    }
112
113    fn range_table() -> Table {
114        Self::build_table(
115            Self::RANGE_TABLE,
116            Schema::builder()
117                .with_fields(vec![
118                    NestedField::new(1, "v1", Type::Primitive(PrimitiveType::Date), true).into(),
119                    NestedField::new(2, "v2", Type::Primitive(PrimitiveType::Timestamp), true)
120                        .into(),
121                    NestedField::new(3, "v3", Type::Primitive(PrimitiveType::Timestamptz), true)
122                        .into(),
123                    NestedField::new(4, "v4", Type::Primitive(PrimitiveType::Timestamptz), true)
124                        .into(),
125                ])
126                .build()
127                .unwrap(),
128            UnboundPartitionSpec::builder()
129                .with_spec_id(1)
130                .add_partition_fields(vec![
131                    UnboundPartitionField {
132                        source_id: 1,
133                        field_id: Some(5),
134                        name: "f1".to_owned(),
135                        transform: Transform::Year,
136                    },
137                    UnboundPartitionField {
138                        source_id: 2,
139                        field_id: Some(6),
140                        name: "f2".to_owned(),
141                        transform: Transform::Month,
142                    },
143                    UnboundPartitionField {
144                        source_id: 3,
145                        field_id: Some(7),
146                        name: "f3".to_owned(),
147                        transform: Transform::Day,
148                    },
149                    UnboundPartitionField {
150                        source_id: 4,
151                        field_id: Some(8),
152                        name: "f4".to_owned(),
153                        transform: Transform::Hour,
154                    },
155                ])
156                .unwrap()
157                .build(),
158        )
159    }
160}
161
162#[async_trait]
163impl CatalogV2 for MockCatalog {
164    /// List namespaces from table.
165    async fn list_namespaces(
166        &self,
167        _parent: Option<&NamespaceIdent>,
168    ) -> iceberg::Result<Vec<NamespaceIdent>> {
169        todo!()
170    }
171
172    /// Create a new namespace inside the catalog.
173    async fn create_namespace(
174        &self,
175        _namespace: &iceberg::NamespaceIdent,
176        _properties: HashMap<String, String>,
177    ) -> iceberg::Result<iceberg::Namespace> {
178        todo!()
179    }
180
181    /// Get a namespace information from the catalog.
182    async fn get_namespace(&self, _namespace: &NamespaceIdent) -> iceberg::Result<Namespace> {
183        todo!()
184    }
185
186    /// Check if namespace exists in catalog.
187    async fn namespace_exists(&self, _namespace: &NamespaceIdent) -> iceberg::Result<bool> {
188        todo!()
189    }
190
191    /// Drop a namespace from the catalog.
192    async fn drop_namespace(&self, _namespace: &NamespaceIdent) -> iceberg::Result<()> {
193        todo!()
194    }
195
196    /// List tables from namespace.
197    async fn list_tables(&self, _namespace: &NamespaceIdent) -> iceberg::Result<Vec<TableIdent>> {
198        todo!()
199    }
200
201    async fn update_namespace(
202        &self,
203        _namespace: &NamespaceIdent,
204        _properties: HashMap<String, String>,
205    ) -> iceberg::Result<()> {
206        todo!()
207    }
208
209    /// Create a new table inside the namespace.
210    async fn create_table(
211        &self,
212        _namespace: &NamespaceIdent,
213        _creation: TableCreation,
214    ) -> iceberg::Result<Table> {
215        todo!()
216    }
217
218    /// Load table from the catalog.
219    async fn load_table(&self, table: &TableIdent) -> iceberg::Result<Table> {
220        match table.name.as_ref() {
221            Self::SPARSE_TABLE => Ok(Self::sparse_table()),
222            Self::RANGE_TABLE => Ok(Self::range_table()),
223            _ => unimplemented!("table {} not found", table.name()),
224        }
225    }
226
227    /// Drop a table from the catalog.
228    async fn drop_table(&self, _table: &TableIdent) -> iceberg::Result<()> {
229        todo!()
230    }
231
232    async fn purge_table(&self, table: &TableIdent) -> iceberg::Result<()> {
233        self.drop_table(table).await
234    }
235
236    /// Check if a table exists in the catalog.
237    async fn table_exists(&self, table: &TableIdent) -> iceberg::Result<bool> {
238        match table.name.as_ref() {
239            Self::SPARSE_TABLE => Ok(true),
240            Self::RANGE_TABLE => Ok(true),
241            _ => Ok(false),
242        }
243    }
244
245    /// Rename a table in the catalog.
246    async fn rename_table(&self, _src: &TableIdent, _dest: &TableIdent) -> iceberg::Result<()> {
247        todo!()
248    }
249
250    /// Update a table to the catalog.
251    async fn update_table(&self, _commit: TableCommit) -> iceberg::Result<Table> {
252        todo!()
253    }
254
255    #[expect(
256        clippy::disallowed_types,
257        reason = "iceberg catalog trait requires returning iceberg::Error"
258    )]
259    async fn register_table(
260        &self,
261        _table_ident: &TableIdent,
262        _metadata_location: String,
263    ) -> iceberg::Result<Table> {
264        Err(iceberg::Error::new(
265            iceberg::ErrorKind::Unexpected,
266            "register_table is not supported in mock catalog",
267        ))
268    }
269}
risingwave_connector/connector_common/iceberg/mock_catalog.rs

risingwave_connector/connector_common/iceberg/
mock_catalog.rs
