[write] add Transaction with commit info and commit implementation

ffi/src/lib.rs

@@ -329,6 +329,8 @@ pub enum KernelError {
     InternalError,
     InvalidExpression,
     InvalidLogPath,
+    InvalidCommitInfo,
+    FileAlreadyExists,
 }
 
 impl From<Error> for KernelError {
@@ -376,6 +378,8 @@ impl From<Error> for KernelError {
             } => Self::from(*source),
             Error::InvalidExpressionEvaluation(_) => KernelError::InvalidExpression,
             Error::InvalidLogPath(_) => KernelError::InvalidLogPath,
+            Error::InvalidCommitInfo(_) => KernelError::InvalidCommitInfo,
+            Error::FileAlreadyExists(_) => KernelError::FileAlreadyExists,
         }
     }
 }

kernel/Cargo.toml

@@ -37,6 +37,8 @@ delta_kernel_derive = { path = "../derive-macros", version = "0.3.1" }
 # used for developer-visibility
 visibility = "0.1.1"
 
+# Used in the sync engine
+tempfile = { version = "3", optional = true }
 # Used in default engine
 arrow-buffer = { workspace = true, optional = true }
 arrow-array = { workspace = true, optional = true, features = ["chrono-tz"] }
@@ -100,6 +102,7 @@ sync-engine = [
   "arrow-json",
   "arrow-select",
   "parquet",
+  "tempfile",
 ]
 integration-test = [
   "hdfs-native-object-store/integration-test",

kernel/src/actions/mod.rs

@@ -417,4 +417,22 @@ mod tests {
         )]));
         assert_eq!(schema, expected);
     }
+
+    #[test]
+    fn test_commit_info_schema() {
+        let schema = get_log_schema()
+            .project(&["commitInfo"])
+            .expect("Couldn't get commitInfo field");
+
+        let expected = Arc::new(StructType::new(vec![StructField::new(
+            "commitInfo",
+            StructType::new(vec![StructField::new(
+                "kernelVersion",
+                DataType::STRING,
+                true,
+            )]),
+            true,
+        )]));
+        assert_eq!(schema, expected);
+    }
 }

kernel/src/engine/arrow_utils.rs

@@ -15,7 +15,7 @@ use arrow_array::{
     cast::AsArray, new_null_array, Array as ArrowArray, GenericListArray, OffsetSizeTrait,
     RecordBatch, StringArray, StructArray,
 };
-use arrow_json::ReaderBuilder;
+use arrow_json::{LineDelimitedWriter, ReaderBuilder};
 use arrow_schema::{
     DataType as ArrowDataType, Field as ArrowField, FieldRef as ArrowFieldRef, Fields,
     SchemaRef as ArrowSchemaRef,
@@ -813,6 +813,22 @@ fn parse_json_impl(json_strings: &StringArray, schema: ArrowSchemaRef) -> DeltaR
     Ok(concat_batches(&schema, output.iter())?)
 }
 
+/// 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>,
+) -> DeltaResult<Vec<u8>> {
+    let mut writer = LineDelimitedWriter::new(Vec::new());
+    for chunk in data.into_iter() {
+        let arrow_data = ArrowEngineData::try_from_engine_data(chunk)?;
+        let record_batch = arrow_data.record_batch();
+        writer.write(record_batch)?;
+    }
+    writer.finish()?;
+    Ok(writer.into_inner())
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -1621,4 +1637,24 @@ mod tests {
         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(
+            "string",
+            ArrowDataType::Utf8,
+            true,
+        )]));
+        let data = RecordBatch::try_new(
+            schema.clone(),
+            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)))?;
+        assert_eq!(
+            json,
+            "{\"string\":\"string1\"}\n{\"string\":\"string2\"}\n".as_bytes()
+        );
+        Ok(())
+    }
 }

kernel/src/engine/default/json.rs

@@ -11,10 +11,12 @@ use bytes::{Buf, Bytes};
 use futures::{StreamExt, TryStreamExt};
 use object_store::path::Path;
 use object_store::{DynObjectStore, GetResultPayload};
+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::schema::SchemaRef;
 use crate::{
     DeltaResult, EngineData, Error, Expression, FileDataReadResultIterator, FileMeta, JsonHandler,
@@ -88,6 +90,33 @@ impl<E: TaskExecutor> JsonHandler for DefaultJsonHandler<E> {
             self.readahead,
         )
     }
+
+    // note: for now we just buffer all the data and write it out all at once
+    fn write_json_file(
+        &self,
+        path: &Url,
+        data: Box<dyn Iterator<Item = Box<dyn EngineData>> + Send>,
+        _overwrite: bool,
+    ) -> DeltaResult<()> {
+        let buffer = write_json(data)?;
+        // Put if absent
+        let store = self.store.clone(); // cheap Arc
+        let path = Path::from(path.path());
+        let path2 = path.clone(); // FIXME gross
+        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 { .. } => {
+                    crate::error::Error::FileAlreadyExists(path2.to_string())
+                }
+                e => e.into(),
+            })?;
+        Ok(())
+    }
 }
 
 /// A [`FileOpener`] that opens a JSON file and yields a [`FileOpenFuture`]

kernel/src/engine/sync/json.rs

@@ -1,13 +1,17 @@
-use std::{fs::File, io::BufReader};
+use std::{fs::File, io::BufReader, io::Write};
 
-use crate::{
-    engine::arrow_utils::parse_json as arrow_parse_json, schema::SchemaRef, DeltaResult,
-    EngineData, Expression, FileDataReadResultIterator, FileMeta, JsonHandler,
-};
 use arrow_schema::SchemaRef as ArrowSchemaRef;
