risingwave_frontend/handler/create_source/external_schema/
iceberg.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 anyhow::Context;
16
17use super::*;
18
19/// TODO: make hidden columns additional columns, instead of normal columns?
20pub async fn extract_iceberg_columns(
21    with_properties: &WithOptionsSecResolved,
22) -> anyhow::Result<Vec<ColumnCatalog>> {
23    let props = ConnectorProperties::extract(with_properties.clone(), true)?;
24    if let ConnectorProperties::Iceberg(properties) = props {
25        let table = properties.load_table().await?;
26        let iceberg_schema: arrow_schema_iceberg::Schema =
27            ::iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema())?;
28
29        let mut columns: Vec<ColumnCatalog> = iceberg_schema
30            .fields()
31            .iter()
32            .enumerate()
33            .map(|(i, field)| -> anyhow::Result<ColumnCatalog> {
34                let column_desc = ColumnDesc::named(
35                    field.name(),
36                    ColumnId::new((i + 1).try_into().unwrap()),
37                    IcebergArrowConvert
38                        .type_from_field(field)
39                        .with_context(|| {
40                            format!(
41                                "failed to infer RisingWave type for Iceberg field {}",
42                                field.name()
43                            )
44                        })?,
45                );
46                Ok(ColumnCatalog {
47                    column_desc,
48                    // hide the _row_id column for iceberg engine table
49                    // This column is auto generated when users define a table without primary key
50                    is_hidden: field.name() == ROW_ID_COLUMN_NAME,
51                })
52            })
53            .collect::<anyhow::Result<Vec<_>>>()?;
54        columns.extend(ColumnCatalog::iceberg_hidden_cols());
55
56        tracing::info!("iceberg columns: {:?}", columns);
57
58        Ok(columns)
59    } else {
60        Err(anyhow!(format!(
61            "Invalid properties for iceberg source: {:?}",
62            props
63        )))
64    }
65}
risingwave_frontend/handler/create_source/external_schema/iceberg.rs

risingwave_frontend/handler/create_source/external_schema/
iceberg.rs
