wip: gotta figure out how to handle structs in structs with create_one

Author: zachschuermann

ffi/src/expressions/kernel.rs

@@ -333,6 +333,7 @@ fn visit_expression_internal(
                 visit_expression_struct_literal(visitor, struct_data, sibling_list_id)
             }
             Scalar::Array(array) => visit_expression_array(visitor, array, sibling_list_id),
+            Scalar::Map(_map_data) => todo!(),
         }
     }
     fn visit_expression_impl(

kernel/src/actions/mod.rs

@@ -91,7 +91,7 @@ pub(crate) fn get_log_commit_info_schema() -> &'static SchemaRef {
     &LOG_COMMIT_INFO_SCHEMA
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Schema)]
+#[derive(Debug, Clone, PartialEq, Eq, Schema, IntoEngineData)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(test, derive(Serialize), serde(rename_all = "camelCase"))]
 pub(crate) struct Format {
@@ -110,7 +110,7 @@ impl Default for Format {
     }
 }
 
-#[derive(Debug, Default, Clone, PartialEq, Eq, Schema)]
+#[derive(Debug, Default, Clone, PartialEq, Eq, Schema, IntoEngineData)]
 #[cfg_attr(test, derive(Serialize), serde(rename_all = "camelCase"))]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 pub(crate) struct Metadata {
@@ -133,6 +133,19 @@ pub(crate) struct Metadata {
 }
 
 impl Metadata {
+    pub(crate) fn new(name: String, schema: StructType, partition_columns: Vec<String>) -> Self {
+        Self {
+            id: uuid::Uuid::new_v4().to_string(),
+            name: Some(name),
+            description: None,
+            format: Format::default(),
+            schema_string: serde_json::to_string(&schema).unwrap_or_default(),
+            partition_columns,
+            created_time: Some(chrono::Utc::now().timestamp_millis()),
+            configuration: HashMap::new(),
+        }
+    }
+
     pub(crate) fn try_new_from_data(data: &dyn EngineData) -> DeltaResult<Option<Metadata>> {
         let mut visitor = MetadataVisitor::default();
         visitor.visit_rows_of(data)?;
@@ -163,7 +176,7 @@ impl Metadata {
     }
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Schema, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Schema, Serialize, Deserialize, IntoEngineData)]
 #[serde(rename_all = "camelCase")]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 // TODO move to another module so that we disallow constructing this struct without using the
@@ -185,6 +198,17 @@ pub(crate) struct Protocol {
     writer_features: Option<Vec<WriterFeature>>,
 }
 
+impl Default for Protocol {
+    fn default() -> Self {
+        Protocol {
+            min_reader_version: 3,
+            min_writer_version: 7,
+            reader_features: Some(SUPPORTED_READER_FEATURES.to_vec()),
+            writer_features: Some(SUPPORTED_WRITER_FEATURES.to_vec()),
+        }
+    }
+}
+
 fn parse_features<T>(features: Option<impl IntoIterator<Item = impl ToString>>) -> Option<Vec<T>>
 where
     T: FromStr,

kernel/src/engine/arrow_expression/mod.rs

@@ -84,6 +84,7 @@ impl Scalar {
                     None,
                 ))
             }
+            Map(map_data) => todo!(),
             Null(DataType::BYTE) => Arc::new(Int8Array::new_null(num_rows)),
             Null(DataType::SHORT) => Arc::new(Int16Array::new_null(num_rows)),
             Null(DataType::INTEGER) => Arc::new(Int32Array::new_null(num_rows)),

kernel/src/expressions/literal_expression_transform.rs

@@ -160,21 +160,57 @@ impl<'a, T: Iterator<Item = &'a Scalar>> SchemaTransform<'a> for LiteralExpressi
         Some(Cow::Borrowed(field))
     }
 
