[write] add Transaction with commit info and commit implementation

ffi/src/lib.rs

@@ -331,6 +331,7 @@ pub enum KernelError {
     InvalidLogPath,
     InvalidCommitInfo,
     FileAlreadyExists,
+    MissingCommitInfo,
 }
 
 impl From<Error> for KernelError {
@@ -380,6 +381,7 @@ impl From<Error> for KernelError {
             Error::InvalidLogPath(_) => KernelError::InvalidLogPath,
             Error::InvalidCommitInfo(_) => KernelError::InvalidCommitInfo,
             Error::FileAlreadyExists(_) => KernelError::FileAlreadyExists,
+            Error::MissingCommitInfo => KernelError::MissingCommitInfo,
         }
     }
 }

kernel/src/actions/mod.rs

@@ -46,6 +46,10 @@ static LOG_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
     .into()
 });
 
+static LOG_COMMIT_INFO_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
+    StructType::new([Option::<CommitInfo>::get_struct_field(COMMIT_INFO_NAME)]).into()
+});
+
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
 fn get_log_schema() -> &'static SchemaRef {
@@ -58,6 +62,10 @@ fn get_log_add_schema() -> &'static SchemaRef {
     &LOG_ADD_SCHEMA
 }
 
+pub(crate) fn get_log_commit_info_schema() -> &'static SchemaRef {
+    &LOG_COMMIT_INFO_SCHEMA
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
 pub struct Format {
     /// Name of the encoding for files in this table
@@ -149,22 +157,21 @@ impl Protocol {
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
-// TODO need to have a way to always write some fields but not always read them
 struct CommitInfo {
     /// The time this logical file was created, as milliseconds since the epoch.
-    /// TODO should this be a Timestamp?
-    pub(crate) timestamp: i64,
+    pub(crate) timestamp: Option<i64>,
     /// An arbitrary string that identifies the operation associated with this commit. This is
     /// specified by the engine.
-    pub(crate) operation: String,
+    pub(crate) operation: Option<String>,
     /// Map of arbitrary string key-value pairs that provide additional information about the
-    /// operation. This is specified by the engine. For now this is always empty.
-    pub(crate) operation_parameters: HashMap<String, String>,
-    /// The version of the delta_kernel crate used to write this commit.
+    /// operation. This is specified by the engine. For now this is always empty on write.
+    pub(crate) operation_parameters: Option<HashMap<String, String>>,
+    /// The version of the delta_kernel crate used to write this commit. The kernel will always
+    /// write this field, but it is optional since many tables will not have this field (i.e. any
+    /// tables not written by kernel).
     pub(crate) kernel_version: Option<String>,
     /// A place for the engine to store additional metadata associated with this commit encoded as
     /// a map of strings.
-    /// TODO need to have a way to always write this but not always read it
     pub(crate) engine_commit_info: Option<HashMap<String, String>>,
 }
 
@@ -454,12 +461,12 @@ mod tests {
         let expected = Arc::new(StructType::new(vec![StructField::new(
             "commitInfo",
             StructType::new(vec![
-                StructField::new("timestamp", DataType::LONG, false),
-                StructField::new("operation", DataType::STRING, false),
+                StructField::new("timestamp", DataType::LONG, true),
+                StructField::new("operation", DataType::STRING, true),
                 StructField::new(
                     "operationParameters",
                     MapType::new(DataType::STRING, DataType::STRING, false),
-                    false,
+                    true,
                 ),
                 StructField::new("kernelVersion", DataType::STRING, true),
                 StructField::new(

kernel/src/engine/arrow_expression.rs

@@ -29,7 +29,7 @@ use crate::engine::arrow_utils::prim_array_cmp;
 use crate::engine::ensure_data_types::ensure_data_types;
 use crate::error::{DeltaResult, Error};
 use crate::expressions::{BinaryOperator, Expression, Scalar, UnaryOperator, VariadicOperator};
-use crate::schema::{DataType, MapType, PrimitiveType, SchemaRef};
+use crate::schema::{ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, StructField};
 use crate::{EngineData, ExpressionEvaluator, ExpressionHandler};
 
 // TODO leverage scalars / Datum
@@ -123,33 +123,7 @@ impl Scalar {
                         ArrowField::new(LIST_ARRAY_ROOT, t.element_type().try_into()?, true);
                     Arc::new(ListArray::new_null(Arc::new(field), num_rows))
                 }
-                DataType::Map(t) => {
-                    let MapType {
-                        key_type,
-                        value_type,
-                        .. // FIXME should use value_contains_null
-                    } = t.as_ref();
-                    if key_type != &DataType::STRING && value_type != &DataType::STRING {
-                        return Err(Error::generic("Only string key/values are supported"));
-                    }
-                    use arrow_array::builder::StringBuilder;
-                    let key_builder = StringBuilder::new();
-                    let val_builder = StringBuilder::new();
-                    let names = arrow_array::builder::MapFieldNames {
-                        entry: "entries".to_string(),
-                        key: "key".to_string(),
-                        value: "value".to_string(),
-                    };
-                    let mut builder = arrow_array::builder::MapBuilder::new(
-                        Some(names),
-                        key_builder,
-                        val_builder,
-                    );
-                    (0..num_rows).for_each(|_| {
-                        builder.append(true).unwrap();
-                    });
-                    Arc::new(builder.finish())
-                }
+                DataType::Map { .. } => unimplemented!(),
             },
         };
         Ok(arr)

kernel/src/engine/arrow_utils.rs

@@ -664,9 +664,9 @@ fn parse_json_impl(json_strings: &StringArray, schema: ArrowSchemaRef) -> DeltaR
 
 /// write an arrow RecordBatch to a JSON string by appending to a buffer.
 ///
-/// TODO: this should stream data to the JSON writer and output an iterator.
-pub(crate) fn write_json(
-    data: Box<dyn Iterator<Item = Box<dyn EngineData>> + Send>,
+/// TODO (zach): this should stream data to the JSON writer and output an iterator.
+pub(crate) fn to_json_bytes(
+    data: impl Iterator<Item = Box<dyn EngineData>> + Send,
 ) -> DeltaResult<Vec<u8>> {
     let mut writer = LineDelimitedWriter::new(Vec::new());
     for chunk in data.into_iter() {
@@ -1425,63 +1425,6 @@ mod tests {
         assert_eq!(reorder_indices, expect_reorder);
     }
 
-    #[test]
-    fn accepts_safe_decimal_casts() {
-        use super::can_upcast_to_decimal;
-        use ArrowDataType::*;
-
-        assert!(can_upcast_to_decimal(&Decimal128(1, 0), 2u8, 0i8));
-        assert!(can_upcast_to_decimal(&Decimal128(1, 0), 2u8, 1i8));
-        assert!(can_upcast_to_decimal(&Decimal128(5, -2), 6u8, -2i8));
-        assert!(can_upcast_to_decimal(&Decimal128(5, -2), 6u8, -1i8));
-        assert!(can_upcast_to_decimal(&Decimal128(5, 1), 6u8, 1i8));
-        assert!(can_upcast_to_decimal(&Decimal128(5, 1), 6u8, 2i8));
-        assert!(can_upcast_to_decimal(
-            &Decimal128(10, 5),
-            arrow_schema::DECIMAL128_MAX_PRECISION,
-            arrow_schema::DECIMAL128_MAX_SCALE - 5
-        ));
-
-        assert!(can_upcast_to_decimal(&Int8, 3u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int8, 4u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int8, 4u8, 1i8));
-        assert!(can_upcast_to_decimal(&Int8, 7u8, 2i8));
-
-        assert!(can_upcast_to_decimal(&Int16, 5u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int16, 6u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int16, 6u8, 1i8));
-        assert!(can_upcast_to_decimal(&Int16, 9u8, 2i8));
-
-        assert!(can_upcast_to_decimal(&Int32, 10u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int32, 11u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int32, 11u8, 1i8));
-        assert!(can_upcast_to_decimal(&Int32, 14u8, 2i8));
-
-        assert!(can_upcast_to_decimal(&Int64, 20u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int64, 21u8, 0i8));
-        assert!(can_upcast_to_decimal(&Int64, 21u8, 1i8));
-        assert!(can_upcast_to_decimal(&Int64, 24u8, 2i8));
-    }
-
-    #[test]
-    fn rejects_unsafe_decimal_casts() {
-        use super::can_upcast_to_decimal;
-        use ArrowDataType::*;
-
-        assert!(!can_upcast_to_decimal(&Decimal128(2, 0), 2u8, 1i8));
-        assert!(!can_upcast_to_decimal(&Decimal128(2, 0), 2u8, -1i8));
-        assert!(!can_upcast_to_decimal(&Decimal128(5, 2), 6u8, 4i8));
-
-        assert!(!can_upcast_to_decimal(&Int8, 2u8, 0i8));
-        assert!(!can_upcast_to_decimal(&Int8, 3u8, 1i8));
-        assert!(!can_upcast_to_decimal(&Int16, 4u8, 0i8));
-        assert!(!can_upcast_to_decimal(&Int16, 5u8, 1i8));
-        assert!(!can_upcast_to_decimal(&Int32, 9u8, 0i8));
-        assert!(!can_upcast_to_decimal(&Int32, 10u8, 1i8));
-        assert!(!can_upcast_to_decimal(&Int64, 19u8, 0i8));
-        assert!(!can_upcast_to_decimal(&Int64, 20u8, 1i8));
-    }
-
     #[test]
     fn test_write_json() -> DeltaResult<()> {
         let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
@@ -1494,7 +1437,7 @@ mod tests {
             vec![Arc::new(StringArray::from(vec!["string1", "string2"]))],
         )?;
         let data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(data));
-        let json = write_json(Box::new(std::iter::once(data)))?;
+        let json = to_json_bytes(Box::new(std::iter::once(data)))?;
         assert_eq!(
             json,
             "{\"string\":\"string1\"}\n{\"string\":\"string2\"}\n".as_bytes()

kernel/src/engine/default/json.rs

@@ -16,7 +16,7 @@ use url::Url;
 use super::executor::TaskExecutor;
 use super::file_stream::{FileOpenFuture, FileOpener, FileStream};
 use crate::engine::arrow_utils::parse_json as arrow_parse_json;
-use crate::engine::arrow_utils::write_json;
+use crate::engine::arrow_utils::to_json_bytes;
 use crate::schema::SchemaRef;
 use crate::{
     DeltaResult, EngineData, Error, ExpressionRef, FileDataReadResultIterator, FileMeta,
@@ -99,21 +99,19 @@ impl<E: TaskExecutor> JsonHandler for DefaultJsonHandler<E> {
         data: Box<dyn Iterator<Item = Box<dyn EngineData>> + Send>,
         _overwrite: bool,
     ) -> DeltaResult<()> {
-        let buffer = write_json(data)?;
+        let buffer = to_json_bytes(data)?;
         // Put if absent
         let store = self.store.clone(); // cheap Arc
         let path = Path::from(path.path());
-        let path2 = path.clone(); // FIXME gross
+        let path_str = path.to_string();
         self.task_executor
             .block_on(async move {
                 store
                     .put_opts(&path, buffer.into(), object_store::PutMode::Create.into())
                     .await
             })
             .map_err(|e| match e {
-                object_store::Error::AlreadyExists { .. } => {
-                    Error::FileAlreadyExists(path2.to_string())
-                }
+                object_store::Error::AlreadyExists { .. } => Error::FileAlreadyExists(path_str),
                 e => e.into(),
             })?;
         Ok(())

kernel/src/engine/sync/json.rs

@@ -7,10 +7,11 @@ use url::Url;
 use super::read_files;
 use crate::engine::arrow_data::ArrowEngineData;
 use crate::engine::arrow_utils::parse_json as arrow_parse_json;
-use crate::engine::arrow_utils::write_json;
+use crate::engine::arrow_utils::to_json_bytes;
 use crate::schema::SchemaRef;
 use crate::{
-    DeltaResult, EngineData, Error, Expression, FileDataReadResultIterator, FileMeta, JsonHandler,
+    DeltaResult, EngineData, Error, ExpressionRef, FileDataReadResultIterator, FileMeta,
+    JsonHandler,
 };
 
 pub(crate) struct SyncJsonHandler;
@@ -66,7 +67,7 @@ impl JsonHandler for SyncJsonHandler {
 
         // write data to tmp file
         let mut tmp_file = NamedTempFile::new_in(parent)?;
-        let buf = write_json(data)?;
+        let buf = to_json_bytes(data)?;
         tmp_file.write_all(&buf)?;
         tmp_file.flush()?;
 

kernel/src/error.rs

@@ -164,6 +164,10 @@ pub enum Error {
     #[error("Invalid commit info: {0}")]
     InvalidCommitInfo(String),
 
+    /// Commit info was not passed to the transaction
+    #[error("Missing commit info")]
+    MissingCommitInfo,
+
     /// The file already exists at the path, prohibiting a non-overwrite write
     #[error("File already exists: {0}")]
     FileAlreadyExists(String),

kernel/src/lib.rs

@@ -208,13 +208,13 @@ pub trait JsonHandler: Send + Sync {
         predicate: Option<ExpressionRef>,
     ) -> DeltaResult<FileDataReadResultIterator>;
 
-    /// Atomically (!) write a single JSON file. Each row of the input data represents an action
-    /// in the delta log. this PUT must:
+    /// Atomically (!) write a single JSON file. Each row of the input data should be written as a
+    /// new JSON object appended to the file. this write must:
     /// (1) serialize the data to newline-delimited json (each row is a json object literal)
-    /// (2) write the data to the object store atomically (i.e. if the file already exists, fail
-    ///     unless the overwrite flag is set)
+    /// (2) write the data to storage atomically (i.e. if the file already exists, fail unless the
+    ///     overwrite flag is set)
     ///
-    /// The JSON data should be written as { "column1": "value1", "column2": "value2", ... }
+    /// For example, the JSON data should be written as { "column1": "val1", "column2": "val2", .. }
     /// with each row on a new line.
     ///
     /// NOTE: Null columns should not be written to the JSON file. For example, if a row has columns
@@ -224,15 +224,12 @@ pub trait JsonHandler: Send + Sync {
     ///
     /// # Parameters
     ///
-    /// - `path` - URL to write the JSON file to
-    /// - `data` - Iterator of EngineData to write to the JSON file. each row should be written as
+    /// - `path` - URL specifying the location to write the JSON file
+    /// - `data` - Iterator of EngineData to write to the JSON file. Each row should be written as
     ///   a new JSON object appended to the file. (that is, the file is newline-delimeted JSON, and
     ///   each row is a JSON object on a single line)
     /// - `overwrite` - If true, overwrite the file if it exists. If false, the call must fail if
     ///   the file exists.
-    ///
-    /// NOTE: the `overwrite` flag isn't used for the existing commit flow, but will be used in the
-    /// future to write `_last_checkpoint` files.
     fn write_json_file(
         &self,
         path: &Url,