+use tempfile::NamedTempFile;
+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::schema::SchemaRef;
+use crate::{
+    DeltaResult, EngineData, Expression, FileDataReadResultIterator, FileMeta, JsonHandler,
+};
 
 pub(crate) struct SyncJsonHandler;
 
@@ -40,4 +44,98 @@ impl JsonHandler for SyncJsonHandler {
     ) -> DeltaResult<Box<dyn EngineData>> {
         arrow_parse_json(json_strings, output_schema)
     }
+
+    // For sync writer we write data to a tmp file then atomically rename it to the final path.
+    // This is highly OS-dependent and for now relies on the atomicity of tempfile's
+    // `persist_noclobber`.
+    fn write_json_file(
+        &self,
+        path: &Url,
+        data: Box<dyn Iterator<Item = Box<dyn EngineData>> + Send>,
+        _overwrite: bool,
+    ) -> DeltaResult<()> {
+        let path = path
+            .to_file_path()
+            .map_err(|_| crate::Error::generic("sync client can only read local files"))?;
+        let Some(parent) = path.parent() else {
+            return Err(crate::Error::generic(format!(
+                "no parent found for {:?}",
+                path
+            )));
+        };
+
+        // write data to tmp file
+        let mut tmp_file = NamedTempFile::new_in(parent)?;
+        let buf = write_json(data)?;
+        tmp_file.write_all(&buf)?;
+        tmp_file.flush()?;
+
+        // use 'persist_noclobber' to atomically rename tmp file to final path
+        tmp_file
+            .persist_noclobber(path.clone())
+            .map_err(|e| match e {
+                tempfile::PersistError { error, .. }
+                    if error.kind() == std::io::ErrorKind::AlreadyExists =>
+                {
+                    crate::Error::FileAlreadyExists(path.to_string_lossy().to_string())
+                }
+                e => crate::Error::IOError(e.into()),
+            })?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use std::sync::Arc;
+
+    use arrow_array::{RecordBatch, StringArray};
+    use arrow_schema::DataType as ArrowDataType;
+    use arrow_schema::Field;
+    use arrow_schema::Schema as ArrowSchema;
+    use serde_json::json;
+    use url::Url;
+
+    #[test]
+    fn test_write_json_file() -> DeltaResult<()> {
+        let test_dir = tempfile::tempdir().unwrap();
+        let path = test_dir.path().join("00000000000000000001.json");
+        let handler = SyncJsonHandler;
+
+        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "dog",
+            ArrowDataType::Utf8,
+            true,
+        )]));
+        let data = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(StringArray::from(vec!["remi", "wilson"]))],
+        )?;
+        let data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(data));
+        let empty: Box<dyn EngineData> =
+            Box::new(ArrowEngineData::new(RecordBatch::new_empty(schema)));
+
+        let url = Url::from_file_path(path.clone()).unwrap();
+        handler
+            .write_json_file(&url, Box::new(std::iter::once(data)), false)
+            .expect("write json file");
+        assert!(matches!(
+            handler.write_json_file(&url, Box::new(std::iter::once(empty)), false),
+            Err(crate::Error::FileAlreadyExists(_))
+        ));
+
+        let file = std::fs::read_to_string(path)?;
+        let json: Vec<_> = serde_json::Deserializer::from_str(&file)
+            .into_iter::<serde_json::Value>()
+            .flatten()
+            .collect();
+        assert_eq!(
+            json,
+            vec![json!({"dog": "remi"}), json!({"dog": "wilson"}),]
+        );
+
+        Ok(())
+    }
 }

kernel/src/error.rs

@@ -159,6 +159,14 @@ pub enum Error {
     /// Unable to parse the name of a log path
     #[error("Invalid log path: {0}")]
     InvalidLogPath(String),
+
+    /// Invalid commit info passed to the transaction
+    #[error("Invalid commit info: {0}")]
+    InvalidCommitInfo(String),
+
+    /// The file already exists at the path, prohibiting a non-overwrite write
+    #[error("File already exists: {0}")]
+    FileAlreadyExists(String),
 }
 
 // Convenience constructors for Error types that take a String argument

kernel/src/lib.rs

@@ -73,6 +73,7 @@ pub mod scan;
 pub mod schema;
 pub mod snapshot;
 pub mod table;
+pub mod transaction;
 pub(crate) mod utils;
 
 pub use engine_data::{DataVisitor, EngineData};
@@ -196,6 +197,38 @@ pub trait JsonHandler: Send + Sync {
         // clone the (potentially large) expression every time we call this function.
         predicate: Option<Expression>,
     ) -> DeltaResult<FileDataReadResultIterator>;
+
+    /// Atomically (!) write a single JSON file. Each row of the input data represents an action
+    /// in the delta log. this PUT 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)
+    ///
+    /// The JSON data should be written as { "column1": "value1", "column2": "value2", ... }
+    /// 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
+    /// ["a", "b"] and the value of "b" is null, the JSON object should be written as
+    /// { "a": "..." }. Note that including nulls is technically valid JSON, but would bloat the
+    /// log, therefore we recommend omitting them.
+    ///
+    /// # 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
+    ///   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,
+        data: Box<dyn Iterator<Item = Box<dyn EngineData>> + Send>,
+        overwrite: bool,
+    ) -> DeltaResult<()>;
 }
 
 /// Provides Parquet file related functionalities to Delta Kernel.