-    // arrays unsupported for now
-    fn transform_array(&mut self, _array_type: &'a ArrayType) -> Option<Cow<'a, ArrayType>> {
+    // arrays treated as leaves
+    fn transform_array(&mut self, array_type: &'a ArrayType) -> Option<Cow<'a, ArrayType>> {
+        // first always check error to terminate early if possible
         self.error.as_ref().ok()?;
-        self.set_error(Error::Unsupported(
-            "ArrayType not yet supported in literal expression transform".to_string(),
-        ));
+        let Some(scalar) = self.scalars.next() else {
+            self.set_error(Error::InsufficientScalars);
+            return None;
+        };
+
+        let DataType::Array(scalar_type) = scalar.data_type() else {
+            self.set_error(Error::Schema(
+                "Non-array scalar type {datatype} provided for array leaf".to_string(),
+            ));
+            return None;
+        };
+        if scalar_type.as_ref() != array_type {
+            self.set_error(Error::Schema(format!(
+                "Mismatched scalar type while creating Expression: expected {:?}, got {:?}",
+                array_type, scalar_type
+            )));
+            return None;
+        }
+
+        self.stack.push(Expression::Literal(scalar.clone()));
         None
     }
 
-    // maps unsupported for now
-    fn transform_map(&mut self, _map_type: &'a MapType) -> Option<Cow<'a, MapType>> {
+    // maps treated as leaves
+    fn transform_map(&mut self, map_type: &'a MapType) -> Option<Cow<'a, MapType>> {
+        // first always check error to terminate early if possible
         self.error.as_ref().ok()?;
-        self.set_error(Error::Unsupported(
-            "MapType not yet supported in literal expression transform".to_string(),
-        ));
+        let Some(scalar) = self.scalars.next() else {
+            self.set_error(Error::InsufficientScalars);
+            return None;
+        };
+
+        let DataType::Map(scalar_type) = scalar.data_type() else {
+            self.set_error(Error::Schema(
+                "Non-map scalar type {datatype} provided for map leaf".to_string(),
+            ));
+            return None;
+        };
+        if scalar_type.as_ref() != map_type {
+            self.set_error(Error::Schema(format!(
+                "Mismatched scalar type while creating Expression: expected {:?}, got {:?}",
+                map_type, scalar_type
+            )));
+            return None;
+        }
+
+        self.stack.push(Expression::Literal(scalar.clone()));
         None
     }
 }

kernel/src/expressions/scalars.rs

@@ -1,10 +1,11 @@
 use std::cmp::Ordering;
+use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 
 use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
 
 use crate::actions::schemas::ToDataType;
-use crate::schema::{ArrayType, DataType, PrimitiveType, StructField};
+use crate::schema::{ArrayType, DataType, MapType, PrimitiveType, StructField};
 use crate::utils::require;
 use crate::{DeltaResult, Error};
 
@@ -22,11 +23,11 @@ pub struct ArrayData {
 }
 
 impl ArrayData {
-    #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
     pub fn new(tpe: ArrayType, elements: impl IntoIterator<Item = impl Into<Scalar>>) -> Self {
         let elements = elements.into_iter().map(Into::into).collect();
         Self { tpe, elements }
     }
+
     pub fn array_type(&self) -> &ArrayType {
         &self.tpe
     }
@@ -89,6 +90,12 @@ impl StructData {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct MapData {
+    data_type: MapType,
+    elements: Vec<(Scalar, Scalar)>,
+}
+
 /// A single value, which can be null. Used for representing literal values
 /// in [Expressions][crate::expressions::Expression].
 #[derive(Debug, Clone)]
@@ -125,6 +132,8 @@ pub enum Scalar {
     Struct(StructData),
     /// Array Value
     Array(ArrayData),
+    /// Map Value
+    Map(MapData),
 }
 
 impl Scalar {
@@ -146,6 +155,7 @@ impl Scalar {
             Self::Null(data_type) => data_type.clone(),
             Self::Struct(data) => DataType::struct_type(data.fields.clone()),
             Self::Array(data) => data.tpe.clone().into(),
+            Self::Map(map_data) => map_data.data_type.clone().into(),
         }
     }
 
@@ -222,6 +232,7 @@ impl Display for Scalar {
                 }
                 write!(f, ")")
             }
+            Self::Map(_map_data) => todo!(),
         }
     }
 }
@@ -269,6 +280,7 @@ impl PartialOrd for Scalar {
             (Null(_), _) => None, // NOTE: NULL values are incomparable by definition
             (Struct(_), _) => None, // TODO: Support Struct?
             (Array(_), _) => None, // TODO: Support Array?
+            (Map(_), _) => None,  // TODO: Support Map?
         }
     }
 }
@@ -348,6 +360,53 @@ impl<T: Into<Scalar> + ToDataType> From<Option<T>> for Scalar {
     }
 }
 
