risingwave_connector/sink/iceberg/
metadata.rs

1// Copyright 2026 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::anyhow;
16use futures_async_stream::try_stream;
17use iceberg::table::Table;
18use risingwave_common::array::DataChunk;
19use risingwave_common::catalog::{Field, Schema};
20use risingwave_common::types::{Fields, JsonbVal, Timestamptz};
21use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
22
23use crate::error::{ConnectorError, ConnectorResult};
24use crate::source::iceberg::{IcebergSplitEnumerator, IcebergTimeTravelInfo};
25
26/// The supported per-table Iceberg metadata relations.
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
28pub enum IcebergMetadataTableType {
29    Snapshots,
30    Manifests,
31    Files,
32}
33
34impl IcebergMetadataTableType {
35    pub fn from_suffix(suffix: &str) -> Option<Self> {
36        match suffix {
37            "snapshots" => Some(Self::Snapshots),
38            "manifests" => Some(Self::Manifests),
39            "files" => Some(Self::Files),
40            _ => None,
41        }
42    }
43
44    pub fn suffix(self) -> &'static str {
45        match self {
46            Self::Snapshots => "snapshots",
47            Self::Manifests => "manifests",
48            Self::Files => "files",
49        }
50    }
51
52    pub fn schema(self) -> Schema {
53        let fields = match self {
54            Self::Snapshots => IcebergSnapshotRow::fields(),
55            Self::Manifests => IcebergManifestRow::fields(),
56            Self::Files => IcebergFileRow::fields(),
57        }
58        .into_iter()
59        .map(|(name, data_type)| Field::with_name(data_type, name))
60        .collect();
61        Schema::new(fields)
62    }
63}
64
65#[derive(Fields)]
66struct IcebergSnapshotRow {
67    committed_at: Timestamptz,
68    snapshot_id: i64,
69    parent_id: Option<i64>,
70    sequence_number: i64,
71    operation: String,
72    manifest_list: String,
73    summary: JsonbVal,
74}
75
76#[derive(Fields)]
77struct IcebergManifestRow {
78    content: String,
79    path: String,
80    length: i64,
81    partition_spec_id: i32,
82    sequence_number: i64,
83    min_sequence_number: i64,
84    added_snapshot_id: i64,
85    added_files_count: Option<i32>,
86    existing_files_count: Option<i32>,
87    deleted_files_count: Option<i32>,
88    added_rows_count: Option<i64>,
89    existing_rows_count: Option<i64>,
90    deleted_rows_count: Option<i64>,
91    partition_summaries: Option<JsonbVal>,
92    first_row_id: Option<i64>,
93}
94
95#[derive(Fields)]
96struct IcebergFileRow {
97    content: String,
98    file_path: String,
99    file_format: String,
100    spec_id: i32,
101    record_count: i64,
102    file_size_in_bytes: i64,
103    equality_ids: Option<Vec<i32>>,
104    sort_order_id: Option<i32>,
105    snapshot_id: Option<i64>,
106    data_sequence_number: Option<i64>,
107    file_sequence_number: Option<i64>,
108    manifest_path: String,
109    referenced_data_file: Option<String>,
110    content_offset: Option<i64>,
111    content_size_in_bytes: Option<i64>,
112}
113
114fn to_i64(value: u64, field: &'static str) -> ConnectorResult<i64> {
115    value
116        .try_into()
117        .map_err(|_| anyhow!("Iceberg {field} value {value} exceeds BIGINT").into())
118}
119
120fn optional_u64_to_i64(value: Option<u64>, field: &'static str) -> ConnectorResult<Option<i64>> {
121    value.map(|value| to_i64(value, field)).transpose()
122}
123
124fn optional_u32_to_i32(value: Option<u32>, field: &'static str) -> ConnectorResult<Option<i32>> {
125    value
126        .map(|value| {
127            value
128                .try_into()
129                .map_err(|_| anyhow!("Iceberg {field} value {value} exceeds INTEGER").into())
130        })
131        .transpose()
132}
133
134fn append_row(
135    builder: &mut DataChunkBuilder,
136    row: impl Fields,
137) -> ConnectorResult<Option<DataChunk>> {
138    Ok(builder.append_one_row(row.into_owned_row()))
139}
140
141/// Read one Iceberg metadata relation and emit bounded [`DataChunk`]s.
142#[try_stream(ok = DataChunk, error = ConnectorError)]
143pub async fn scan_iceberg_metadata(
144    table: Table,
145    metadata_type: IcebergMetadataTableType,
146    time_travel_info: Option<IcebergTimeTravelInfo>,
147    chunk_size: usize,
148) {
149    let mut builder = DataChunkBuilder::new(metadata_type.schema().data_types(), chunk_size);
150
151    match metadata_type {
152        IcebergMetadataTableType::Snapshots => {
153            for snapshot in table.metadata().snapshots() {
154                let committed_at = Timestamptz::from_millis(snapshot.timestamp_ms())
155                    .ok_or_else(|| anyhow!("invalid Iceberg snapshot timestamp"))?;
156                let summary =
157                    serde_json::to_value(&snapshot.summary().additional_properties)?.into();
158                let row = IcebergSnapshotRow {
159                    committed_at,
160                    snapshot_id: snapshot.snapshot_id(),
161                    parent_id: snapshot.parent_snapshot_id(),
162                    sequence_number: snapshot.sequence_number(),
163                    operation: snapshot.summary().operation.as_str().to_owned(),
164                    manifest_list: snapshot.manifest_list().to_owned(),
165                    summary,
166                };
167                if let Some(chunk) = append_row(&mut builder, row)? {
168                    yield chunk;
169                }
170            }
171        }
172        IcebergMetadataTableType::Manifests | IcebergMetadataTableType::Files => {
173            let Some(snapshot_id) =
174                IcebergSplitEnumerator::get_snapshot_id(&table, time_travel_info)?
175            else {
176                return Ok(());
177            };
178            let snapshot = table
179                .metadata()
180                .snapshot_by_id(snapshot_id)
181                .ok_or_else(|| anyhow!("Iceberg snapshot {snapshot_id} not found"))?;
182            let metadata = table.metadata_ref();
183            let object_cache = table.object_cache();
184            let manifest_list = object_cache.get_manifest_list(snapshot, &metadata).await?;
185
186            if metadata_type == IcebergMetadataTableType::Manifests {
187                for manifest in manifest_list.entries() {
188                    let content = manifest.content.to_string();
189                    let row = IcebergManifestRow {
190                        content,
191                        path: manifest.manifest_path.clone(),
192                        length: manifest.manifest_length,
193                        partition_spec_id: manifest.partition_spec_id,
194                        sequence_number: manifest.sequence_number,
195                        min_sequence_number: manifest.min_sequence_number,
196                        added_snapshot_id: manifest.added_snapshot_id,
197                        added_files_count: optional_u32_to_i32(
198                            manifest.added_files_count,
199                            "added_files_count",
200                        )?,
201                        existing_files_count: optional_u32_to_i32(
202                            manifest.existing_files_count,
203                            "existing_files_count",
204                        )?,
205                        deleted_files_count: optional_u32_to_i32(
206                            manifest.deleted_files_count,
207                            "deleted_files_count",
208                        )?,
209                        added_rows_count: optional_u64_to_i64(
210                            manifest.added_rows_count,
211                            "added_rows_count",
212                        )?,
213                        existing_rows_count: optional_u64_to_i64(
214                            manifest.existing_rows_count,
215                            "existing_rows_count",
216                        )?,
217                        deleted_rows_count: optional_u64_to_i64(
218                            manifest.deleted_rows_count,
219                            "deleted_rows_count",
220                        )?,
221                        partition_summaries: manifest
222                            .partitions
223                            .as_ref()
224                            .map(serde_json::to_value)
225                            .transpose()?
226                            .map(Into::into),
227                        first_row_id: optional_u64_to_i64(manifest.first_row_id, "first_row_id")?,
228                    };
229                    if let Some(chunk) = append_row(&mut builder, row)? {
230                        yield chunk;
231                    }
232                }
233            } else {
234                for manifest_file in manifest_list.entries() {
235                    let manifest_path = manifest_file.manifest_path.clone();
236                    let manifest = manifest_file.load_manifest(table.file_io()).await?;
237                    for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) {
238                        let file = entry.data_file();
239                        let content = format!("{:?}", file.content_type());
240                        let row = IcebergFileRow {
241                            content,
242                            file_path: file.file_path().to_owned(),
243                            file_format: file.file_format().to_string(),
244                            spec_id: file.partition_spec_id(),
245                            record_count: to_i64(file.record_count(), "record_count")?,
246                            file_size_in_bytes: to_i64(
247                                file.file_size_in_bytes(),
248                                "file_size_in_bytes",
249                            )?,
250                            equality_ids: file.equality_ids(),
251                            sort_order_id: file.sort_order_id(),
252                            snapshot_id: entry.snapshot_id(),
253                            data_sequence_number: entry.sequence_number(),
254                            file_sequence_number: entry.file_sequence_number(),
255                            manifest_path: manifest_path.clone(),
256                            referenced_data_file: file.referenced_data_file(),
257                            content_offset: file.content_offset(),
258                            content_size_in_bytes: file.content_size_in_bytes(),
259                        };
260                        if let Some(chunk) = append_row(&mut builder, row)? {
261                            yield chunk;
262                        }
263                    }
264                }
265            }
266        }
267    }
268
269    if let Some(chunk) = builder.consume_all() {
270        yield chunk;
271    }
272}
273
274#[cfg(test)]
275mod tests {
276    use risingwave_common::types::DataType;
277
278    use super::*;
279
280    #[test]
281    fn test_metadata_table_suffix_and_schema() {
282        assert_eq!(
283            IcebergMetadataTableType::from_suffix("snapshots"),
284            Some(IcebergMetadataTableType::Snapshots)
285        );
286        assert_eq!(IcebergMetadataTableType::from_suffix("entries"), None);
287        assert_eq!(
288            IcebergMetadataTableType::Files.schema().fields[0].data_type,
289            DataType::Varchar
290        );
291        assert_eq!(
292            IcebergMetadataTableType::Files.schema().fields[0].name,
293            "content"
294        );
295    }
296}
risingwave_connector/sink/iceberg/metadata.rs

risingwave_connector/sink/iceberg/
metadata.rs
