crates/paimon/src/spec/snapshot.rs - paimon-rust - Git at Google

 // 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.

 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use typed_builder::TypedBuilder;

 /// Type of changes in this snapshot.
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
 pub enum CommitKind {
     /// Changes flushed from the mem table.
     APPEND,

     /// Changes by compacting existing data files.
     COMPACT,

     /// Changes that clear up the whole partition and then add new records.
     OVERWRITE,

     /// Collect statistics.
     ANALYZE,
 }

 impl std::fmt::Display for CommitKind {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::APPEND => write!(f, "APPEND"),
             Self::COMPACT => write!(f, "COMPACT"),
             Self::OVERWRITE => write!(f, "OVERWRITE"),
             Self::ANALYZE => write!(f, "ANALYZE"),
         }
     }
 }

 /// Snapshot for paimon.
 ///
 /// Impl Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-core/src/main/java/org/apache/paimon/Snapshot.java#L68>.
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, TypedBuilder)]
 #[serde(rename_all = "camelCase")]
 pub struct Snapshot {
     /// version of snapshot
     version: i32,
     id: i64,
     schema_id: i64,
     /// a manifest list recording all changes from the previous snapshots
     base_manifest_list: String,
     /// a manifest list recording all new changes occurred in this snapshot
     delta_manifest_list: String,
     /// a manifest list recording all changelog produced in this snapshot
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     changelog_manifest_list: Option<String>,
     /// byte size of changelog manifest list file
     #[builder(default = None)]
     #[serde(default, skip_serializing_if = "Option::is_none")]
     changelog_manifest_list_size: Option<i64>,
     /// a manifest recording all index files of this table
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     index_manifest: Option<String>,
     /// user who committed this snapshot
     commit_user: String,
     /// Mainly for snapshot deduplication.
     ///
     /// If multiple snapshots have the same commitIdentifier, reading from any of these snapshots
     /// must produce the same table.
     ///
     /// If snapshot A has a smaller commitIdentifier than snapshot B, then snapshot A must be
     /// committed before snapshot B, and thus snapshot A must contain older records than snapshot B.
     commit_identifier: i64,
     commit_kind: CommitKind,
     /// timestamp of this snapshot
     time_millis: u64,
     /// log offsets of all changes occurred in this snapshot
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     log_offsets: Option<HashMap<i32, i64>>,
     /// record count of all changes occurred in this snapshot
     #[builder(default = None)]
     total_record_count: Option<i64>,
     /// record count of all new changes occurred in this snapshot
     #[builder(default = None)]
     delta_record_count: Option<i64>,
     /// record count of all changelog produced in this snapshot
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     changelog_record_count: Option<i64>,
     /// watermark for input records
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     watermark: Option<i64>,
     /// stats file name for statistics of this table
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     statistics: Option<String>,
     /// user-facing properties of this snapshot
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     properties: Option<HashMap<String, String>>,
     /// next row id for row tracking
     #[builder(default = None)]
     #[serde(skip_serializing_if = "Option::is_none")]
     next_row_id: Option<i64>,
 }

 impl Snapshot {
     /// Get the version of this snapshot.
     #[inline]
     pub fn version(&self) -> i32 {
         self.version
     }

     /// Get the id of this snapshot.
     #[inline]
     pub fn id(&self) -> i64 {
         self.id
     }

     /// Get the schema id of this snapshot.
     #[inline]
     pub fn schema_id(&self) -> i64 {
         self.schema_id
     }

     /// Get the base manifest list of this snapshot.
     #[inline]
     pub fn base_manifest_list(&self) -> &str {
         &self.base_manifest_list
     }

     /// Get the delta manifest list of this snapshot.
     #[inline]
     pub fn delta_manifest_list(&self) -> &str {
         &self.delta_manifest_list
     }

     /// Get the changelog manifest list of this snapshot.
     #[inline]
     pub fn changelog_manifest_list(&self) -> Option<&str> {
         self.changelog_manifest_list.as_deref()
     }

     /// Get the byte size of changelog manifest list of this snapshot.
     #[inline]
     pub fn changelog_manifest_list_size(&self) -> Option<i64> {
         self.changelog_manifest_list_size
     }

     /// Get the index manifest of this snapshot.
     #[inline]
     pub fn index_manifest(&self) -> Option<&str> {
         self.index_manifest.as_deref()
     }

     /// Get the commit user of this snapshot.
     #[inline]
     pub fn commit_user(&self) -> &str {
         &self.commit_user
     }

     /// Get the commit time of this snapshot.
     #[inline]
     pub fn time_millis(&self) -> u64 {
         self.time_millis
     }

     /// Get the commit identifier of this snapshot.
     #[inline]
     pub fn commit_identifier(&self) -> i64 {
         self.commit_identifier
     }

     /// Get the log offsets of this snapshot.
     #[inline]
     pub fn log_offsets(&self) -> Option<&HashMap<i32, i64>> {
         self.log_offsets.as_ref()
     }

     /// Get the total record count of this snapshot.
     #[inline]
     pub fn total_record_count(&self) -> Option<i64> {
         self.total_record_count
     }

     /// Get the delta record count of this snapshot.
     #[inline]
     pub fn delta_record_count(&self) -> Option<i64> {
         self.delta_record_count
     }

     /// Get the changelog record count of this snapshot.
     #[inline]
     pub fn changelog_record_count(&self) -> Option<i64> {
         self.changelog_record_count
     }

     /// Get the watermark of this snapshot.
     #[inline]
     pub fn watermark(&self) -> Option<i64> {
         self.watermark
     }

     /// Get the statistics of this snapshot.
     #[inline]
     pub fn statistics(&self) -> Option<&str> {
         self.statistics.as_deref()
     }

     /// Get the properties of this snapshot.
     #[inline]
     pub fn properties(&self) -> Option<&HashMap<String, String>> {
         self.properties.as_ref()
     }

     /// Get the next row id of this snapshot.
     #[inline]
     pub fn next_row_id(&self) -> Option<i64> {
         self.next_row_id
     }

     /// Get the commit kind of this snapshot.
     #[inline]
     pub fn commit_kind(&self) -> &CommitKind {
         &self.commit_kind
     }
 }

 #[cfg(test)]
 mod tests {
     use super::*;
     use pretty_assertions::assert_eq;
     use serde_json;
     use std::env::current_dir;

     fn load_fixture(name: &str) -> String {
         let path = current_dir()
             .unwrap_or_else(|err| panic!("current_dir must exist: {err}"))
             .join(format!("tests/fixtures/snapshot/{name}.json"));
         let bytes = std::fs::read(&path)
             .unwrap_or_else(|err| panic!("fixtures {path:?} load failed: {err}"));
         String::from_utf8(bytes).expect("fixtures content must be valid utf8")
     }

     fn test_cases() -> Vec<(&'static str, Snapshot)> {
         vec![
             (
                 "snapshot-v3",
                 Snapshot::builder()
                     .version(3)
                     .id(2)
                     .schema_id(0)
                     .base_manifest_list(
                         "manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-0".to_string(),
                     )
                     .delta_manifest_list(
                         "manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-1".to_string(),
                     )
                     .commit_user("abbaac9e-4a17-43e3-b135-2269da263e3a".to_string())
                     .commit_identifier(9223372036854775807)
                     .changelog_manifest_list(Some(
                         "manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-2".to_string(),
                     ))
                     .commit_kind(CommitKind::APPEND)
                     .time_millis(1724509030368)
                     .log_offsets(Some(HashMap::default()))
                     .total_record_count(Some(4))
                     .delta_record_count(Some(2))
                     .changelog_record_count(Some(2))
                     .statistics(Some("statistics_string".to_string()))
                     .properties(Some(HashMap::from([(
                         "scan.snapshot-id".to_string(),
                         "2".to_string(),
                     )])))
                     .build(),
             ),
             (
                 "snapshot-v3-none-field",
                 Snapshot::builder()
                     .version(3)
                     .id(2)
                     .schema_id(0)
                     .base_manifest_list(
                         "manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-0".to_string(),
                     )
                     .delta_manifest_list(
                         "manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-1".to_string(),
                     )
                     .commit_user("abbaac9e-4a17-43e3-b135-2269da263e3a".to_string())
                     .commit_identifier(9223372036854775807)
                     .changelog_manifest_list(None)
                     .commit_kind(CommitKind::APPEND)
                     .time_millis(1724509030368)
                     .log_offsets(Some(HashMap::default()))
                     .total_record_count(Some(4))
                     .delta_record_count(Some(2))
                     .changelog_record_count(Some(2))
                     .build(),
             ),
         ]
     }

     #[test]
     fn test_snapshot_serialization_deserialization() {
         for (name, expect) in test_cases() {
             let content = load_fixture(name);
             let snapshot: Snapshot =
                 serde_json::from_str(content.as_str()).expect("Failed to deserialize Snapshot");
             assert_eq!(snapshot, expect);
             let serialized =
                 serde_json::to_string(&snapshot).expect("Failed to serialize Snapshot");

             let deserialized: Snapshot = serde_json::from_str(&serialized)
                 .expect("Failed to deserialize serialized Snapshot");

             assert_eq!(snapshot, deserialized);
         }
     }
 }
	// 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.

	use serde::{Deserialize, Serialize};
	use std::collections::HashMap;
	use typed_builder::TypedBuilder;

	/// Type of changes in this snapshot.
	#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
	pub enum CommitKind {
	/// Changes flushed from the mem table.
	APPEND,

	/// Changes by compacting existing data files.
	COMPACT,

	/// Changes that clear up the whole partition and then add new records.
	OVERWRITE,

	/// Collect statistics.
	ANALYZE,
	}

	impl std::fmt::Display for CommitKind {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
	match self {
	Self::APPEND => write!(f, "APPEND"),
	Self::COMPACT => write!(f, "COMPACT"),
	Self::OVERWRITE => write!(f, "OVERWRITE"),
	Self::ANALYZE => write!(f, "ANALYZE"),
	}
	}
	}

	/// Snapshot for paimon.
	///
	/// Impl Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-core/src/main/java/org/apache/paimon/Snapshot.java#L68>.
	#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, TypedBuilder)]
	#[serde(rename_all = "camelCase")]
	pub struct Snapshot {
	/// version of snapshot
	version: i32,
	id: i64,
	schema_id: i64,
	/// a manifest list recording all changes from the previous snapshots
	base_manifest_list: String,
	/// a manifest list recording all new changes occurred in this snapshot
	delta_manifest_list: String,
	/// a manifest list recording all changelog produced in this snapshot
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	changelog_manifest_list: Option<String>,
	/// byte size of changelog manifest list file
	#[builder(default = None)]
	#[serde(default, skip_serializing_if = "Option::is_none")]
	changelog_manifest_list_size: Option<i64>,
	/// a manifest recording all index files of this table
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	index_manifest: Option<String>,
	/// user who committed this snapshot
	commit_user: String,
	/// Mainly for snapshot deduplication.
	///
	/// If multiple snapshots have the same commitIdentifier, reading from any of these snapshots
	/// must produce the same table.
	///
	/// If snapshot A has a smaller commitIdentifier than snapshot B, then snapshot A must be
	/// committed before snapshot B, and thus snapshot A must contain older records than snapshot B.
	commit_identifier: i64,
	commit_kind: CommitKind,
	/// timestamp of this snapshot
	time_millis: u64,
	/// log offsets of all changes occurred in this snapshot
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	log_offsets: Option<HashMap<i32, i64>>,
	/// record count of all changes occurred in this snapshot
	#[builder(default = None)]
	total_record_count: Option<i64>,
	/// record count of all new changes occurred in this snapshot
	#[builder(default = None)]
	delta_record_count: Option<i64>,
	/// record count of all changelog produced in this snapshot
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	changelog_record_count: Option<i64>,
	/// watermark for input records
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	watermark: Option<i64>,
	/// stats file name for statistics of this table
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	statistics: Option<String>,
	/// user-facing properties of this snapshot
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	properties: Option<HashMap<String, String>>,
	/// next row id for row tracking
	#[builder(default = None)]
	#[serde(skip_serializing_if = "Option::is_none")]
	next_row_id: Option<i64>,
	}

	impl Snapshot {
	/// Get the version of this snapshot.
	#[inline]
	pub fn version(&self) -> i32 {
	self.version
	}

	/// Get the id of this snapshot.
	#[inline]
	pub fn id(&self) -> i64 {
	self.id
	}

	/// Get the schema id of this snapshot.
	#[inline]
	pub fn schema_id(&self) -> i64 {
	self.schema_id
	}

	/// Get the base manifest list of this snapshot.
	#[inline]
	pub fn base_manifest_list(&self) -> &str {
	&self.base_manifest_list
	}

	/// Get the delta manifest list of this snapshot.
	#[inline]
	pub fn delta_manifest_list(&self) -> &str {
	&self.delta_manifest_list
	}

	/// Get the changelog manifest list of this snapshot.
	#[inline]
	pub fn changelog_manifest_list(&self) -> Option<&str> {
	self.changelog_manifest_list.as_deref()
	}

	/// Get the byte size of changelog manifest list of this snapshot.
	#[inline]
	pub fn changelog_manifest_list_size(&self) -> Option<i64> {
	self.changelog_manifest_list_size
	}

	/// Get the index manifest of this snapshot.
	#[inline]
	pub fn index_manifest(&self) -> Option<&str> {
	self.index_manifest.as_deref()
	}

	/// Get the commit user of this snapshot.
	#[inline]
	pub fn commit_user(&self) -> &str {
	&self.commit_user
	}

	/// Get the commit time of this snapshot.
	#[inline]
	pub fn time_millis(&self) -> u64 {
	self.time_millis
	}

	/// Get the commit identifier of this snapshot.
	#[inline]
	pub fn commit_identifier(&self) -> i64 {
	self.commit_identifier
	}

	/// Get the log offsets of this snapshot.
	#[inline]
	pub fn log_offsets(&self) -> Option<&HashMap<i32, i64>> {
	self.log_offsets.as_ref()
	}

	/// Get the total record count of this snapshot.
	#[inline]
	pub fn total_record_count(&self) -> Option<i64> {
	self.total_record_count
	}

	/// Get the delta record count of this snapshot.
	#[inline]
	pub fn delta_record_count(&self) -> Option<i64> {
	self.delta_record_count
	}

	/// Get the changelog record count of this snapshot.
	#[inline]
	pub fn changelog_record_count(&self) -> Option<i64> {
	self.changelog_record_count
	}

	/// Get the watermark of this snapshot.
	#[inline]
	pub fn watermark(&self) -> Option<i64> {
	self.watermark
	}

	/// Get the statistics of this snapshot.
	#[inline]
	pub fn statistics(&self) -> Option<&str> {
	self.statistics.as_deref()
	}

	/// Get the properties of this snapshot.
	#[inline]
	pub fn properties(&self) -> Option<&HashMap<String, String>> {
	self.properties.as_ref()
	}

	/// Get the next row id of this snapshot.
	#[inline]
	pub fn next_row_id(&self) -> Option<i64> {
	self.next_row_id
	}

	/// Get the commit kind of this snapshot.
	#[inline]
	pub fn commit_kind(&self) -> &CommitKind {
	&self.commit_kind
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use pretty_assertions::assert_eq;
	use serde_json;
	use std::env::current_dir;

	fn load_fixture(name: &str) -> String {
	let path = current_dir()
	.unwrap_or_else(\|err\| panic!("current_dir must exist: {err}"))
	.join(format!("tests/fixtures/snapshot/{name}.json"));
	let bytes = std::fs::read(&path)
	.unwrap_or_else(\|err\| panic!("fixtures {path:?} load failed: {err}"));
	String::from_utf8(bytes).expect("fixtures content must be valid utf8")
	}

	fn test_cases() -> Vec<(&'static str, Snapshot)> {
	vec![
	(
	"snapshot-v3",
	Snapshot::builder()
	.version(3)
	.id(2)
	.schema_id(0)
	.base_manifest_list(
	"manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-0".to_string(),
	)
	.delta_manifest_list(
	"manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-1".to_string(),
	)
	.commit_user("abbaac9e-4a17-43e3-b135-2269da263e3a".to_string())
	.commit_identifier(9223372036854775807)
	.changelog_manifest_list(Some(
	"manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-2".to_string(),
	))
	.commit_kind(CommitKind::APPEND)
	.time_millis(1724509030368)
	.log_offsets(Some(HashMap::default()))
	.total_record_count(Some(4))
	.delta_record_count(Some(2))
	.changelog_record_count(Some(2))
	.statistics(Some("statistics_string".to_string()))
	.properties(Some(HashMap::from([(
	"scan.snapshot-id".to_string(),
	"2".to_string(),
	)])))
	.build(),
	),
	(
	"snapshot-v3-none-field",
	Snapshot::builder()
	.version(3)
	.id(2)
	.schema_id(0)
	.base_manifest_list(
	"manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-0".to_string(),
	)
	.delta_manifest_list(
	"manifest-list-ea4b892d-edc8-4ee7-9eee-7068b83a947b-1".to_string(),
	)
	.commit_user("abbaac9e-4a17-43e3-b135-2269da263e3a".to_string())
	.commit_identifier(9223372036854775807)
	.changelog_manifest_list(None)
	.commit_kind(CommitKind::APPEND)
	.time_millis(1724509030368)
	.log_offsets(Some(HashMap::default()))
	.total_record_count(Some(4))
	.delta_record_count(Some(2))
	.changelog_record_count(Some(2))
	.build(),
	),
	]
	}

	#[test]
	fn test_snapshot_serialization_deserialization() {
	for (name, expect) in test_cases() {
	let content = load_fixture(name);
	let snapshot: Snapshot =
	serde_json::from_str(content.as_str()).expect("Failed to deserialize Snapshot");
	assert_eq!(snapshot, expect);
	let serialized =
	serde_json::to_string(&snapshot).expect("Failed to serialize Snapshot");

	let deserialized: Snapshot = serde_json::from_str(&serialized)
	.expect("Failed to deserialize serialized Snapshot");

	assert_eq!(snapshot, deserialized);
	}
	}
	}