From f1f8a82d40a763743d8c8bf1f32ecd9505c36ae4 Mon Sep 17 00:00:00 2001
From: Zach Schuermann <zachary.zvs@gmail.com>
Date: Mon, 25 Nov 2024 10:14:21 -0800
Subject: [PATCH] use TableProperties for CDF check (#536)

## What changes are proposed in this pull request?
use `TableProperties` for CDF enablement check in `TableChanges`. some
minor nits/cleanup in table properties


## How was this change tested?
existing UT
---
 kernel/src/table_changes/mod.rs            |  8 +++-----
 kernel/src/table_properties/deserialize.rs | 15 +--------------
 2 files changed, 4 insertions(+), 19 deletions(-)

diff --git a/kernel/src/table_changes/mod.rs b/kernel/src/table_changes/mod.rs
index 7f4b8963a..2005762f9 100644
--- a/kernel/src/table_changes/mod.rs
+++ b/kernel/src/table_changes/mod.rs
@@ -95,12 +95,10 @@ impl TableChanges {
         // Verify CDF is enabled at the beginning and end of the interval to fail early. We must
         // still check that CDF is enabled for every metadata action in the CDF range.
         let is_cdf_enabled = |snapshot: &Snapshot| {
-            static ENABLE_CDF_FLAG: &str = "delta.enableChangeDataFeed";
             snapshot
-                .metadata()
-                .configuration
-                .get(ENABLE_CDF_FLAG)
-                .is_some_and(|val| val == "true")
+                .table_properties()
+                .enable_change_data_feed
+                .unwrap_or(false)
         };
         if !is_cdf_enabled(&start_snapshot) {
             return Err(Error::change_data_feed_unsupported(start_version));
diff --git a/kernel/src/table_properties/deserialize.rs b/kernel/src/table_properties/deserialize.rs
index e22befdb2..c9da1495b 100644
--- a/kernel/src/table_properties/deserialize.rs
+++ b/kernel/src/table_properties/deserialize.rs
@@ -103,7 +103,7 @@ pub(crate) fn parse_bool(s: &str) -> Option<bool> {
 /// `Some` if successfully parses, and `None` otherwise.
 pub(crate) fn parse_column_names(s: &str) -> Option<Vec<ColumnName>> {
     ColumnName::parse_column_name_list(s)
-        .map_err(|e| warn!("column name list failed to parse: {e}"))
+        .inspect_err(|e| warn!("column name list failed to parse: {e}"))
         .ok()
 }
 
@@ -190,19 +190,6 @@ fn parse_interval_impl(value: &str) -> Result<Duration, ParseIntervalError> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::column_name;
-
-    #[test]
-    fn test_parse_column_names() {
-        assert_eq!(
-            parse_column_names("`col 1`, col.a2,col3").unwrap(),
-            vec![
-                ColumnName::new(["col 1"]),
-                column_name!("col.a2"),
-                column_name!("col3")
-            ]
-        );
-    }
 
     #[test]
     fn test_parse_bool() {