liurenjie1024 commented on code in PR #79:
URL: https://github.com/apache/iceberg-rust/pull/79#discussion_r1413462308
##########
crates/iceberg/src/spec/manifest_list.rs:
##########
@@ -30,6 +30,9 @@ use self::{
use super::{FormatVersion, StructType};
+/// The seq number when no added files are present.
+pub const UNASSIGNED_SEQ_NUMBER: i64 = -1;
Review Comment:
Is this snapshot seq number? How about moving it to `snapshot` module? Also
how about making `pub(crate)`
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -0,0 +1,1847 @@
+// 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.
+
+//! Manifest for Iceberg.
+use self::_const_schema::{manifest_schema_v1, manifest_schema_v2};
+
+use super::Literal;
+use super::{
+ FieldSummary, FormatVersion, ManifestContentType, ManifestListEntry,
PartitionSpec, Schema,
+ Struct, UNASSIGNED_SEQ_NUMBER,
+};
+use crate::io::OutputFile;
+use crate::spec::PartitionField;
+use crate::{Error, ErrorKind};
+use apache_avro::{from_value, to_value, Reader as AvroReader, Writer as
AvroWriter};
+use futures::AsyncWriteExt;
+use serde_json::to_vec;
+use std::cmp::min;
+use std::collections::HashMap;
+use std::str::FromStr;
+
+/// A manifest contains metadata and a list of entries.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct Manifest {
+ metadata: ManifestMetadata,
+ entries: Vec<ManifestEntry>,
+}
+
+impl Manifest {
+ /// Parse manifest from bytes of avro file.
+ pub fn parse_avro(bs: &[u8]) -> Result<Self, Error> {
+ let reader = AvroReader::new(bs)?;
+
+ // Parse manifest metadata
+ let meta = reader.user_metadata();
+ let metadata = ManifestMetadata::parse(meta)?;
+
+ // Parse manifest entries
+ let partition_type =
metadata.partition_spec.partition_type(&metadata.schema)?;
+
+ let entries = match metadata.format_version {
+ FormatVersion::V1 => {
+ let schema = manifest_schema_v1(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV1>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ FormatVersion::V2 => {
+ let schema = manifest_schema_v2(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV2>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ };
+
+ Ok(Manifest { metadata, entries })
+ }
+}
+
+/// A manifest writer.
+pub struct ManifestWriter {
+ output: OutputFile,
+
+ snapshot_id: i64,
+
+ added_files: i32,
+ added_rows: i64,
+ existing_files: i32,
+ existing_rows: i64,
+ deleted_files: i32,
+ deleted_rows: i64,
+
+ seq_num: i64,
+ min_seq_num: Option<i64>,
+
+ key_metadata: Option<Vec<u8>>,
+
+ field_summary: HashMap<i32, FieldSummary>,
+}
+
+impl ManifestWriter {
+ /// Create a new manifest writer.
+ pub fn new(
+ output: OutputFile,
+ snapshot_id: i64,
+ seq_num: i64,
+ key_metadata: Option<Vec<u8>>,
+ ) -> Self {
+ Self {
+ output,
+ snapshot_id,
+ added_files: 0,
+ added_rows: 0,
+ existing_files: 0,
+ existing_rows: 0,
+ deleted_files: 0,
+ deleted_rows: 0,
+ seq_num,
+ min_seq_num: None,
+ key_metadata,
+ field_summary: HashMap::new(),
+ }
+ }
+
+ fn update_field_summary(&mut self, entry: &ManifestEntry) {
+ // Update field summary
+ if let Some(null) = &entry.data_file.null_value_counts {
+ for (&k, &v) in null {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if v > 0 {
+ field_summary.contains_null = true;
+ }
+ }
+ }
+ if let Some(nan) = &entry.data_file.nan_value_counts {
+ for (&k, &v) in nan {
+ let field_summary = self.field_summary.entry(k).or_default();
+ assert!(v >= 0);
+ if v > 0 {
+ field_summary.contains_nan = Some(true);
+ }
+ if v == 0 {
+ field_summary.contains_nan = Some(false);
+ }
+ }
+ }
+ if let Some(lower_bound) = &entry.data_file.lower_bounds {
+ for (&k, v) in lower_bound {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if let Some(cur) = &field_summary.lower_bound {
+ if v < cur {
+ field_summary.lower_bound = Some(v.clone());
+ }
+ } else {
+ field_summary.lower_bound = Some(v.clone());
+ }
+ }
+ }
+ if let Some(upper_bound) = &entry.data_file.upper_bounds {
+ for (&k, v) in upper_bound {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if let Some(cur) = &field_summary.upper_bound {
+ if v > cur {
+ field_summary.upper_bound = Some(v.clone());
+ }
+ } else {
+ field_summary.upper_bound = Some(v.clone());
+ }
+ }
+ }
+ }
+
+ fn get_field_summary_vec(&mut self, spec_fields: &[PartitionField]) ->
Vec<FieldSummary> {
+ let mut partition_summary =
Vec::with_capacity(self.field_summary.len());
+ for field in spec_fields {
+ let entry = self
+ .field_summary
+ .remove(&field.source_id)
+ .unwrap_or(FieldSummary::default());
+ partition_summary.push(entry);
+ }
+ partition_summary
+ }
+
+ /// Write a manifest entry.
+ pub async fn write(mut self, manifest: Manifest) ->
Result<ManifestListEntry, Error> {
+ // Create the avro writer
+ let partition_type = manifest
+ .metadata
+ .partition_spec
+ .partition_type(&manifest.metadata.schema)?;
+ let table_schema = &manifest.metadata.schema;
+ let avro_schema = match manifest.metadata.format_version {
+ FormatVersion::V1 => manifest_schema_v1(partition_type.clone())?,
+ FormatVersion::V2 => manifest_schema_v2(partition_type.clone())?,
+ };
+ let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new());
+ avro_writer.add_user_metadata(
+ "schema".to_string(),
+ to_vec(table_schema).map_err(|err| {
+ Error::new(ErrorKind::DataInvalid, "Fail to serialize table
schema")
+ .with_source(err)
+ })?,
+ )?;
+ avro_writer.add_user_metadata(
+ "schema-id".to_string(),
+ table_schema.schema_id().to_string(),
+ )?;
+ avro_writer.add_user_metadata(
+ "partition-spec".to_string(),
+ to_vec(&manifest.metadata.partition_spec.fields).map_err(|err| {
+ Error::new(ErrorKind::DataInvalid, "Fail to serialize
partition spec")
+ .with_source(err)
+ })?,
+ )?;
+ avro_writer.add_user_metadata(
+ "partition-spec-id".to_string(),
+ manifest.metadata.partition_spec.spec_id.to_string(),
+ )?;
+ avro_writer.add_user_metadata(
+ "format-version".to_string(),
+ manifest.metadata.format_version.to_string(),
+ )?;
+ if manifest.metadata.format_version == FormatVersion::V2 {
+ avro_writer
+ .add_user_metadata("content".to_string(),
manifest.metadata.content.to_string())?;
+ }
+
+ // Write manifest entries
+ for entry in manifest.entries {
+ match entry.status {
+ ManifestStatus::Added => {
+ self.added_files += 1;
+ self.added_rows += entry.data_file.record_count;
+ }
+ ManifestStatus::Deleted => {
+ self.deleted_files += 1;
+ self.deleted_rows += entry.data_file.record_count;
+ }
+ ManifestStatus::Existing => {
+ self.existing_files += 1;
+ self.existing_rows += entry.data_file.record_count;
+ }
+ }
+
+ if entry.is_alive() {
+ if let Some(cur_min_seq_num) = self.min_seq_num {
+ self.min_seq_num = Some(
+ entry
+ .sequence_number
+ .map(|v| min(v, cur_min_seq_num))
+ .unwrap_or(cur_min_seq_num),
+ );
+ } else {
+ self.min_seq_num = entry.sequence_number;
+ }
+ }
+
+ self.update_field_summary(&entry);
+
+ let value = match manifest.metadata.format_version {
+ FormatVersion::V1 => {
+ to_value(_serde::ManifestEntryV1::try_from(entry,
&partition_type)?)?
+ .resolve(&avro_schema)?
+ }
+ FormatVersion::V2 => {
+ to_value(_serde::ManifestEntryV2::try_from(entry,
&partition_type)?)?
+ .resolve(&avro_schema)?
+ }
+ };
+
+ avro_writer.append(value)?;
+ }
+
+ let length = avro_writer.flush()?;
+ let content = avro_writer.into_inner()?;
+ let mut writer = self.output.writer().await?;
+ writer.write_all(&content).await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "Fail to write Manifest
Entry").with_source(err)
+ })?;
+ writer.close().await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "Fail to write Manifest
Entry").with_source(err)
+ })?;
+
+ let partition_summary =
+
self.get_field_summary_vec(&manifest.metadata.partition_spec.fields);
+
+ Ok(ManifestListEntry {
+ manifest_path: self.output.location().to_string(),
+ manifest_length: length as i64,
+ partition_spec_id: manifest.metadata.partition_spec.spec_id,
+ content: manifest.metadata.content,
+ sequence_number: self.seq_num,
+ min_sequence_number:
self.min_seq_num.unwrap_or(UNASSIGNED_SEQ_NUMBER),
+ added_snapshot_id: self.snapshot_id,
+ added_data_files_count: Some(self.added_files),
+ existing_data_files_count: Some(self.existing_files),
+ deleted_data_files_count: Some(self.deleted_files),
+ added_rows_count: Some(self.added_rows),
+ existing_rows_count: Some(self.existing_rows),
+ deleted_rows_count: Some(self.deleted_rows),
+ partitions: partition_summary,
+ key_metadata: self.key_metadata.unwrap_or_default(),
+ })
+ }
+}
+
+/// This is a helper module that defines the schema field of the manifest list
entry.
+mod _const_schema {
+ use std::sync::Arc;
+
+ use apache_avro::Schema as AvroSchema;
+ use once_cell::sync::Lazy;
+
+ use crate::{
+ avro::schema_to_avro_schema,
+ spec::{
+ ListType, MapType, NestedField, NestedFieldRef, PrimitiveType,
Schema, StructType, Type,
+ },
+ Error,
+ };
+
+ pub static STATUS: Lazy<NestedFieldRef> = {
Review Comment:
Do we need make these fields `pub`? I think make them private would be
enough?
##########
crates/iceberg/src/spec/transform.rs:
##########
@@ -126,17 +126,20 @@ pub enum Transform {
impl Transform {
/// Get the return type of transform given the input type.
/// Returns `None` if it can't be transformed.
- pub fn result_type(&self, input_type: &Type) -> Option<Type> {
+ pub fn result_type(&self, input_type: &Type) -> Result<Type> {
Review Comment:
Why we need to change this signature? I think an `Option` is more
appropriate here. I think we can have a method `try_infer_return_type` to
convert `Option` to `Result`?
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -0,0 +1,1847 @@
+// 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.
+
+//! Manifest for Iceberg.
+use self::_const_schema::{manifest_schema_v1, manifest_schema_v2};
+
+use super::Literal;
+use super::{
+ FieldSummary, FormatVersion, ManifestContentType, ManifestListEntry,
PartitionSpec, Schema,
+ Struct, UNASSIGNED_SEQ_NUMBER,
+};
+use crate::io::OutputFile;
+use crate::spec::PartitionField;
+use crate::{Error, ErrorKind};
+use apache_avro::{from_value, to_value, Reader as AvroReader, Writer as
AvroWriter};
+use futures::AsyncWriteExt;
+use serde_json::to_vec;
+use std::cmp::min;
+use std::collections::HashMap;
+use std::str::FromStr;
+
+/// A manifest contains metadata and a list of entries.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct Manifest {
+ metadata: ManifestMetadata,
+ entries: Vec<ManifestEntry>,
+}
+
+impl Manifest {
+ /// Parse manifest from bytes of avro file.
+ pub fn parse_avro(bs: &[u8]) -> Result<Self, Error> {
+ let reader = AvroReader::new(bs)?;
+
+ // Parse manifest metadata
+ let meta = reader.user_metadata();
+ let metadata = ManifestMetadata::parse(meta)?;
+
+ // Parse manifest entries
+ let partition_type =
metadata.partition_spec.partition_type(&metadata.schema)?;
+
+ let entries = match metadata.format_version {
+ FormatVersion::V1 => {
+ let schema = manifest_schema_v1(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV1>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ FormatVersion::V2 => {
+ let schema = manifest_schema_v2(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV2>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ };
+
+ Ok(Manifest { metadata, entries })
+ }
+}
+
+/// A manifest writer.
+pub struct ManifestWriter {
+ output: OutputFile,
+
+ snapshot_id: i64,
+
+ added_files: i32,
+ added_rows: i64,
+ existing_files: i32,
+ existing_rows: i64,
+ deleted_files: i32,
+ deleted_rows: i64,
+
+ seq_num: i64,
+ min_seq_num: Option<i64>,
+
+ key_metadata: Option<Vec<u8>>,
+
+ field_summary: HashMap<i32, FieldSummary>,
+}
+
+impl ManifestWriter {
+ /// Create a new manifest writer.
+ pub fn new(
+ output: OutputFile,
+ snapshot_id: i64,
+ seq_num: i64,
+ key_metadata: Option<Vec<u8>>,
+ ) -> Self {
+ Self {
+ output,
+ snapshot_id,
+ added_files: 0,
+ added_rows: 0,
+ existing_files: 0,
+ existing_rows: 0,
+ deleted_files: 0,
+ deleted_rows: 0,
+ seq_num,
+ min_seq_num: None,
+ key_metadata,
+ field_summary: HashMap::new(),
+ }
+ }
+
+ fn update_field_summary(&mut self, entry: &ManifestEntry) {
+ // Update field summary
+ if let Some(null) = &entry.data_file.null_value_counts {
+ for (&k, &v) in null {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if v > 0 {
+ field_summary.contains_null = true;
+ }
+ }
+ }
+ if let Some(nan) = &entry.data_file.nan_value_counts {
+ for (&k, &v) in nan {
+ let field_summary = self.field_summary.entry(k).or_default();
+ assert!(v >= 0);
Review Comment:
This assertion makes me thinking that maybe we should use unsigned values
for in memory data structs when it's counting? Java uses i64 by default since
it has no unsigned values, but in rust we should use this to ensure safety?
What do you think? cc @Fokko @JanKaul @Xuanwo
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -0,0 +1,1847 @@
+// 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.
+
+//! Manifest for Iceberg.
+use self::_const_schema::{manifest_schema_v1, manifest_schema_v2};
+
+use super::Literal;
+use super::{
+ FieldSummary, FormatVersion, ManifestContentType, ManifestListEntry,
PartitionSpec, Schema,
+ Struct, UNASSIGNED_SEQ_NUMBER,
+};
+use crate::io::OutputFile;
+use crate::spec::PartitionField;
+use crate::{Error, ErrorKind};
+use apache_avro::{from_value, to_value, Reader as AvroReader, Writer as
AvroWriter};
+use futures::AsyncWriteExt;
+use serde_json::to_vec;
+use std::cmp::min;
+use std::collections::HashMap;
+use std::str::FromStr;
+
+/// A manifest contains metadata and a list of entries.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct Manifest {
+ metadata: ManifestMetadata,
+ entries: Vec<ManifestEntry>,
+}
+
+impl Manifest {
+ /// Parse manifest from bytes of avro file.
+ pub fn parse_avro(bs: &[u8]) -> Result<Self, Error> {
+ let reader = AvroReader::new(bs)?;
+
+ // Parse manifest metadata
+ let meta = reader.user_metadata();
+ let metadata = ManifestMetadata::parse(meta)?;
+
+ // Parse manifest entries
+ let partition_type =
metadata.partition_spec.partition_type(&metadata.schema)?;
+
+ let entries = match metadata.format_version {
+ FormatVersion::V1 => {
+ let schema = manifest_schema_v1(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV1>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ FormatVersion::V2 => {
+ let schema = manifest_schema_v2(partition_type.clone())?;
+ let reader = AvroReader::with_schema(&schema, bs)?;
+ reader
+ .into_iter()
+ .map(|value| {
+ from_value::<_serde::ManifestEntryV2>(&value?)?
+ .try_into(&partition_type, &metadata.schema)
+ })
+ .collect::<Result<Vec<_>, Error>>()?
+ }
+ };
+
+ Ok(Manifest { metadata, entries })
+ }
+}
+
+/// A manifest writer.
+pub struct ManifestWriter {
+ output: OutputFile,
+
+ snapshot_id: i64,
+
+ added_files: i32,
+ added_rows: i64,
+ existing_files: i32,
+ existing_rows: i64,
+ deleted_files: i32,
+ deleted_rows: i64,
+
+ seq_num: i64,
+ min_seq_num: Option<i64>,
+
+ key_metadata: Option<Vec<u8>>,
+
+ field_summary: HashMap<i32, FieldSummary>,
+}
+
+impl ManifestWriter {
+ /// Create a new manifest writer.
+ pub fn new(
+ output: OutputFile,
+ snapshot_id: i64,
+ seq_num: i64,
+ key_metadata: Option<Vec<u8>>,
+ ) -> Self {
+ Self {
+ output,
+ snapshot_id,
+ added_files: 0,
+ added_rows: 0,
+ existing_files: 0,
+ existing_rows: 0,
+ deleted_files: 0,
+ deleted_rows: 0,
+ seq_num,
+ min_seq_num: None,
+ key_metadata,
+ field_summary: HashMap::new(),
+ }
+ }
+
+ fn update_field_summary(&mut self, entry: &ManifestEntry) {
+ // Update field summary
+ if let Some(null) = &entry.data_file.null_value_counts {
+ for (&k, &v) in null {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if v > 0 {
+ field_summary.contains_null = true;
+ }
+ }
+ }
+ if let Some(nan) = &entry.data_file.nan_value_counts {
+ for (&k, &v) in nan {
+ let field_summary = self.field_summary.entry(k).or_default();
+ assert!(v >= 0);
+ if v > 0 {
+ field_summary.contains_nan = Some(true);
+ }
+ if v == 0 {
+ field_summary.contains_nan = Some(false);
+ }
+ }
+ }
+ if let Some(lower_bound) = &entry.data_file.lower_bounds {
+ for (&k, v) in lower_bound {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if let Some(cur) = &field_summary.lower_bound {
+ if v < cur {
+ field_summary.lower_bound = Some(v.clone());
+ }
+ } else {
+ field_summary.lower_bound = Some(v.clone());
+ }
+ }
+ }
+ if let Some(upper_bound) = &entry.data_file.upper_bounds {
+ for (&k, v) in upper_bound {
+ let field_summary = self.field_summary.entry(k).or_default();
+ if let Some(cur) = &field_summary.upper_bound {
+ if v > cur {
+ field_summary.upper_bound = Some(v.clone());
+ }
+ } else {
+ field_summary.upper_bound = Some(v.clone());
+ }
+ }
+ }
+ }
+
+ fn get_field_summary_vec(&mut self, spec_fields: &[PartitionField]) ->
Vec<FieldSummary> {
+ let mut partition_summary =
Vec::with_capacity(self.field_summary.len());
+ for field in spec_fields {
+ let entry = self
+ .field_summary
+ .remove(&field.source_id)
+ .unwrap_or(FieldSummary::default());
+ partition_summary.push(entry);
+ }
+ partition_summary
+ }
+
+ /// Write a manifest entry.
+ pub async fn write(mut self, manifest: Manifest) ->
Result<ManifestListEntry, Error> {
+ // Create the avro writer
+ let partition_type = manifest
+ .metadata
+ .partition_spec
+ .partition_type(&manifest.metadata.schema)?;
+ let table_schema = &manifest.metadata.schema;
+ let avro_schema = match manifest.metadata.format_version {
+ FormatVersion::V1 => manifest_schema_v1(partition_type.clone())?,
+ FormatVersion::V2 => manifest_schema_v2(partition_type.clone())?,
+ };
+ let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new());
+ avro_writer.add_user_metadata(
+ "schema".to_string(),
+ to_vec(table_schema).map_err(|err| {
+ Error::new(ErrorKind::DataInvalid, "Fail to serialize table
schema")
+ .with_source(err)
+ })?,
+ )?;
+ avro_writer.add_user_metadata(
+ "schema-id".to_string(),
+ table_schema.schema_id().to_string(),
+ )?;
+ avro_writer.add_user_metadata(
+ "partition-spec".to_string(),
+ to_vec(&manifest.metadata.partition_spec.fields).map_err(|err| {
+ Error::new(ErrorKind::DataInvalid, "Fail to serialize
partition spec")
+ .with_source(err)
+ })?,
+ )?;
+ avro_writer.add_user_metadata(
+ "partition-spec-id".to_string(),
+ manifest.metadata.partition_spec.spec_id.to_string(),
+ )?;
+ avro_writer.add_user_metadata(
+ "format-version".to_string(),
+ manifest.metadata.format_version.to_string(),
+ )?;
+ if manifest.metadata.format_version == FormatVersion::V2 {
+ avro_writer
+ .add_user_metadata("content".to_string(),
manifest.metadata.content.to_string())?;
+ }
+
+ // Write manifest entries
+ for entry in manifest.entries {
+ match entry.status {
+ ManifestStatus::Added => {
+ self.added_files += 1;
+ self.added_rows += entry.data_file.record_count;
+ }
+ ManifestStatus::Deleted => {
+ self.deleted_files += 1;
+ self.deleted_rows += entry.data_file.record_count;
+ }
+ ManifestStatus::Existing => {
+ self.existing_files += 1;
+ self.existing_rows += entry.data_file.record_count;
+ }
+ }
+
+ if entry.is_alive() {
+ if let Some(cur_min_seq_num) = self.min_seq_num {
+ self.min_seq_num = Some(
+ entry
+ .sequence_number
+ .map(|v| min(v, cur_min_seq_num))
+ .unwrap_or(cur_min_seq_num),
+ );
+ } else {
+ self.min_seq_num = entry.sequence_number;
+ }
+ }
+
+ self.update_field_summary(&entry);
+
+ let value = match manifest.metadata.format_version {
+ FormatVersion::V1 => {
+ to_value(_serde::ManifestEntryV1::try_from(entry,
&partition_type)?)?
+ .resolve(&avro_schema)?
+ }
+ FormatVersion::V2 => {
+ to_value(_serde::ManifestEntryV2::try_from(entry,
&partition_type)?)?
+ .resolve(&avro_schema)?
+ }
+ };
+
+ avro_writer.append(value)?;
+ }
+
+ let length = avro_writer.flush()?;
+ let content = avro_writer.into_inner()?;
+ let mut writer = self.output.writer().await?;
+ writer.write_all(&content).await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "Fail to write Manifest
Entry").with_source(err)
+ })?;
+ writer.close().await.map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "Fail to write Manifest
Entry").with_source(err)
+ })?;
+
+ let partition_summary =
+
self.get_field_summary_vec(&manifest.metadata.partition_spec.fields);
+
+ Ok(ManifestListEntry {
+ manifest_path: self.output.location().to_string(),
+ manifest_length: length as i64,
+ partition_spec_id: manifest.metadata.partition_spec.spec_id,
+ content: manifest.metadata.content,
+ sequence_number: self.seq_num,
+ min_sequence_number:
self.min_seq_num.unwrap_or(UNASSIGNED_SEQ_NUMBER),
+ added_snapshot_id: self.snapshot_id,
+ added_data_files_count: Some(self.added_files),
+ existing_data_files_count: Some(self.existing_files),
+ deleted_data_files_count: Some(self.deleted_files),
+ added_rows_count: Some(self.added_rows),
+ existing_rows_count: Some(self.existing_rows),
+ deleted_rows_count: Some(self.deleted_rows),
+ partitions: partition_summary,
+ key_metadata: self.key_metadata.unwrap_or_default(),
+ })
+ }
+}
+
+/// This is a helper module that defines the schema field of the manifest list
entry.
+mod _const_schema {
+ use std::sync::Arc;
+
+ use apache_avro::Schema as AvroSchema;
+ use once_cell::sync::Lazy;
+
+ use crate::{
+ avro::schema_to_avro_schema,
+ spec::{
+ ListType, MapType, NestedField, NestedFieldRef, PrimitiveType,
Schema, StructType, Type,
+ },
+ Error,
+ };
+
+ pub static STATUS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 0,
+ "status",
+ Type::Primitive(PrimitiveType::Int),
+ ))
+ })
+ };
+
+ pub static SNAPSHOT_ID_V1: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 1,
+ "snapshot_id",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static SNAPSHOT_ID_V2: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 1,
+ "snapshot_id",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 3,
+ "sequence_number",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static FILE_SEQUENCE_NUMBER: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 4,
+ "file_sequence_number",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static CONTENT: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 134,
+ "content",
+ Type::Primitive(PrimitiveType::Int),
+ ))
+ })
+ };
+
+ pub static FILE_PATH: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 100,
+ "file_path",
+ Type::Primitive(PrimitiveType::String),
+ ))
+ })
+ };
+
+ pub static FILE_FORMAT: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 101,
+ "file_format",
+ Type::Primitive(PrimitiveType::String),
+ ))
+ })
+ };
+
+ pub static RECORD_COUNT: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 103,
+ "record_count",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static FILE_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 104,
+ "file_size_in_bytes",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ // Deprecated. Always write a default in v1. Do not write in v2.
+ pub static BLOCK_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::required(
+ 105,
+ "block_size_in_bytes",
+ Type::Primitive(PrimitiveType::Long),
+ ))
+ })
+ };
+
+ pub static COLUMN_SIZES: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 108,
+ "column_sizes",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 117,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 118,
+ "value",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static VALUE_COUNTS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 109,
+ "value_counts",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 119,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 120,
+ "value",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static NULL_VALUE_COUNTS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 110,
+ "null_value_counts",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 121,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 122,
+ "value",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static NAN_VALUE_COUNTS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 137,
+ "nan_value_counts",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 138,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 139,
+ "value",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static DISTINCT_COUNTS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 111,
+ "distinct_counts",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 123,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 124,
+ "value",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static LOWER_BOUNDS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 125,
+ "lower_bounds",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 126,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 127,
+ "value",
+ Type::Primitive(PrimitiveType::Binary),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static UPPER_BOUNDS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 128,
+ "upper_bounds",
+ Type::Map(MapType {
+ key_field: Arc::new(NestedField::required(
+ 129,
+ "key",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ value_field: Arc::new(NestedField::required(
+ 130,
+ "value",
+ Type::Primitive(PrimitiveType::Binary),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static KEY_METADATA: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 131,
+ "key_metadata",
+ Type::Primitive(PrimitiveType::Binary),
+ ))
+ })
+ };
+
+ pub static SPLIT_OFFSETS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 132,
+ "split_offsets",
+ Type::List(ListType {
+ element_field: Arc::new(NestedField::required(
+ 133,
+ "element",
+ Type::Primitive(PrimitiveType::Long),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static EQUALITY_IDS: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 135,
+ "equality_ids",
+ Type::List(ListType {
+ element_field: Arc::new(NestedField::required(
+ 136,
+ "element",
+ Type::Primitive(PrimitiveType::Int),
+ )),
+ }),
+ ))
+ })
+ };
+
+ pub static SORT_ORDER_ID: Lazy<NestedFieldRef> = {
+ Lazy::new(|| {
+ Arc::new(NestedField::optional(
+ 140,
+ "sort_order_id",
+ Type::Primitive(PrimitiveType::Int),
+ ))
+ })
+ };
+
+ pub fn manifest_schema_v2(partition_type: StructType) ->
Result<AvroSchema, Error> {
+ let fields = vec![
+ STATUS.clone(),
+ SNAPSHOT_ID_V2.clone(),
+ SEQUENCE_NUMBER.clone(),
+ FILE_SEQUENCE_NUMBER.clone(),
+ Arc::new(NestedField::required(
+ 2,
+ "data_file",
+ Type::Struct(StructType::new(vec![
+ CONTENT.clone(),
+ FILE_PATH.clone(),
+ FILE_FORMAT.clone(),
+ Arc::new(NestedField::required(
+ 102,
+ "partition",
+ Type::Struct(partition_type),
+ )),
+ RECORD_COUNT.clone(),
+ FILE_SIZE_IN_BYTES.clone(),
+ COLUMN_SIZES.clone(),
+ VALUE_COUNTS.clone(),
+ NULL_VALUE_COUNTS.clone(),
+ NAN_VALUE_COUNTS.clone(),
+ DISTINCT_COUNTS.clone(),
+ LOWER_BOUNDS.clone(),
+ UPPER_BOUNDS.clone(),
+ KEY_METADATA.clone(),
+ SPLIT_OFFSETS.clone(),
+ EQUALITY_IDS.clone(),
+ SORT_ORDER_ID.clone(),
+ ])),
+ )),
+ ];
+ let schema = Schema::builder().with_fields(fields).build().unwrap();
+ schema_to_avro_schema("manifest", &schema)
+ }
+
+ pub fn manifest_schema_v1(partition_type: StructType) ->
Result<AvroSchema, Error> {
+ let fields = vec![
+ STATUS.clone(),
+ SNAPSHOT_ID_V1.clone(),
+ Arc::new(NestedField::required(
+ 2,
+ "data_file",
+ Type::Struct(StructType::new(vec![
+ FILE_PATH.clone(),
+ FILE_FORMAT.clone(),
+ Arc::new(NestedField::required(
+ 102,
+ "partition",
+ Type::Struct(partition_type),
+ )),
+ RECORD_COUNT.clone(),
+ FILE_SIZE_IN_BYTES.clone(),
+ BLOCK_SIZE_IN_BYTES.clone(),
+ COLUMN_SIZES.clone(),
+ VALUE_COUNTS.clone(),
+ NULL_VALUE_COUNTS.clone(),
+ NAN_VALUE_COUNTS.clone(),
+ DISTINCT_COUNTS.clone(),
+ LOWER_BOUNDS.clone(),
+ UPPER_BOUNDS.clone(),
+ KEY_METADATA.clone(),
+ SPLIT_OFFSETS.clone(),
+ SORT_ORDER_ID.clone(),
+ ])),
+ )),
+ ];
+ let schema = Schema::builder().with_fields(fields).build().unwrap();
+ schema_to_avro_schema("manifest", &schema)
+ }
+}
+
+/// Meta data of a manifest that is stored in the key-value metadata of the
Avro file
+#[derive(Debug, PartialEq, Clone, Eq)]
+pub struct ManifestMetadata {
+ /// The table schema at the time the manifest
+ /// was written
+ schema: Schema,
+ /// ID of the schema used to write the manifest as a string
+ schema_id: i32,
+ /// The partition spec used to write the manifest
+ partition_spec: PartitionSpec,
+ /// Table format version number of the manifest as a string
+ format_version: FormatVersion,
+ /// Type of content files tracked by the manifest: “data” or “deletes”
+ content: ManifestContentType,
+}
+
+impl ManifestMetadata {
+ /// Parse from metadata in avro file.
+ pub fn parse(meta: &HashMap<String, Vec<u8>>) -> Result<Self, Error> {
+ let schema = {
+ let bs = meta.get("schema").ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "schema is required in manifest metadata but not found",
+ )
+ })?;
+ serde_json::from_slice::<Schema>(bs).map_err(|err| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Fail to parse schema in manifest metadata",
+ )
+ .with_source(err)
+ })?
+ };
+ let schema_id: i32 = meta
+ .get("schema-id")
+ .map(|bs| {
+ String::from_utf8_lossy(bs).parse().map_err(|err| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Fail to parse schema id in manifest metadata",
+ )
+ .with_source(err)
+ })
+ })
+ .transpose()?
+ .unwrap_or(0);
+ let partition_spec = {
+ let fields = {
+ let bs = meta.get("partition-spec").ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "partition-spec is required in manifest metadata but
not found",
+ )
+ })?;
+
serde_json::from_slice::<Vec<PartitionField>>(bs).map_err(|err| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Fail to parse partition spec in manifest metadata",
+ )
+ .with_source(err)
+ })?
+ };
+ let spec_id = meta
+ .get("partition-spec-id")
+ .map(|bs| {
+ String::from_utf8_lossy(bs).parse().map_err(|err| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Fail to parse partition spec id in manifest
metadata",
+ )
+ .with_source(err)
+ })
+ })
+ .transpose()?
+ .unwrap_or(0);
+ PartitionSpec { spec_id, fields }
+ };
+ let format_version = if let Some(bs) = meta.get("format-version") {
+ serde_json::from_slice::<FormatVersion>(bs).map_err(|err| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Fail to parse format version in manifest metadata",
+ )
+ .with_source(err)
+ })?
+ } else {
+ FormatVersion::V1
+ };
+ let content = if let Some(v) = meta.get("content") {
+ let v = String::from_utf8_lossy(v);
+ v.parse()?
+ } else {
+ ManifestContentType::Data
+ };
+ Ok(ManifestMetadata {
+ schema,
+ schema_id,
+ partition_spec,
+ format_version,
+ content,
+ })
+ }
+}
+
+/// A manifest is an immutable Avro file that lists data files or delete
+/// files, along with each file’s partition data tuple, metrics, and tracking
+/// information.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct ManifestEntry {
+ /// field: 0
+ ///
+ /// Used to track additions and deletions.
+ status: ManifestStatus,
+ /// field id: 1
+ ///
+ /// Snapshot id where the file was added, or deleted if status is 2.
+ /// Inherited when null.
+ snapshot_id: Option<i64>,
+ /// field id: 3
+ ///
+ /// Data sequence number of the file.
+ /// Inherited when null and status is 1 (added).
+ sequence_number: Option<i64>,
+ /// field id: 4
+ ///
+ /// File sequence number indicating when the file was added.
+ /// Inherited when null and status is 1 (added).
+ file_sequence_number: Option<i64>,
+ /// field id: 2
+ ///
+ /// File path, partition tuple, metrics, …
+ data_file: DataFile,
+}
+
+impl ManifestEntry {
+ /// Check if this manifest entry is deleted.
+ pub fn is_alive(&self) -> bool {
+ matches!(
+ self.status,
+ ManifestStatus::Added | ManifestStatus::Existing
+ )
+ }
+}
+
+/// Used to track additions and deletions in ManifestEntry.
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub enum ManifestStatus {
+ /// Value: 0
+ Existing = 0,
+ /// Value: 1
+ Added = 1,
+ /// Value: 2
+ ///
+ /// Deletes are informational only and not used in scans.
+ Deleted = 2,
+}
+
+impl TryFrom<i32> for ManifestStatus {
+ type Error = Error;
+
+ fn try_from(v: i32) -> Result<ManifestStatus, Error> {
+ match v {
+ 0 => Ok(ManifestStatus::Existing),
+ 1 => Ok(ManifestStatus::Added),
+ 2 => Ok(ManifestStatus::Deleted),
+ _ => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!("manifest status {} is invalid", v),
+ )),
+ }
+ }
+}
+
+/// Data file carries data file path, partition tuple, metrics, …
+#[derive(Debug, PartialEq, Clone, Eq)]
+pub struct DataFile {
+ /// field id: 134
+ ///
+ /// Type of content stored by the data file: data, equality deletes,
+ /// or position deletes (all v1 files are data files)
+ content: DataContentType,
+ /// field id: 100
+ ///
+ /// Full URI for the file with FS scheme
+ file_path: String,
+ /// field id: 101
+ ///
+ /// String file format name, avro, orc or parquet
+ file_format: DataFileFormat,
+ /// field id: 102
+ ///
+ /// Partition data tuple, schema based on the partition spec output using
+ /// partition field ids for the struct field ids
+ partition: Struct,
+ /// field id: 103
+ ///
+ /// Number of records in this file
+ record_count: i64,
+ /// field id: 104
+ ///
+ /// Total file size in bytes
+ file_size_in_bytes: i64,
+ /// field id: 108
+ /// key field id: 117
+ /// value field id: 118
+ ///
+ /// Map from column id to the total size on disk of all regions that
+ /// store the column. Does not include bytes necessary to read other
+ /// columns, like footers. Leave null for row-oriented formats (Avro)
+ column_sizes: Option<HashMap<i32, i64>>,
Review Comment:
Using `HashMap` directly? This will simplify operators on these fields. Same
as belowing fields.
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
