laskoviymishka commented on code in PR #2825: URL: https://github.com/apache/iceberg-rust/pull/2825#discussion_r3577130429
########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); + while let Some(ancestor) = snapshot { + // corrupt metadata with a parent cycle must not hang the scan Review Comment: This guard is the one thing here that isn't exercised by the tests, and it's exactly the kind of thing a refactor removes silently. I'd add a metadata fixture with a parent cycle and assert `scan()` terminates, so the guard is actually locked in. ########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); + while let Some(ancestor) = snapshot { + // corrupt metadata with a parent cycle must not hang the scan + if !ancestors.insert(ancestor.snapshot_id()) { + break; + } + snapshot = ancestor + .parent_snapshot_id() + .and_then(|id| metadata.snapshot_by_id(id)); + } + + let mut made_current_at = + PrimitiveBuilder::<TimestampMicrosecondType>::new().with_timezone("+00:00"); + let mut snapshot_id = PrimitiveBuilder::<Int64Type>::new(); + let mut parent_id = PrimitiveBuilder::<Int64Type>::new(); + let mut is_current_ancestor = BooleanBuilder::new(); + + for entry in metadata.history() { + made_current_at.append_value(entry.timestamp_ms() * 1000); // ms -> µs + snapshot_id.append_value(entry.snapshot_id); Review Comment: Small consistency thing: `entry.timestamp_ms()` uses the method one line up and `entry.snapshot_id` uses the field here. `SnapshotLog` exposes both, so it works, but reading them back to back it's a little jarring. I'd use field access for both — `entry.timestamp_ms`. ########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); + while let Some(ancestor) = snapshot { Review Comment: This ancestor walk is basically `crate::util::snapshot::ancestors_of` (util/snapshot.rs:38) inlined — except that one has no cycle guard and this one does. I'd rather not maintain two copies of the traversal that disagree on a safety property. Could we push the cycle guard down into `ancestors_of` and reuse it here? If there's a real API mismatch (`&TableMetadata` vs `&TableMetadataRef`) that makes reuse awkward, a one-line comment noting why it's inlined would do. wdyt? ########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); + while let Some(ancestor) = snapshot { + // corrupt metadata with a parent cycle must not hang the scan + if !ancestors.insert(ancestor.snapshot_id()) { + break; + } + snapshot = ancestor + .parent_snapshot_id() + .and_then(|id| metadata.snapshot_by_id(id)); + } + + let mut made_current_at = + PrimitiveBuilder::<TimestampMicrosecondType>::new().with_timezone("+00:00"); + let mut snapshot_id = PrimitiveBuilder::<Int64Type>::new(); + let mut parent_id = PrimitiveBuilder::<Int64Type>::new(); + let mut is_current_ancestor = BooleanBuilder::new(); + + for entry in metadata.history() { + made_current_at.append_value(entry.timestamp_ms() * 1000); // ms -> µs Review Comment: `timestamp_ms() * 1000` is an unchecked i64 multiply — it'll wrap for pre-epoch or huge values. `snapshots.rs:112` has the same thing, so this isn't a regression, but a saturating/checked multiply would be more defensive if you're touching it anyway. ########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); + while let Some(ancestor) = snapshot { + // corrupt metadata with a parent cycle must not hang the scan + if !ancestors.insert(ancestor.snapshot_id()) { + break; + } + snapshot = ancestor + .parent_snapshot_id() + .and_then(|id| metadata.snapshot_by_id(id)); + } + + let mut made_current_at = + PrimitiveBuilder::<TimestampMicrosecondType>::new().with_timezone("+00:00"); Review Comment: There's a `crate::arrow::UTC_TIME_ZONE` constant (arrow/schema.rs:45) for exactly this. `snapshots.rs:86` also hard-codes the string, so this'd be the second occurrence — I'd use the constant here rather than grow the string literals. ########## crates/iceberg/src/inspect/history.rs: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! History table: the chronological log of every snapshot that was ever the +//! table's current snapshot. +//! +//! Each row is one entry of the metadata snapshot-log and answers two questions +//! about the table's lineage: *when* a snapshot became current, and *whether* +//! it is still an ancestor of the current state. Snapshots that were rolled back +//! remain visible in the log yet are flagged `is_current_ancestor = false`, so +//! the live lineage stays distinguishable from abandoned history. +//! +//! References: +//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54> + +use std::collections::HashSet; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder}; +use arrow_array::types::{Int64Type, TimestampMicrosecondType}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, Type}; +use crate::table::Table; + +/// History table. +pub struct HistoryTable<'a> { + table: &'a Table, +} + +impl<'a> HistoryTable<'a> { + /// Create a new History table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the history table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required( + 1, + "made_current_at", + Type::Primitive(PrimitiveType::Timestamptz), + ), + NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)), + NestedField::required( + 4, + "is_current_ancestor", + Type::Primitive(PrimitiveType::Boolean), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the history table. + pub async fn scan(&self) -> Result<ArrowRecordBatchStream> { + let schema = schema_to_arrow_schema(&self.schema())?; + let metadata = self.table.metadata(); + + // Walk the current snapshot's parent chain once: membership in this set + // is what tells the live lineage apart from rolled-back history entries. + let mut ancestors = HashSet::new(); + let mut snapshot = metadata.current_snapshot(); Review Comment: Heads up that `current_snapshot()` has a hidden `.expect()` (table_metadata.rs:315) that panics if `current_snapshot_id` is set but the snapshot is missing from the map — corrupt or GC'd metadata. `validate_refs` probably makes that unreachable for a well-formed `Table`, so I don't think it needs code here, but a short comment noting we rely on that invariant would save the next reader the trip. (Java's `SnapshotUtil` returns null instead of panicking, fwiw.) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
