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        columns_from_table(&table, properties.streaming_updates)
27    } else {
28        anyhow::bail!("invalid connector for Iceberg schema inference")
29    }
30}
31
32fn columns_from_table(
33    table: &::iceberg::table::Table,
34    streaming_updates: bool,
35) -> anyhow::Result<Vec<ColumnCatalog>> {
36    let iceberg_schema: arrow_schema_iceberg::Schema =
37        ::iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema())?;
38
39    let mut columns: Vec<ColumnCatalog> = iceberg_schema
40        .fields()
41        .iter()
42        .enumerate()
43        .map(|(i, field)| -> anyhow::Result<ColumnCatalog> {
44            let column_desc = ColumnDesc::named(
45                field.name(),
46                ColumnId::new((i + 1).try_into().unwrap()),
47                IcebergArrowConvert
48                    .type_from_field(field)
49                    .with_context(|| {
50                        format!(
51                            "failed to infer RisingWave type for Iceberg field {}",
52                            field.name()
53                        )
54                    })?,
55            );
56            Ok(ColumnCatalog {
57                column_desc,
58                // hide the _row_id column for iceberg engine table
59                // This column is auto generated when users define a table without primary key
60                is_hidden: field.name() == ROW_ID_COLUMN_NAME,
61            })
62        })
63        .collect::<anyhow::Result<Vec<_>>>()?;
64    if streaming_updates {
65        anyhow::ensure!(
66            !columns.iter().any(ColumnCatalog::is_iceberg_hidden_column),
67            "Iceberg update sources cannot expose reserved physical metadata columns"
68        );
69        risingwave_connector::source::iceberg::update_planner::IcebergUpdateBinding::bind_columns(
70            table, &columns,
71        )?;
72    } else {
73        columns.extend(ColumnCatalog::iceberg_hidden_cols());
74    }
75
76    tracing::info!("iceberg columns: {:?}", columns);
77
78    Ok(columns)
79}
risingwave_frontend/handler/create_source/external_schema/iceberg.rs

risingwave_frontend/handler/create_source/external_schema/
iceberg.rs
