feat!: add in-commit timestamps table properties (#558)

Parse: `delta.enableInCommitTimestamps`,
`delta.inCommitTimestampEnablementVersion`, and
`delta.inCommitTimestampEnablementTimestamp` and expose table property
methods to access.

### This PR affects the following public APIs

New pub fields in `TableProperties`: 
-  `enable_in_commit_timestamps`
- `in_commit_timestamp_enablement_version`
- `in_commit_timestamp_enablement_timestamp`

Author: zachschuermann

ffi/src/lib.rs

@@ -462,7 +462,8 @@ pub unsafe extern "C" fn set_builder_option(
 }
 
 /// Consume the builder and return a `default` engine. After calling, the passed pointer is _no
-/// longer valid_.
+/// longer valid_. Note that this _consumes_ and frees the builder, so there is no need to
+/// drop/free it afterwards.
 ///
 ///
 /// # Safety

kernel/src/table_properties.rs

@@ -16,7 +16,7 @@ use std::time::Duration;
 
 use crate::expressions::ColumnName;
 use crate::table_features::ColumnMappingMode;
-use crate::Error;
+use crate::{Error, Version};
 
 use strum::EnumString;
 
@@ -137,6 +137,20 @@ pub struct TableProperties {
     /// whether to enable row tracking during writes.
     pub enable_row_tracking: Option<bool>,
 
+    /// Whether to enable [In-Commit Timestamps]. The in-commit timestamps writer feature strongly
+    /// associates a monotonically increasing timestamp with each commit by storing it in the
+    /// commit's metadata.
+    ///
+    /// [In-Commit Timestamps]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#in-commit-timestamps
+    pub enable_in_commit_timestamps: Option<bool>,
+
+    /// The version of the table at which in-commit timestamps were enabled.
+    pub in_commit_timestamp_enablement_version: Option<Version>,
+
+    /// The timestamp of the table at which in-commit timestamps were enabled. This must be the same
+    /// as the inCommitTimestamp of the commit when this feature was enabled.
+    pub in_commit_timestamp_enablement_timestamp: Option<i64>,
+
     /// any unrecognized properties are passed through and ignored by the parser
     pub unknown_properties: HashMap<String, String>,
 }
@@ -268,6 +282,9 @@ mod tests {
             ("delta.tuneFileSizesForRewrites", "true"),
             ("delta.checkpointPolicy", "v2"),
             ("delta.enableRowTracking", "true"),
+            ("delta.enableInCommitTimestamps", "true"),
+            ("delta.inCommitTimestampEnablementVersion", "15"),
+            ("delta.inCommitTimestampEnablementTimestamp", "1612345678"),
         ];
         let actual = TableProperties::from(properties.into_iter());
         let expected = TableProperties {
@@ -293,6 +310,9 @@ mod tests {
             tune_file_sizes_for_rewrites: Some(true),
             checkpoint_policy: Some(CheckpointPolicy::V2),
             enable_row_tracking: Some(true),
+            enable_in_commit_timestamps: Some(true),
+            in_commit_timestamp_enablement_version: Some(15),
+            in_commit_timestamp_enablement_timestamp: Some(1_612_345_678),
             unknown_properties: HashMap::new(),
         };
         assert_eq!(actual, expected);

kernel/src/table_properties/deserialize.rs

@@ -37,6 +37,9 @@ where
 // attempt to parse a key-value pair into a `TableProperties` struct. Returns Some(()) if the key
 // was successfully parsed, and None otherwise.
 fn try_parse(props: &mut TableProperties, k: &str, v: &str) -> Option<()> {
+    // NOTE!! we do Some(parse(v)?) instead of just parse(v) because we want to return None if the
+    // parsing fails. If we simply call 'parse(v)', then we would (incorrectly) return Some(()) and
+    // just set the property to None.
     match k {
         "delta.appendOnly" => props.append_only = Some(parse_bool(v)?),
         "delta.autoOptimize.autoCompact" => props.auto_compact = Some(parse_bool(v)?),
@@ -76,17 +79,36 @@ fn try_parse(props: &mut TableProperties, k: &str, v: &str) -> Option<()> {
         }
         "delta.checkpointPolicy" => props.checkpoint_policy = CheckpointPolicy::try_from(v).ok(),
         "delta.enableRowTracking" => props.enable_row_tracking = Some(parse_bool(v)?),
+        "delta.enableInCommitTimestamps" => {
+            props.enable_in_commit_timestamps = Some(parse_bool(v)?)
+        }
+        "delta.inCommitTimestampEnablementVersion" => {
+            props.in_commit_timestamp_enablement_version = Some(parse_non_negative(v)?)
+        }
+        "delta.inCommitTimestampEnablementTimestamp" => {
+            props.in_commit_timestamp_enablement_timestamp = Some(parse_non_negative(v)?)
+        }
         _ => return None,
     }
     Some(())
 }
 
-/// Deserialize a string representing a positive integer into an `Option<u64>`. Returns `Some` if
-/// successfully parses, and `None` otherwise.
+/// Deserialize a string representing a positive (> 0) integer into an `Option<u64>`. Returns `Some`
+/// if successfully parses, and `None` otherwise.
 pub(crate) fn parse_positive_int(s: &str) -> Option<NonZero<u64>> {
-    // parse to i64 (then check n > 0) since java doesn't even allow u64
-    let n: i64 = s.parse().ok()?;
-    NonZero::new(n.try_into().ok()?)
+    // parse as non-negative and verify the result is non-zero
+    NonZero::new(parse_non_negative(s)?)
+}
+
+/// Deserialize a string representing a non-negative integer into an `Option<u64>`. Returns `Some` if
+/// successfully parses, and `None` otherwise.
+pub(crate) fn parse_non_negative<T>(s: &str) -> Option<T>
+where
+    i64: TryInto<T>,
+{
+    // parse to i64 since java doesn't even allow u64
+    let n: i64 = s.parse().ok().filter(|&i| i >= 0)?;
+    n.try_into().ok()
 }
 
 /// Deserialize a string representing a boolean into an `Option<bool>`. Returns `Some` if
@@ -199,10 +221,16 @@ mod tests {
     }
 
     #[test]
-    fn test_parse_positive_int() {
-        assert_eq!(parse_positive_int("123").unwrap().get(), 123);
+    fn test_parse_int() {
+        assert_eq!(parse_positive_int("12").unwrap().get(), 12);
         assert_eq!(parse_positive_int("0"), None);
-        assert_eq!(parse_positive_int("-123"), None);
+        assert_eq!(parse_positive_int("-12"), None);
+        assert_eq!(parse_non_negative::<u64>("12").unwrap(), 12);
+        assert_eq!(parse_non_negative::<u64>("0").unwrap(), 0);
+        assert_eq!(parse_non_negative::<u64>("-12"), None);
+        assert_eq!(parse_non_negative::<i64>("12").unwrap(), 12);
+        assert_eq!(parse_non_negative::<i64>("0").unwrap(), 0);
+        assert_eq!(parse_non_negative::<i64>("-12"), None);
     }
 
     #[test]