feat: improve test fixtures to generate actions and data

crates/core/src/kernel/models/actions.rs

@@ -1,21 +1,18 @@
 use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::str::FromStr;
-// use std::io::{Cursor, Read};
-// use std::sync::Arc;
 
-// use roaring::RoaringTreemap;
-use crate::DeltaConfigKey;
 use maplit::hashset;
 use serde::{Deserialize, Serialize};
 use tracing::warn;
 use url::Url;
 
 use super::schema::StructType;
 use crate::kernel::{error::Error, DeltaResult};
+use crate::DeltaConfigKey;
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 /// Defines a file format used in table
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
 pub struct Format {
  /// Name of the encoding for files in this table
  pub provider: String,
@@ -1126,15 +1123,9 @@ pub(crate) mod serde_path {
 #[cfg(test)]
 mod tests {
  use std::path::PathBuf;
- // use std::sync::Arc;
-
- // use object_store::local::LocalFileSystem;
-
- use crate::kernel::PrimitiveType;
 
  use super::*;
- // use crate::client::filesystem::ObjectStoreFileSystemClient;
- // use crate::executor::tokio::TokioBackgroundExecutor;
+ use crate::kernel::PrimitiveType;
 
  fn dv_relateive() -> DeletionVectorDescriptor {
  DeletionVectorDescriptor {

crates/core/src/kernel/scalars.rs

@@ -8,6 +8,8 @@ use delta_kernel::{
  schema::StructField,
 };
 use object_store::path::Path;
+#[cfg(test)]
+use serde_json::Value;
 use std::cmp::Ordering;
 use urlencoding::encode;
 
@@ -21,6 +23,9 @@ pub trait ScalarExt: Sized {
  fn serialize_encoded(&self) -> String;
  /// Create a [`Scalar`] from an arrow array row
  fn from_array(arr: &dyn Array, index: usize) -> Option<Self>;
+ /// Serialize as serde_json::Value
+ #[cfg(test)]
+ fn to_json(&self) -> serde_json::Value;
 }
 
 impl ScalarExt for Scalar {
@@ -218,6 +223,54 @@ impl ScalarExt for Scalar {
  | Null => None,
  }
  }
+
+ /// Serializes this scalar as a serde_json::Value.
+ #[cfg(test)]
+ fn to_json(&self) -> serde_json::Value {
+ match self {
+ Self::String(s) => Value::String(s.to_owned()),
+ Self::Byte(b) => Value::Number(serde_json::Number::from(*b)),
+ Self::Short(s) => Value::Number(serde_json::Number::from(*s)),
+ Self::Integer(i) => Value::Number(serde_json::Number::from(*i)),
+ Self::Long(l) => Value::Number(serde_json::Number::from(*l)),
+ Self::Float(f) => Value::Number(serde_json::Number::from_f64(*f as f64).unwrap()),
+ Self::Double(d) => Value::Number(serde_json::Number::from_f64(*d).unwrap()),
+ Self::Boolean(b) => Value::Bool(*b),
+ Self::TimestampNtz(ts) | Self::Timestamp(ts) => {
+ let ts = Utc.timestamp_micros(*ts).single().unwrap();
+ Value::String(ts.format("%Y-%m-%d %H:%M:%S%.6f").to_string())
+ }
+ Self::Date(days) => {
+ let date = DateTime::from_timestamp(*days as i64 * 24 * 3600, 0).unwrap();
+ Value::String(date.format("%Y-%m-%d").to_string())
+ }
+ Self::Decimal(value, _, scale) => match scale.cmp(&0) {
+ Ordering::Equal => Value::String(value.to_string()),
+ Ordering::Greater => {
+ let scalar_multiple = 10_i128.pow(*scale as u32);
+ let mut s = String::new();
+ s.push_str((value / scalar_multiple).to_string().as_str());
+ s.push('.');
+ s.push_str(&format!(
+ "{:0>scale$}",
+ value % scalar_multiple,
+ scale = *scale as usize
+ ));
+ Value::String(s)
+ }
+ Ordering::Less => {
+ let mut s = value.to_string();
+ for _ in 0..*scale {
+ s.push('0');
+ }
+ Value::String(s)
+ }
+ },
+ Self::Binary(val) => Value::String(create_escaped_binary_string(val.as_slice())),
+ Self::Null(_) => Value::Null,
+ Self::Struct(_) => unimplemented!(),
+ }
+ }
 }
 
 fn create_escaped_binary_string(data: &[u8]) -> String {

crates/core/src/kernel/snapshot/mod.rs

@@ -777,6 +777,7 @@ mod tests {
  use super::*;
  use crate::kernel::Remove;
  use crate::protocol::{DeltaOperation, SaveMode};
+ use crate::test_utils::ActionFactory;
 
  #[tokio::test]
  async fn test_snapshots() -> TestResult {
@@ -991,31 +992,17 @@ mod tests {
  ]);
 
  let partition_columns = vec!["date".to_string()];
- let metadata = Metadata {
- id: "id".to_string(),
- name: None,
- description: None,
- format: Default::default(),
- schema_string: serde_json::to_string(&schema).unwrap(),
- partition_columns,
- configuration: Default::default(),
- created_time: None,
- };
+ let metadata = ActionFactory::metadata(&schema, Some(&partition_columns), None);
+ let protocol = ActionFactory::protocol(None, None, None::<Vec<_>>, None::<Vec<_>>);
 
- let protocol = Protocol {
- min_reader_version: 1,
- min_writer_version: 2,
- reader_features: Default::default(),
- writer_features: Default::default(),
- };
  let commit_data = CommitData::new(
  vec![
  Action::Protocol(protocol.clone()),
  Action::Metadata(metadata.clone()),
  ],
  DeltaOperation::Write {
  mode: SaveMode::Append,
- partition_by: None,
+ partition_by: Some(partition_columns),
  predicate: None,
  },
  HashMap::new(),
@@ -1025,27 +1012,17 @@ mod tests {
 
  let snapshot = Snapshot {
  log_segment: log_segment.clone(),
- config: Default::default(),
- protocol: Default::default(),
+ protocol: protocol.clone(),
  metadata,
  schema: schema.clone(),
  table_url: "table".to_string(),
+ config: Default::default(),
  };
 
  let expected = StructType::new(vec![StructField::new("date", DataType::DATE, true)]);
  assert_eq!(snapshot.partitions_schema(None).unwrap(), Some(expected));
 
- let metadata = Metadata {
- id: "id".to_string(),
- name: None,
- description: None,
- format: Default::default(),
- schema_string: serde_json::to_string(&schema).unwrap(),
- partition_columns: vec![],
- configuration: Default::default(),
- created_time: None,
- };
-
+ let metadata = ActionFactory::metadata(&schema, None::<Vec<&str>>, None);
  let commit_data = CommitData::new(
  vec![
  Action::Protocol(protocol.clone()),

crates/core/src/lib.rs

@@ -79,6 +79,9 @@ pub mod schema;
 pub mod storage;
 pub mod table;
 
+#[cfg(test)]
+pub mod test_utils;
+
 #[cfg(feature = "datafusion")]
 pub mod delta_datafusion;
 pub mod writer;