icepick 0.4.1

Experimental Rust client for Apache Iceberg with WASM support for AWS S3 Tables and Cloudflare R2
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

//! Manifest entry parsing

use super::extract::*;
use super::{DataFileEntry, DataFileStats, ManifestFileInfo};
use crate::error::{Error, Result};
use apache_avro::types::Value;
use std::collections::HashMap;

/// Parse a manifest file info record from Avro fields
pub(super) fn parse_manifest_file_info(fields: Vec<(String, Value)>) -> Result<ManifestFileInfo> {
    let mut info = ManifestFileInfo::default();
    for (name, field_value) in fields {
        match name.as_str() {
            "manifest_path" => {
                info.manifest_path = extract_required_string(&field_value, "manifest_path")?
            }
            "manifest_length" => {
                info.manifest_length = extract_required_long(&field_value, "manifest_length")?
            }
            "partition_spec_id" => info.partition_spec_id = extract_int(&field_value).unwrap_or(0),
            "content" => info.content = extract_int(&field_value).unwrap_or(0),
            "sequence_number" => info.sequence_number = extract_long(&field_value).unwrap_or(0),
            "min_sequence_number" => {
                info.min_sequence_number = extract_long(&field_value).unwrap_or(0)
            }
            "added_snapshot_id" => info.added_snapshot_id = extract_long(&field_value).unwrap_or(0),
            "added_files_count" => info.added_files_count = extract_int(&field_value).unwrap_or(0),
            "existing_files_count" => {
                info.existing_files_count = extract_int(&field_value).unwrap_or(0)
            }
            "deleted_files_count" => {
                info.deleted_files_count = extract_int(&field_value).unwrap_or(0)
            }
            "added_rows_count" => info.added_rows_count = extract_long(&field_value).unwrap_or(0),
            "existing_rows_count" => {
                info.existing_rows_count = extract_long(&field_value).unwrap_or(0)
            }
            "deleted_rows_count" => {
                info.deleted_rows_count = extract_long(&field_value).unwrap_or(0)
            }
            _ => {}
        }
    }

    // Validate required fields
    if info.manifest_path.is_empty() {
        return Err(Error::invalid_input(
            "manifest_path is required but missing or empty".to_string(),
        ));
    }

    Ok(info)
}

/// Extract status and data_file from manifest entry fields
pub(super) fn extract_manifest_entry_parts(
    fields: Vec<(String, Value)>,
) -> (Option<i32>, Option<Value>) {
    let mut status = None;
    let mut data_file_value = None;
    for (name, field_value) in fields {
        match name.as_str() {
            "status" => status = extract_int(&field_value),
            "data_file" => data_file_value = Some(field_value),
            _ => {}
        }
    }
    (status, data_file_value)
}

/// Parse data file basic fields from Avro record
pub(super) fn parse_data_file_basic(fields: Vec<(String, Value)>) -> Result<DataFileEntry> {
    let mut file_path = None;
    let mut file_format = None;
    let mut record_count = None;
    let mut file_size = None;

    for (name, field_value) in fields {
        match name.as_str() {
            "file_path" => file_path = extract_string(&field_value),
            "file_format" => file_format = extract_string(&field_value),
            "record_count" => record_count = extract_long(&field_value),
            "file_size_in_bytes" => file_size = extract_long(&field_value),
            _ => {}
        }
    }

    Ok(DataFileEntry {
        file_path: file_path.ok_or_else(|| missing_field_error("file_path"))?,
        file_format: file_format.ok_or_else(|| missing_field_error("file_format"))?,
        record_count: record_count.ok_or_else(|| missing_field_error("record_count"))?,
        file_size_in_bytes: file_size.ok_or_else(|| missing_field_error("file_size_in_bytes"))?,
    })
}

/// Parse manifest entry with full stats, skipping deleted entries
pub(super) fn parse_manifest_entry_with_stats(
    fields: Vec<(String, Value)>,
) -> Result<Option<DataFileStats>> {
    let (status, data_file_value) = extract_manifest_entry_parts(fields);
    if status == Some(2) {
        return Ok(None);
    }
    if let Some(Value::Record(data_file_fields)) = data_file_value {
        Ok(Some(parse_data_file_stats(data_file_fields)?))
    } else {
        Err(missing_field_error("data_file"))
    }
}

/// Parse a manifest entry and extract data file entry if not deleted
pub(super) fn parse_manifest_entry(fields: Vec<(String, Value)>) -> Result<Option<DataFileEntry>> {
    let (status, data_file_value) = extract_manifest_entry_parts(fields);
    if status == Some(2) {
        return Ok(None);
    }

    if let Some(Value::Record(data_file_fields)) = data_file_value {
        Ok(Some(parse_data_file_basic(data_file_fields)?))
    } else {
        Err(missing_field_error("data_file"))
    }
}

/// Parse a data_file record into DataFileStats
pub(super) fn parse_data_file_stats(fields: Vec<(String, Value)>) -> Result<DataFileStats> {
    let mut file_path = None;
    let mut file_format = None;
    let mut record_count = None;
    let mut file_size = None;
    let mut partition = HashMap::new();
    let mut lower_bounds = HashMap::new();
    let mut upper_bounds = HashMap::new();
    let mut null_value_counts = HashMap::new();
    let mut value_counts = HashMap::new();

    for (name, field_value) in fields {
        match name.as_str() {
            "file_path" => file_path = extract_string(&field_value),
            "file_format" => file_format = extract_string(&field_value),
            "record_count" => record_count = extract_long(&field_value),
            "file_size_in_bytes" => file_size = extract_long(&field_value),
            "partition" => partition = extract_partition_values(&field_value),
            "lower_bounds" => lower_bounds = extract_bounds_map(&field_value),
            "upper_bounds" => upper_bounds = extract_bounds_map(&field_value),
            "null_value_counts" => null_value_counts = extract_count_map(&field_value),
            "value_counts" => value_counts = extract_count_map(&field_value),
            _ => {}
        }
    }

    Ok(DataFileStats {
        file_path: file_path.ok_or_else(|| missing_field_error("file_path"))?,
        file_format: file_format.ok_or_else(|| missing_field_error("file_format"))?,
        record_count: record_count.ok_or_else(|| missing_field_error("record_count"))?,
        file_size_in_bytes: file_size.ok_or_else(|| missing_field_error("file_size_in_bytes"))?,
        partition,
        lower_bounds,
        upper_bounds,
        null_value_counts,
        value_counts,
    })
}