+impl<T: Into<Scalar> + ToDataType> From<Vec<T>> for Scalar {
+    fn from(v: Vec<T>) -> Self {
+        Scalar::Array(ArrayData::new(ArrayType::new(T::to_data_type(), true), v))
+    }
+}
+
+impl<K, V> From<HashMap<K, V>> for Scalar
+where
+    K: Into<Scalar> + ToDataType,
+    V: Into<Scalar> + ToDataType,
+{
+    fn from(map: HashMap<K, V>) -> Self {
+        Scalar::Map(MapData {
+            data_type: MapType::new(K::to_data_type(), V::to_data_type(), true),
+            elements: map.into_iter().map(|(k, v)| (k.into(), v.into())).collect(),
+        })
+    }
+}
+
+use crate::actions::Format;
+impl From<Format> for Scalar {
+    fn from(format: Format) -> Self {
+        let map_type = MapType::new(DataType::STRING, DataType::STRING, true);
+        let fields = vec![
+            StructField::new("provider", DataType::STRING, false),
+            StructField::new("options", DataType::Map(Box::new(map_type)), true),
+        ];
+        let values = vec![
+            Scalar::String(format.provider.to_string()),
+            format.options.into(),
+        ];
+        Scalar::Struct(StructData::try_new(fields, values).unwrap())
+    }
+}
+
+use crate::table_features::{ReaderFeature, WriterFeature};
+impl From<ReaderFeature> for Scalar {
+    fn from(feature: ReaderFeature) -> Self {
+        Scalar::String(feature.to_string())
+    }
+}
+impl From<WriterFeature> for Scalar {
+    fn from(feature: WriterFeature) -> Self {
+        Scalar::String(feature.to_string())
+    }
+}
+
 // TODO: add more From impls
 
 impl PrimitiveType {

kernel/src/lib.rs

@@ -376,6 +376,7 @@ trait EvaluationHandlerExtension: EvaluationHandler {
         let null_row = self.null_row(null_row_schema.clone())?;
 
         // Convert schema and leaf values to an expression
+        println!("{:?}, {:?}", schema, values);
         let mut schema_transform = LiteralExpressionTransform::new(values);
         schema_transform.transform_struct(schema.as_ref());
         let row_expr = schema_transform.try_into_expr()?;

kernel/src/table.rs

@@ -7,10 +7,13 @@ use std::path::PathBuf;
 
 use url::Url;
 
+use crate::actions::{Metadata, Protocol};
+use crate::path::ParsedLogPath;
 use crate::snapshot::Snapshot;
 use crate::table_changes::TableChanges;
 use crate::transaction::Transaction;
-use crate::{DeltaResult, Engine, Error, Version};
+use crate::IntoEngineData;
+use crate::{schema::Schema, DeltaResult, Engine, Error, Version};
 
 /// In-memory representation of a Delta table, which acts as an immutable root entity for reading
 /// the different versions (see [`Snapshot`]) of the table located in storage.
@@ -28,11 +31,38 @@ impl std::fmt::Debug for Table {
 }
 
 impl Table {
-    /// Create a new Delta table with the given parameters
+    /// Read a Delta table from the given location
     pub fn new(location: Url) -> Self {
         Self { location }
     }
 
+    /// Create a new Delta table with a given schema (with default protocol/metadata) at the
+    /// location provided
+    pub fn create(
+        name: &str,
+        schema: Schema,
+        location: Url,
+        engine: &dyn Engine,
+    ) -> DeltaResult<Self> {
+        let protocol = Protocol::default();
+        let metadata = Metadata::new(name.to_string(), schema, vec![]);
+
+        let actions = vec![
+            protocol.into_engine_data(engine),
+            metadata.into_engine_data(engine),
+        ];
+
+        let json_handler = engine.json_handler();
+        let commit_path = ParsedLogPath::new_commit(&location, 0)?;
+        json_handler.write_json_file(
+            &commit_path.location,
+            Box::new(actions.into_iter()),
+            false,
+        )?;
+
+        Ok(Self::new(location))
+    }
+
     /// Try to create a new table from a string uri. This will do it's best to handle things like
     /// `/local/paths`, and even `../relative/paths`.
     pub fn try_from_uri(uri: impl AsRef<str>) -> DeltaResult<Self> {

kernel/tests/create.rs

@@ -0,0 +1,16 @@
+use delta_kernel::engine::sync::SyncEngine;
+use delta_kernel::schema::{DataType, Schema, StructField};
+use delta_kernel::Table;
+
+use url::Url;
+
+#[test]
+fn test_create_table() {
+    let engine = SyncEngine::new();
+    let schema = Schema::new([StructField::nullable("id", DataType::INTEGER)]);
+    let location = Url::parse("file:////Users/zach.schuermann/Desktop/test_table/").unwrap();
+    let table = Table::create("test_table", schema, location, &engine).unwrap();
+
+    let snapshot = table.snapshot(&engine, None).unwrap();
+    assert_eq!(snapshot.version(), 0);
+}