fix: Change InCommitTimestamp enablement getter function

kernel/src/table_configuration.rs

@@ -23,6 +23,19 @@ use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Error, Version};
 use delta_kernel_derive::internal_api;
 
+/// Information about in-commit timestamp enablement state.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum InCommitTimestampEnablement {
+    /// In-commit timestamps is not enabled
+    NotEnabled,
+    /// In-commit timestamps is enabled
+    Enabled {
+        /// Enablement information, if available. `None` indicates the table was created
+        /// with ICT enabled from the beginning (no enablement properties needed).
+        enablement: Option<(Version, i64)>,
+    },
+}
+
 /// Holds all the configuration for a table at a specific version. This includes the supported
 /// reader and writer features, table properties, schema, version, and table root. This can be used
 /// to check whether a table supports a feature or has it enabled. For example, deletion vector
@@ -317,36 +330,38 @@ impl TableConfiguration {
                 .unwrap_or(false)
     }
 
-    /// If in-commit timestamps is enabled, returns a tuple of the in-commit timestamp enablement
-    /// version and timestamp.
+    /// Returns information about in-commit timestamp enablement state.
     ///
-    /// If in-commit timestamps is not supported, or not enabled, this returns `None`.
-    /// If in-commit timestams is enabled, but the enablement version or timestamp is not present,
-    /// this returns an error.
+    /// Returns an error if only one of the enablement properties is present, as this indicates
+    /// an inconsistent state.
     #[allow(unused)]
-    pub(crate) fn in_commit_timestamp_enablement(&self) -> DeltaResult<Option<(Version, i64)>> {
+    pub(crate) fn in_commit_timestamp_enablement(
+        &self,
+    ) -> DeltaResult<InCommitTimestampEnablement> {
         if !self.is_in_commit_timestamps_enabled() {
-            return Ok(None);
+            return Ok(InCommitTimestampEnablement::NotEnabled);
         }
+
         let enablement_version = self
             .table_properties()
             .in_commit_timestamp_enablement_version;
         let enablement_timestamp = self
             .table_properties()
             .in_commit_timestamp_enablement_timestamp;
 
-        let ict_error = |err: &str| {
-            Error::generic(format!(
-                "In-commit timestamp enabled, but missing Enablement version or timestamp. {err}"
-            ))
-        };
         match (enablement_version, enablement_timestamp) {
-            (Some(version), Some(timestamp)) => Ok(Some((version, timestamp))),
-            (Some(_), None) => Err(ict_error("Enablement timestamp is not present")),
-            (None, Some(_)) => Err(ict_error("Enablement version is not present")),
-            (None, None) => Err(ict_error(
-                "Enablement version and timestamp are not present.",
+            (Some(version), Some(timestamp)) => Ok(InCommitTimestampEnablement::Enabled {
+                enablement: Some((version, timestamp))
+            }),
+            (Some(_), None) => Err(Error::generic(
+                "In-commit timestamp enabled, but enablement timestamp is missing while enablement version is present"
+            )),
+            (None, Some(_)) => Err(Error::generic(
+                "In-commit timestamp enabled, but enablement version is missing while enablement timestamp is present"
             )),
+            (None, None) => Ok(InCommitTimestampEnablement::Enabled {
+                enablement: None
+            }),
         }
     }
 
@@ -423,7 +438,7 @@ mod test {
     use crate::utils::test_utils::assert_result_error_with_message;
     use crate::Error;
 
-    use super::TableConfiguration;
+    use super::{InCommitTimestampEnablement, TableConfiguration};
 
     #[test]
     fn dv_supported_not_enabled() {
@@ -480,6 +495,40 @@ mod test {
         assert!(table_config.is_deletion_vector_enabled());
     }
     #[test]
+    fn ict_enabled_from_table_creation() {
+        let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
+        let metadata = Metadata::try_new(
+            None,
+            None,
+            schema,
+            vec![],
+            0, // Table creation version
+            HashMap::from_iter([(
+                "delta.enableInCommitTimestamps".to_string(),
+                "true".to_string(),
+            )]),
+        )
+        .unwrap();
+        let protocol = Protocol::try_new(
+            3,
+            7,
+            Some::<Vec<String>>(vec![]),
+            Some([WriterFeature::InCommitTimestamp]),
+        )
+        .unwrap();
+        let table_root = Url::try_from("file:///").unwrap();
+        let table_config = TableConfiguration::try_new(metadata, protocol, table_root, 0).unwrap();
+        assert!(table_config.is_in_commit_timestamps_supported());
+        assert!(table_config.is_in_commit_timestamps_enabled());
+        // When ICT is enabled from table creation (version 0), it's perfectly normal
+        // for enablement properties to be missing
+        let info = table_config.in_commit_timestamp_enablement().unwrap();
+        assert_eq!(
+            info,
+            InCommitTimestampEnablement::Enabled { enablement: None }
+        );
+    }
+    #[test]
     fn ict_supported_and_enabled() {
         let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
         let metadata = Metadata::try_new(
@@ -515,22 +564,34 @@ mod test {
         let table_config = TableConfiguration::try_new(metadata, protocol, table_root, 0).unwrap();
         assert!(table_config.is_in_commit_timestamps_supported());
         assert!(table_config.is_in_commit_timestamps_enabled());
-        let enablement = table_config.in_commit_timestamp_enablement().unwrap();
-        assert_eq!(enablement, Some((5, 100)))
+        let info = table_config.in_commit_timestamp_enablement().unwrap();
+        assert_eq!(
+            info,
+            InCommitTimestampEnablement::Enabled {
+                enablement: Some((5, 100))
+            }
+        )
     }
     #[test]
-    fn ict_supported_and_enabled_without_enablement_info() {
+    fn ict_enabled_with_partial_enablement_info() {
         let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
         let metadata = Metadata::try_new(
             None,
             None,
             schema,
             vec![],
             0,
-            HashMap::from_iter([(
-                "delta.enableInCommitTimestamps".to_string(),
-                "true".to_string(),
-            )]),
+            HashMap::from_iter([
+                (
+                    "delta.enableInCommitTimestamps".to_string(),
+                    "true".to_string(),
+                ),
+                (
+                    "delta.inCommitTimestampEnablementVersion".to_string(),
+                    "5".to_string(),
+                ),
+                // Missing enablement timestamp
+            ]),
         )
         .unwrap();
         let protocol = Protocol::try_new(
@@ -545,8 +606,9 @@ mod test {
         assert!(table_config.is_in_commit_timestamps_supported());
         assert!(table_config.is_in_commit_timestamps_enabled());
         assert!(matches!(
-                table_config.in_commit_timestamp_enablement(),
-                Err(Error::Generic(msg)) if msg.contains("Enablement version and timestamp are not present.")));
+            table_config.in_commit_timestamp_enablement(),
+            Err(Error::Generic(msg)) if msg.contains("enablement timestamp is missing while enablement version is present")
+        ));
     }
     #[test]
     fn ict_supported_and_not_enabled() {
@@ -563,6 +625,8 @@ mod test {
         let table_config = TableConfiguration::try_new(metadata, protocol, table_root, 0).unwrap();
         assert!(table_config.is_in_commit_timestamps_supported());
         assert!(!table_config.is_in_commit_timestamps_enabled());
+        let info = table_config.in_commit_timestamp_enablement().unwrap();
+        assert_eq!(info, InCommitTimestampEnablement::NotEnabled);
     }
     #[test]
     fn fails_on_unsupported_feature() {