View Spec implementation (#331)

* Add support for ViewSpec

* Fix typos

* Fix typos

* clippy is always right

* Add tests

* Remove new_view_version test function

* Remove append_version

* View Representations Struct

* ViewRepresentation case insensitive

* Add fallible methods for ViewRepresentationsBuilder

* Add tests for fallibe ViewRepresentationsBuilder methods

* Introduce ViewVersionId as i32

* Iterator for &'a ViewRepresentations

* Improve comments

Co-authored-by: Renjie Liu <[email protected]>

* Add test_view_metadata_v1_file_valid

* Fix view_version iter

* Remove ViewRepresentationsBuilder

* Fix comment

* Timestamp error handling

* Fallible Timestamp Conversion from Millis

* Fix Initial view Version = 1

* Cleanup

* Hide ViewMetadata iter() type

* timestamp_ms_to_utc -> error.rs

* TableMetadata timestamp conversion -> utility function

* Improve error context

* timestamp_ms_to_utc: LocalResult::None -> DataInvalid

* Fix obsolete comment

* ViewRepresentation::SqlViewRepresentation -> ::Sql

* Fix broken clippy from rebase

---------

Co-authored-by: Renjie Liu <[email protected]>

Author: c-thiel

Cargo.toml

@@ -85,7 +85,8 @@ tempfile = "3.8"
 tokio = { version = "1", default-features = false }
 typed-builder = "^0.19"
 url = "2"
-uuid = "1.6.1"
+urlencoding = "2"
+uuid = { version = "1.6.1", features = ["v7"] }
 volo-thrift = "0.10"
 hive_metastore = "0.1.0"
 tera = "1"

crates/catalog/rest/src/catalog.rs

@@ -1321,7 +1321,7 @@ mod tests {
         );
         assert_eq!(
             Utc.timestamp_millis_opt(1646787054459).unwrap(),
-            table.metadata().last_updated_ms()
+            table.metadata().last_updated_timestamp().unwrap()
         );
         assert_eq!(
             vec![&Arc::new(
@@ -1511,7 +1511,11 @@ mod tests {
         );
         assert_eq!(
             1657810967051,
-            table.metadata().last_updated_ms().timestamp_millis()
+            table
+                .metadata()
+                .last_updated_timestamp()
+                .unwrap()
+                .timestamp_millis()
         );
         assert_eq!(
             vec![&Arc::new(
@@ -1682,7 +1686,11 @@ mod tests {
         );
         assert_eq!(
             1657810967051,
-            table.metadata().last_updated_ms().timestamp_millis()
+            table
+                .metadata()
+                .last_updated_timestamp()
+                .unwrap()
+                .timestamp_millis()
         );
         assert_eq!(
             vec![&Arc::new(

crates/iceberg/src/catalog/mod.rs

@@ -29,7 +29,7 @@ use uuid::Uuid;
 
 use crate::spec::{
     FormatVersion, Schema, Snapshot, SnapshotReference, SortOrder, TableMetadataBuilder,
-    UnboundPartitionSpec,
+    UnboundPartitionSpec, ViewRepresentations,
 };
 use crate::table::Table;
 use crate::{Error, ErrorKind, Result};
@@ -439,6 +439,31 @@ impl TableUpdate {
     }
 }
 
+/// ViewCreation represents the creation of a view in the catalog.
+#[derive(Debug, TypedBuilder)]
+pub struct ViewCreation {
+    /// The name of the view.
+    pub name: String,
+    /// The view's base location; used to create metadata file locations
+    pub location: String,
+    /// Representations for the view.
+    pub representations: ViewRepresentations,
+    /// The schema of the view.
+    pub schema: Schema,
+    /// The properties of the view.
+    #[builder(default)]
+    pub properties: HashMap<String, String>,
+    /// The default namespace to use when a reference in the SELECT is a single identifier
+    pub default_namespace: NamespaceIdent,
+    /// Default catalog to use when a reference in the SELECT does not contain a catalog
+    #[builder(default)]
+    pub default_catalog: Option<String>,
+    /// A string to string map of summary metadata about the version
+    /// Typical keys are "engine-name" and "engine-version"
+    #[builder(default)]
+    pub summary: HashMap<String, String>,
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

crates/iceberg/src/error.rs

@@ -19,6 +19,8 @@ use std::backtrace::{Backtrace, BacktraceStatus};
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
 
+use chrono::{DateTime, TimeZone as _, Utc};
+
 /// Result that is a wrapper of `Result<T, iceberg::Error>`
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -331,6 +333,28 @@ define_from_err!(
 
 define_from_err!(std::io::Error, ErrorKind::Unexpected, "IO Operation failed");
 
+/// Converts a timestamp in milliseconds to `DateTime<Utc>`, handling errors.
+///
+/// # Arguments
+///
+/// * `timestamp_ms` - The timestamp in milliseconds to convert.
+///
+/// # Returns
+///
+/// This function returns a `Result<DateTime<Utc>, Error>` which is `Ok` with the `DateTime<Utc>` if the conversion is successful,
+/// or an `Err` with an appropriate error if the timestamp is ambiguous or invalid.
+pub(crate) fn timestamp_ms_to_utc(timestamp_ms: i64) -> Result<DateTime<Utc>> {
+    match Utc.timestamp_millis_opt(timestamp_ms) {
+        chrono::LocalResult::Single(t) => Ok(t),
+        chrono::LocalResult::Ambiguous(_, _) => Err(Error::new(
+            ErrorKind::Unexpected,
+            "Ambiguous timestamp with two possible results",
+        )),
+        chrono::LocalResult::None => Err(Error::new(ErrorKind::DataInvalid, "Invalid timestamp")),
+    }
+    .map_err(|e| e.with_context("timestamp value", timestamp_ms.to_string()))
+}
+
 /// Helper macro to check arguments.
 ///
 ///

crates/iceberg/src/lib.rs

@@ -29,7 +29,7 @@ mod catalog;
 
 pub use catalog::{
     Catalog, Namespace, NamespaceIdent, TableCommit, TableCreation, TableIdent, TableRequirement,
-    TableUpdate,
+    TableUpdate, ViewCreation,
 };
 
 pub mod table;

crates/iceberg/src/spec/mod.rs

@@ -27,6 +27,8 @@ mod sort;
 mod table_metadata;
 mod transform;
 mod values;
+mod view_metadata;
+mod view_version;
 
 pub use datatypes::*;
 pub use manifest::*;
@@ -38,3 +40,5 @@ pub use sort::*;
 pub use table_metadata::*;
 pub use transform::*;
 pub use values::*;
+pub use view_metadata::*;
+pub use view_version::*;

crates/iceberg/src/spec/table_metadata.rs

@@ -24,7 +24,7 @@ use std::fmt::{Display, Formatter};
 use std::sync::Arc;
 
 use _serde::TableMetadataEnum;
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 use uuid::Uuid;
@@ -33,7 +33,7 @@ use super::snapshot::{Snapshot, SnapshotReference, SnapshotRetention};
 use super::{
     PartitionSpec, PartitionSpecRef, SchemaId, SchemaRef, SnapshotRef, SortOrder, SortOrderRef,
 };
-use crate::error::Result;
+use crate::error::{timestamp_ms_to_utc, Result};
 use crate::{Error, ErrorKind, TableCreation};
 
 static MAIN_BRANCH: &str = "main";
@@ -143,8 +143,14 @@ impl TableMetadata {
 
     /// Returns last updated time.
     #[inline]
-    pub fn last_updated_ms(&self) -> DateTime<Utc> {
-        Utc.timestamp_millis_opt(self.last_updated_ms).unwrap()
+    pub fn last_updated_timestamp(&self) -> Result<DateTime<Utc>> {
+        timestamp_ms_to_utc(self.last_updated_ms)
+    }
+
+    /// Returns last updated time in milliseconds.
+    #[inline]
+    pub fn last_updated_ms(&self) -> i64 {
+        self.last_updated_ms
     }
 
     /// Returns schemas
@@ -328,7 +334,7 @@ impl TableMetadataBuilder {
 
         let table_metadata = TableMetadata {
             format_version: FormatVersion::V2,
-            table_uuid: Uuid::new_v4(),
+            table_uuid: Uuid::now_v7(),
             location: location.ok_or_else(|| {
                 Error::new(
                     ErrorKind::DataInvalid,
@@ -472,7 +478,7 @@ pub(super) mod _serde {
 
     /// Helper to serialize and deserialize the format version.
     #[derive(Debug, PartialEq, Eq)]
-    pub(super) struct VersionNumber<const V: u8>;
+    pub(crate) struct VersionNumber<const V: u8>;
 
     impl Serialize for TableMetadata {
         fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
@@ -903,8 +909,14 @@ pub struct SnapshotLog {
 
 impl SnapshotLog {
     /// Returns the last updated timestamp as a DateTime<Utc> with millisecond precision
-    pub fn timestamp(self) -> DateTime<Utc> {
-        Utc.timestamp_millis_opt(self.timestamp_ms).unwrap()
+    pub fn timestamp(self) -> Result<DateTime<Utc>> {
+        timestamp_ms_to_utc(self.timestamp_ms)
+    }
+
+    /// Returns the timestamp in milliseconds
+    #[inline]
+    pub fn timestamp_ms(&self) -> i64 {
+        self.timestamp_ms
     }
 }