Repository: apache/hudi-rs (language: python)

Issue:
Implement `HudiConfigs` for parsing, validating, and managing configs



Discussion / hints:

Fix patch:
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index 5d04337..06e15a4 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -61,6 +61,7 @@ bytes = { workspace = true }
 chrono = { workspace = true, default-features = false, features = ["clock"] }
 hashbrown = "0.14.3"
 regex = { workspace = true }
+strum_macros = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
 url = { workspace = true }
 
diff --git a/crates/core/src/config/mod.rs b/crates/core/src/config/mod.rs
index 3322df3..2399f93 100644
--- a/crates/core/src/config/mod.rs
+++ b/crates/core/src/config/mod.rs
@@ -16,103 +16,144 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+use std::any::type_name;
 use std::collections::HashMap;
+use std::sync::Arc;
 
-use anyhow::{anyhow, Context, Result};
+use anyhow::Result;
 
-pub trait OptionsParser {
+pub mod read;
+pub mod table;
+
+pub trait ConfigParser: AsRef<str> {
     type Output;
 
-    fn parse_value(&self, options: &HashMap<String, String>) -> Result<Self::Output>;
+    fn default_value(&self) -> Option<Self::Output>;
 
-    fn parse_value_or_default(&self, options: &HashMap<String, String>) -> Self::Output;
-}
+    fn is_required(&self) -> bool {
+        false
+    }
+
+    fn validate(&self, configs: &HashMap<String, String>) -> Result<()> {
+        match self.parse_value(configs) {
+            Ok(_) => Ok(()),
+            Err(e) => {
+                if !self.is_required() && e.to_string().ends_with("not found") {
+                    // TODO: introduce error type to avoid checking "not found"
+                    Ok(())
+                } else {
+                    Err(e)
+                }
+            }
+        }
+    }
 
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-pub enum HudiConfig {
-    ReadInputPartitions,
+    fn parse_value(&self, configs: &HashMap<String, String>) -> Result<Self::Output>;
+
+    fn parse_value_or_default(&self, configs: &HashMap<String, String>) -> Self::Output {
+        self.parse_value(configs).unwrap_or_else(|_| {
+            self.default_value()
+                .unwrap_or_else(|| panic!("No default value for config '{}'", self.as_ref()))
+        })
+    }
 }
 
 #[derive(Debug)]
 pub enum HudiConfigValue {
+    Boolean(bool),
     Integer(isize),
+    UInteger(usize),
+    String(String),
+    List(Vec<String>),
 }
 
 impl HudiConfigValue {
-    pub fn cast<T: 'static + TryFrom<isize> + TryFrom<usize> + std::fmt::Debug>(&self) -> T {
-        match self {
-            HudiConfigValue::Integer(value) => T::try_from(*value).unwrap_or_else(|_| {
-                panic!("Failed to convert isize to {}", std::any::type_name::<T>())
-            }),
+    pub fn to<T: 'static + std::fmt::Debug + From<HudiConfigValue>>(self) -> T {
+        T::from(self)
+    }
+}
+
+impl From<HudiConfigValue> for bool {
+    fn from(value: HudiConfigValue) -> Self {
+        match value {
+            HudiConfigValue::Boolean(v) => v,
+            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
         }
     }
 }
 
-impl HudiConfig {
-    fn default_value(&self) -> Option<HudiConfigValue> {
-        match self {
-            Self::ReadInputPartitions => Some(HudiConfigValue::Integer(0)),
+impl From<HudiConfigValue> for isize {
+    fn from(value: HudiConfigValue) -> Self {
+        match value {
+            HudiConfigValue::Integer(v) => v,
+            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
         }
     }
 }
 
-impl AsRef<str> for HudiConfig {
-    fn as_ref(&self) -> &str {
-        match self {
-            Self::ReadInputPartitions => "hoodie.read.input.partitions",
+impl From<HudiConfigValue> for usize {
+    fn from(value: HudiConfigValue) -> Self {
+        match value {
+            HudiConfigValue::UInteger(v) => v,
+            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
         }
     }
 }
 
-impl OptionsParser for HudiConfig {
-    type Output = HudiConfigValue;
-
-    fn parse_value(&self, options: &HashMap<String, String>) -> Result<Self::Output> {
-        match self {
-            HudiConfig::ReadInputPartitions => options.get(self.as_ref()).map_or_else(
-                || Err(anyhow!("Config '{}' not found", self.as_ref())),
-                |v| {
-                    v.parse::<isize>()
-                        .map(HudiConfigValue::Integer)
-                        .with_context(|| {
-                            format!("Failed to parse '{}' for config '{}'", v, self.as_ref())
-                        })
-                },
-            ),
+impl From<HudiConfigValue> for String {
+    fn from(value: HudiConfigValue) -> Self {
+        match value {
+            HudiConfigValue::Boolean(v) => v.to_string(),
+            HudiConfigValue::Integer(v) => v.to_string(),
+            HudiConfigValue::UInteger(v) => v.to_string(),
+            HudiConfigValue::String(v) => v,
+            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
         }
     }
+}
 
-    fn parse_value_or_default(&self, options: &HashMap<String, String>) -> Self::Output {
-        self.parse_value(options).unwrap_or_else(|_| {
-            self.default_value()
-                .unwrap_or_else(|| panic!("No default value for config '{}'", self.as_ref()))
-        })
+impl From<HudiConfigValue> for Vec<String> {
+    fn from(value: HudiConfigValue) -> Self {
+        match value {
+            HudiConfigValue::List(v) => v,
+            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
+        }
     }
 }
 
-#[cfg(test)]
-mod tests {
-    use crate::config::HudiConfig::ReadInputPartitions;
-    use crate::config::OptionsParser;
-    use std::collections::HashMap;
-
-    #[test]
-    fn parse_invalid_config_value() {
-        let options =
-            HashMap::from([(ReadInputPartitions.as_ref().to_string(), "foo".to_string())]);
-        let value = ReadInputPartitions.parse_value(&options);
-        assert_eq!(
-            value.err().unwrap().to_string(),
-            format!(
-                "Failed to parse 'foo' for config '{}'",
-                ReadInputPartitions.as_ref()
-            )
-        );
-        assert_eq!(
-            ReadInputPartitions
-                .parse_value_or_default(&options)
-                .cast::<isize>(),
-            0
-        );
+#[derive(Clone, Debug)]
+pub struct HudiConfigs {
+    pub raw_configs: Arc<HashMap<String, String>>,
+}
+
+impl HudiConfigs {
+    pub fn new(raw_configs: HashMap<String, String>) -> Self {
+        Self {
+            raw_configs: Arc::new(raw_configs),
+        }
+    }
+
+    pub fn empty() -> Self {
+        Self {
+            raw_configs: Arc::new(HashMap::new()),
+        }
+    }
+
+    pub fn validate(&self, parser: impl ConfigParser<Output = HudiConfigValue>) -> Result<()> {
+        parser.validate(&self.raw_configs)
+    }
+
+    pub fn get(
+        &self,
+        parser: impl ConfigParser<Output = HudiConfigValue>,
+    ) -> Result<HudiConfigValue> {
+        parser.parse_value(&self.raw_configs)
+    }
+
+    pub fn get_or_default(
+        &self,
+        parser: impl ConfigParser<Output = HudiConfigValue>,
+    ) -> HudiConfigValue {
+        parser.parse_value_or_default(&self.raw_configs)
     }
 }
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
new file mode 100644
index 0000000..43879de
--- /dev/null
+++ b/crates/core/src/config/read.rs
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use anyhow::{anyhow, Result};
+
+use crate::config::{ConfigParser, HudiConfigValue};
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum HudiReadConfig {
+    InputPartitions,
+}
+
+impl AsRef<str> for HudiReadConfig {
+    fn as_ref(&self) -> &str {
+        match self {
+            Self::InputPartitions => "hoodie.read.input.partitions",
+        }
+    }
+}
+
+impl ConfigParser for HudiReadConfig {
+    type Output = HudiConfigValue;
+
+    fn default_value(&self) -> Option<HudiConfigValue> {
+        match self {
+            HudiReadConfig::InputPartitions => Some(HudiConfigValue::UInteger(0usize)),
+        }
+    }
+
+    fn parse_value(&self, configs: &HashMap<String, String>) -> Result<Self::Output> {
+        let get_result = configs
+            .get(self.as_ref())
+            .map(|v| v.as_str())
+            .ok_or(anyhow!("Config '{}' not found", self.as_ref()));
+
+        match self {
+            HudiReadConfig::InputPartitions => get_result
+                .and_then(|v| {
+                    usize::from_str(v).map_err(|e| {
+                        anyhow!(
+                            "Failed to parse '{}' for config '{}': {}",
+                            v,
+                            self.as_ref(),
+                            e
+                        )
+                    })
+                })
+                .map(HudiConfigValue::UInteger),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use crate::config::read::HudiReadConfig::InputPartitions;
+    use crate::config::ConfigParser;
+
+    #[test]
+    fn parse_valid_config_value() {
+        let options = HashMap::from([(InputPartitions.as_ref().to_string(), "100".to_string())]);
+        let value = InputPartitions.parse_value(&options).unwrap().to::<usize>();
+        assert_eq!(value, 100usize);
+    }
+
+    #[test]
+    fn parse_invalid_config_value() {
+        let options = HashMap::from([(InputPartitions.as_ref().to_string(), "foo".to_string())]);
+        let value = InputPartitions.parse_value(&options);
+        assert!(value.err().unwrap().to_string().starts_with(&format!(
+            "Failed to parse 'foo' for config '{}'",
+            InputPartitions.as_ref()
+        )));
+        assert_eq!(
+            InputPartitions
+                .parse_value_or_default(&options)
+                .to::<usize>(),
+            0
+        );
+    }
+}
diff --git a/crates/core/src/config/table.rs b/crates/core/src/config/table.rs
new file mode 100644
index 0000000..d503e32
--- /dev/null
+++ b/crates/core/src/config/table.rs
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use anyhow::anyhow;
+use anyhow::Result;
+use strum_macros::AsRefStr;
+
+use crate::config::{ConfigParser, HudiConfigValue};
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum HudiTableConfig {
+    BaseFileFormat,
+    Checksum,
+    DatabaseName,
+    DropsPartitionFields,
+    IsHiveStylePartitioning,
+    IsPartitionPathUrlencoded,
+    KeyGeneratorClass,
+    PartitionFields,
+    PrecombineField,
+    PopulatesMetaFields,
+    RecordKeyFields,
+    TableName,
+    TableType,
+    TableVersion,
+    TimelineLayoutVersion,
+}
+
+impl AsRef<str> for HudiTableConfig {
+    fn as_ref(&self) -> &str {
+        match self {
+            Self::BaseFileFormat => "hoodie.table.base.file.format",
+            Self::Checksum => "hoodie.table.checksum",
+            Self::DatabaseName => "hoodie.database.name",
+            Self::DropsPartitionFields => "hoodie.datasource.write.drop.partition.columns",
+            Self::IsHiveStylePartitioning => "hoodie.datasource.write.hive_style_partitioning",
+            Self::IsPartitionPathUrlencoded => "hoodie.datasource.write.partitionpath.urlencode",
+            Self::KeyGeneratorClass => "hoodie.table.keygenerator.class",
+            Self::PartitionFields => "hoodie.table.partition.fields",
+            Self::PrecombineField => "hoodie.table.precombine.field",
+            Self::PopulatesMetaFields => "hoodie.populate.meta.fields",
+            Self::RecordKeyFields => "hoodie.table.recordkey.fields",
+            Self::TableName => "hoodie.table.name",
+            Self::TableType => "hoodie.table.type",
+            Self::TableVersion => "hoodie.table.version",
+            Self::TimelineLayoutVersion => "hoodie.timeline.layout.version",
+        }
+    }
+}
+
+impl ConfigParser for HudiTableConfig {
+    type Output = HudiConfigValue;
+
+    fn default_value(&self) -> Option<Self::Output> {
+        match self {
+            Self::DatabaseName => Some(HudiConfigValue::String("default".to_string())),
+            Self::DropsPartitionFields => Some(HudiConfigValue::Boolean(false)),
+            Self::PopulatesMetaFields => Some(HudiConfigValue::Boolean(true)),
+            _ => None,
+        }
+    }
+
+    fn is_required(&self) -> bool {
+        matches!(
+            self,
+            Self::BaseFileFormat | Self::TableName | Self::TableType | Self::TableVersion
+        )
+    }
+
+    fn parse_value(&self, configs: &HashMap<String, String>) -> Result<Self::Output> {
+        let get_result = configs
+            .get(self.as_ref())
+            .map(|v| v.as_str())
+            .ok_or(anyhow!("Config '{}' not found", self.as_ref()));
+
+        match self {
+            Self::BaseFileFormat => get_result
+                .and_then(BaseFileFormatValue::from_str)
+                .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
+            Self::Checksum => get_result
+                .and_then(|v| isize::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Integer),
+            Self::DatabaseName => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+            Self::DropsPartitionFields => get_result
+                .and_then(|v| bool::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Boolean),
+            Self::IsHiveStylePartitioning => get_result
+                .and_then(|v| bool::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Boolean),
+            Self::IsPartitionPathUrlencoded => get_result
+                .and_then(|v| bool::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Boolean),
+            Self::KeyGeneratorClass => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+            Self::PartitionFields => get_result
+                .map(|v| HudiConfigValue::List(v.split(',').map(str::to_string).collect())),
+            Self::PrecombineField => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+            Self::PopulatesMetaFields => get_result
+                .and_then(|v| bool::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Boolean),
+            Self::RecordKeyFields => get_result
+                .map(|v| HudiConfigValue::List(v.split(',').map(str::to_string).collect())),
+            Self::TableName => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+            Self::TableType => get_result
+                .and_then(TableTypeValue::from_str)
+                .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
+            Self::TableVersion => get_result
+                .and_then(|v| isize::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Integer),
+            Self::TimelineLayoutVersion => get_result
+                .and_then(|v| isize::from_str(v).map_err(|e| anyhow!(e)))
+                .map(HudiConfigValue::Integer),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, AsRefStr)]
+pub enum TableTypeValue {
+    #[strum(serialize = "COPY_ON_WRITE")]
+    CopyOnWrite,
+    #[strum(serialize = "MERGE_ON_READ")]
+    MergeOnRead,
+}
+
+impl FromStr for TableTypeValue {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self> {
+        match s.to_ascii_lowercase().as_str() {
+            "copy_on_write" | "copy-on-write" | "cow" => Ok(Self::CopyOnWrite),
+            "merge_on_read" | "merge-on-read" | "mor" => Ok(Self::MergeOnRead),
+            _ => Err(anyhow!("Unsupported table type: {}", s)),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, AsRefStr)]
+pub enum BaseFileFormatValue {
+    #[strum(serialize = "parquet")]
+    Parquet,
+}
+
+impl FromStr for BaseFileFormatValue {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self> {
+        match s.to_ascii_lowercase().as_str() {
+            "parquet" => Ok(Self::Parquet),
+            _ => Err(anyhow!("Unsupported base file format: {}", s)),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use crate::config::table::{BaseFileFormatValue, TableTypeValue};
+
+    #[test]
+    fn create_table_type() {
+        assert_eq!(
+            TableTypeValue::from_str("cow").unwrap(),
+            TableTypeValue::CopyOnWrite
+        );
+        assert_eq!(
+            TableTypeValue::from_str("copy_on_write").unwrap(),
+            TableTypeValue::CopyOnWrite
+        );
+        assert_eq!(
+            TableTypeValue::from_str("COPY-ON-WRITE").unwrap(),
+            TableTypeValue::CopyOnWrite
+        );
+        assert_eq!(
+            TableTypeValue::from_str("MOR").unwrap(),
+            TableTypeValue::MergeOnRead
+        );
+        assert_eq!(
+            TableTypeValue::from_str("Merge_on_read").unwrap(),
+            TableTypeValue::MergeOnRead
+        );
+        assert_eq!(
+            TableTypeValue::from_str("Merge-on-read").unwrap(),
+            TableTypeValue::MergeOnRead
+        );
+        assert!(TableTypeValue::from_str("").is_err());
+        assert!(TableTypeValue::from_str("copyonwrite").is_err());
+        assert!(TableTypeValue::from_str("MERGEONREAD").is_err());
+        assert!(TableTypeValue::from_str("foo").is_err());
+    }
+
+    #[test]
+    fn create_base_file_format() {
+        assert_eq!(
+            BaseFileFormatValue::from_str("parquet").unwrap(),
+            BaseFileFormatValue::Parquet
+        );
+        assert_eq!(
+            BaseFileFormatValue::from_str("PArquet").unwrap(),
+            BaseFileFormatValue::Parquet
+        );
+        assert!(TableTypeValue::from_str("").is_err());
+        assert!(
+            TableTypeValue::from_str("orc").is_err(),
+            "orc is not yet supported."
+        );
+    }
+}
diff --git a/crates/core/src/table/config.rs b/crates/core/src/table/config.rs
deleted file mode 100644
index 6a1a6ce..0000000
--- a/crates/core/src/table/config.rs
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use anyhow::anyhow;
-use anyhow::Result;
-use std::str::FromStr;
-
-pub enum ConfigKey {
-    BaseFileFormat,
-    Checksum,
-    DatabaseName,
-    DropsPartitionFields,
-    IsHiveStylePartitioning,
-    IsPartitionPathUrlencoded,
-    KeyGeneratorClass,
-    PartitionFields,
-    PrecombineField,
-    PopulatesMetaFields,
-    RecordKeyFields,
-    TableName,
-    TableType,
-    TableVersion,
-    TimelineLayoutVersion,
-}
-
-impl AsRef<str> for ConfigKey {
-    fn as_ref(&self) -> &str {
-        match self {
-            Self::BaseFileFormat => "hoodie.table.base.file.format",
-            Self::Checksum => "hoodie.table.checksum",
-            Self::DatabaseName => "hoodie.database.name",
-            Self::DropsPartitionFields => "hoodie.datasource.write.drop.partition.columns",
-            Self::IsHiveStylePartitioning => "hoodie.datasource.write.hive_style_partitioning",
-            Self::IsPartitionPathUrlencoded => "hoodie.datasource.write.partitionpath.urlencode",
-            Self::KeyGeneratorClass => "hoodie.table.keygenerator.class",
-            Self::PartitionFields => "hoodie.table.partition.fields",
-            Self::PrecombineField => "hoodie.table.precombine.field",
-            Self::PopulatesMetaFields => "hoodie.populate.meta.fields",
-            Self::RecordKeyFields => "hoodie.table.recordkey.fields",
-            Self::TableName => "hoodie.table.name",
-            Self::TableType => "hoodie.table.type",
-            Self::TableVersion => "hoodie.table.version",
-            Self::TimelineLayoutVersion => "hoodie.timeline.layout.version",
-        }
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub enum TableType {
-    CopyOnWrite,
-    MergeOnRead,
-}
-
-impl FromStr for TableType {
-    type Err = anyhow::Error;
-
-    fn from_str(s: &str) -> Result<Self> {
-        match s.to_ascii_lowercase().as_str() {
-            "copy_on_write" | "copy-on-write" | "cow" => Ok(Self::CopyOnWrite),
-            "merge_on_read" | "merge-on-read" | "mor" => Ok(Self::MergeOnRead),
-            _ => Err(anyhow!("Unsupported table type: {}", s)),
-        }
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub enum BaseFileFormat {
-    Parquet,
-}
-
-impl FromStr for BaseFileFormat {
-    type Err = anyhow::Error;
-
-    fn from_str(s: &str) -> Result<Self> {
-        match s.to_ascii_lowercase().as_str() {
-            "parquet" => Ok(Self::Parquet),
-            _ => Err(anyhow!("Unsupported base file format: {}", s)),
-        }
-    }
-}
-#[cfg(test)]
-mod tests {
-    use crate::table::config::{BaseFileFormat, TableType};
-    use std::str::FromStr;
-
-    #[test]
-    fn create_table_type() {
-        assert_eq!(TableType::from_str("cow").unwrap(), TableType::CopyOnWrite);
-        assert_eq!(
-            TableType::from_str("copy_on_write").unwrap(),
-            TableType::CopyOnWrite
-        );
-        assert_eq!(
-            TableType::from_str("COPY-ON-WRITE").unwrap(),
-            TableType::CopyOnWrite
-        );
-        assert_eq!(TableType::from_str("MOR").unwrap(), TableType::MergeOnRead);
-        assert_eq!(
-            TableType::from_str("Merge_on_read").unwrap(),
-            TableType::MergeOnRead
-        );
-        assert_eq!(
-            TableType::from_str("Merge-on-read").unwrap(),
-            TableType::MergeOnRead
-        );
-        assert!(TableType::from_str("").is_err());
-        assert!(TableType::from_str("copyonwrite").is_err());
-        assert!(TableType::from_str("MERGEONREAD").is_err());
-        assert!(TableType::from_str("foo").is_err());
-    }
-
-    #[test]
-    fn create_base_file_format() {
-        assert_eq!(
-            BaseFileFormat::from_str("parquet").unwrap(),
-            BaseFileFormat::Parquet
-        );
-        assert_eq!(
-            BaseFileFormat::from_str("PArquet").unwrap(),
-            BaseFileFormat::Parquet
-        );
-        assert!(TableType::from_str("").is_err());
-        assert!(
-            TableType::from_str("orc").is_err(),
-            "orc is not yet supported."
-        );
-    }
-}
diff --git a/crates/core/src/table/fs_view.rs b/crates/core/src/table/fs_view.rs
index ad4f5f5..02d3ca4 100644
--- a/crates/core/src/table/fs_view.rs
+++ b/crates/core/src/table/fs_view.rs
@@ -25,6 +25,7 @@ use arrow::record_batch::RecordBatch;
 use dashmap::DashMap;
 use url::Url;
 
+use crate::config::HudiConfigs;
 use crate::file_group::{BaseFile, FileGroup, FileSlice};
 use crate::storage::file_info::FileInfo;
 use crate::storage::{get_leaf_dirs, Storage};
@@ -32,7 +33,7 @@ use crate::storage::{get_leaf_dirs, Storage};
 #[derive(Clone, Debug)]
 #[allow(dead_code)]
 pub struct FileSystemView {
-    props: Arc<HashMap<String, String>>,
+    configs: Arc<HudiConfigs>,
     storage: Arc<Storage>,
     partition_to_file_groups: Arc<DashMap<String, Vec<FileGroup>>>,
 }
@@ -41,7 +42,7 @@ impl FileSystemView {
     pub async fn new(
         base_url: Arc<Url>,
         storage_options: Arc<HashMap<String, String>>,
-        props: Arc<HashMap<String, String>>,
+        configs: Arc<HudiConfigs>,
     ) -> Result<Self> {
         let storage = Storage::new(base_url, storage_options)?;
         let partition_paths = Self::load_partition_paths(&storage).await?;
@@ -49,7 +50,7 @@ impl FileSystemView {
             Self::load_file_groups_for_partitions(&storage, partition_paths).await?;
         let partition_to_file_groups = Arc::new(DashMap::from_iter(partition_to_file_groups));
         Ok(FileSystemView {
-            props,
+            configs,
             storage,
             partition_to_file_groups,
         })
@@ -168,6 +169,7 @@ mod tests {
 
     use hudi_tests::TestTable;
 
+    use crate::config::HudiConfigs;
     use crate::storage::Storage;
     use crate::table::fs_view::FileSystemView;
 
@@ -208,7 +210,7 @@ mod tests {
         let fs_view = FileSystemView::new(
             Arc::new(base_url),
             Arc::new(HashMap::new()),
-            Arc::new(HashMap::new()),
+            Arc::new(HudiConfigs::empty()),
         )
         .await
         .unwrap();
diff --git a/crates/core/src/table/metadata.rs b/crates/core/src/table/metadata.rs
deleted file mode 100644
index fb2d41b..0000000
--- a/crates/core/src/table/metadata.rs
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-use crate::table::config::{BaseFileFormat, TableType};
-
-pub trait ProvidesTableMetadata {
-    fn base_file_format(&self) -> BaseFileFormat;
-
-    fn checksum(&self) -> i64;
-
-    fn database_name(&self) -> String;
-
-    fn drops_partition_fields(&self) -> bool;
-
-    fn is_hive_style_partitioning(&self) -> bool;
-
-    fn is_partition_path_urlencoded(&self) -> bool;
-
-    fn is_partitioned(&self) -> bool;
-
-    fn key_generator_class(&self) -> String;
-
-    fn location(&self) -> String;
-
-    fn partition_fields(&self) -> Vec<String>;
-
-    fn precombine_field(&self) -> String;
-
-    fn populates_meta_fields(&self) -> bool;
-
-    fn record_key_fields(&self) -> Vec<String>;
-
-    fn table_name(&self) -> String;
-
-    fn table_type(&self) -> TableType;
-
-    fn table_version(&self) -> u32;
-
-    fn timeline_layout_version(&self) -> u32;
-}
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 849f0e0..e09fd10 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -19,7 +19,6 @@
 
 use std::collections::HashMap;
 use std::io::{BufRead, BufReader};
-use std::str::FromStr;
 use std::sync::Arc;
 
 use anyhow::{anyhow, Context, Result};
@@ -27,25 +26,21 @@ use arrow::record_batch::RecordBatch;
 use arrow_schema::Schema;
 use url::Url;
 
+use crate::config::HudiConfigs;
 use crate::file_group::FileSlice;
 use crate::storage::utils::parse_uri;
 use crate::storage::Storage;
-use crate::table::config::BaseFileFormat;
-use crate::table::config::{ConfigKey, TableType};
 use crate::table::fs_view::FileSystemView;
-use crate::table::metadata::ProvidesTableMetadata;
 use crate::table::timeline::Timeline;
 
-mod config;
 mod fs_view;
-mod metadata;
 mod timeline;
 
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
 pub struct Table {
     pub base_url: Arc<Url>,
     pub storage_options: Arc<HashMap<String, String>>,
-    pub props: Arc<HashMap<String, String>>,
+    pub configs: Arc<HudiConfigs>,
     pub timeline: Timeline,
     pub file_system_view: FileSystemView,
 }
@@ -55,24 +50,24 @@ impl Table {
         let base_url = Arc::new(parse_uri(base_uri)?);
         let storage_options = Arc::new(storage_options);
 
-        let props = Self::load_properties(base_url.clone(), storage_options.clone())
+        let configs = Self::load_properties(base_url.clone(), storage_options.clone())
             .await
             .context("Failed to load table properties")?;
+        let configs = Arc::new(configs);
 
-        let props = Arc::new(props);
-        let timeline = Timeline::new(base_url.clone(), storage_options.clone(), props.clone())
+        let timeline = Timeline::new(base_url.clone(), storage_options.clone(), configs.clone())
             .await
             .context("Failed to load timeline")?;
 
         let file_system_view =
-            FileSystemView::new(base_url.clone(), storage_options.clone(), props.clone())
+            FileSystemView::new(base_url.clone(), storage_options.clone(), configs.clone())
                 .await
                 .context("Failed to load file system view")?;
 
         Ok(Table {
             base_url,
             storage_options,
-            props,
+            configs,
             timeline,
             file_system_view,
         })
@@ -81,7 +76,7 @@ impl Table {
     async fn load_properties(
         base_url: Arc<Url>,
         storage_options: Arc<HashMap<String, String>>,
-    ) -> Result<HashMap<String, String>> {
+    ) -> Result<HudiConfigs> {
         let storage = Storage::new(base_url, storage_options)?;
         let data = storage.get_file_data(".hoodie/hoodie.properties").await?;
         let cursor = std::io::Cursor::new(data);
@@ -98,14 +93,7 @@ impl Table {
             let value = parts.next().unwrap_or("").to_owned();
             properties.insert(key, value);
         }
-        Ok(properties)
-    }
-
-    pub fn get_property(&self, key: &str) -> &str {
-        match self.props.get(key) {
-            Some(value) => value,
-            None => panic!("Failed to retrieve property {}", key),
-        }
+        Ok(HudiConfigs::new(properties))
     }
 
     pub async fn get_schema(&self) -> Result<Schema> {
@@ -178,101 +166,24 @@ impl Table {
     }
 }
 
-impl ProvidesTableMetadata for Table {
-    fn base_file_format(&self) -> BaseFileFormat {
-        BaseFileFormat::from_str(self.get_property(ConfigKey::BaseFileFormat.as_ref())).unwrap()
-    }
-
-    fn checksum(&self) -> i64 {
-        i64::from_str(self.get_property(ConfigKey::Checksum.as_ref())).unwrap()
-    }
-
-    fn database_name(&self) -> String {
-        match self.props.get(ConfigKey::DatabaseName.as_ref()) {
-            Some(value) => value.to_string(),
-            None => "default".to_string(),
-        }
-    }
-
-    fn drops_partition_fields(&self) -> bool {
-        bool::from_str(self.get_property(ConfigKey::DropsPartitionFields.as_ref())).unwrap()
-    }
-
-    fn is_hive_style_partitioning(&self) -> bool {
-        bool::from_str(self.get_property(ConfigKey::IsHiveStylePartitioning.as_ref())).unwrap()
-    }
-
-    fn is_partition_path_urlencoded(&self) -> bool {
-        bool::from_str(self.get_property(ConfigKey::IsPartitionPathUrlencoded.as_ref())).unwrap()
-    }
-
-    fn is_partitioned(&self) -> bool {
-        !self
-            .key_generator_class()
-            .ends_with("NonpartitionedKeyGenerator")
-    }
-
-    fn key_generator_class(&self) -> String {
-        self.get_property(ConfigKey::KeyGeneratorClass.as_ref())
-            .to_string()
-    }
-
-    fn location(&self) -> String {
-        self.base_url.path().to_string()
-    }
-
-    fn partition_fields(&self) -> Vec<String> {
-        self.get_property(ConfigKey::PartitionFields.as_ref())
-            .split(',')
-            .map(str::to_string)
-            .collect()
-    }
-
-    fn precombine_field(&self) -> String {
-        self.get_property(ConfigKey::PrecombineField.as_ref())
-            .to_string()
-    }
-
-    fn populates_meta_fields(&self) -> bool {
-        bool::from_str(self.get_property(ConfigKey::PopulatesMetaFields.as_ref())).unwrap()
-    }
-
-    fn record_key_fields(&self) -> Vec<String> {
-        self.get_property(ConfigKey::RecordKeyFields.as_ref())
-            .split(',')
-            .map(str::to_string)
-            .collect()
-    }
-
-    fn table_name(&self) -> String {
-        self.get_property(ConfigKey::TableName.as_ref()).to_string()
-    }
-
-    fn table_type(&self) -> TableType {
-        TableType::from_str(self.get_property(ConfigKey::TableType.as_ref())).unwrap()
-    }
-
-    fn table_version(&self) -> u32 {
-        u32::from_str(self.get_property(ConfigKey::TableVersion.as_ref())).unwrap()
-    }
-
-    fn timeline_layout_version(&self) -> u32 {
-        u32::from_str(self.get_property(ConfigKey::TimelineLayoutVersion.as_ref())).unwrap()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::collections::{HashMap, HashSet};
     use std::fs::canonicalize;
+    use std::panic;
     use std::path::Path;
 
-    use hudi_tests::TestTable;
+    use url::Url;
 
+    use hudi_tests::{assert_not, TestTable};
+
+    use crate::config::table::HudiTableConfig::{
+        BaseFileFormat, Checksum, DatabaseName, DropsPartitionFields, IsHiveStylePartitioning,
+        IsPartitionPathUrlencoded, KeyGeneratorClass, PartitionFields, PopulatesMetaFields,
+        PrecombineField, RecordKeyFields, TableName, TableType, TableVersion,
+        TimelineLayoutVersion,
+    };
     use crate::storage::utils::join_url_segments;
-    use crate::table::config::BaseFileFormat::Parquet;
-    use crate::table::config::TableType::CopyOnWrite;
-    use crate::table::metadata::ProvidesTableMetadata;
     use crate::table::Table;
 
     #[tokio::test]
@@ -411,33 +322,146 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn hudi_table_get_table_metadata() {
-        let base_path = canonicalize(Path::new(
-            "tests/data/table_metadata/sample_table_properties",
-        ))
-        .unwrap();
-        let table = Table::new(base_path.to_str().unwrap(), HashMap::new())
-            .await
-            .unwrap();
-        assert_eq!(table.base_file_format(), Parquet);
-        assert_eq!(table.checksum(), 3761586722);
-        assert_eq!(table.database_name(), "default");
-        assert!(!table.drops_partition_fields());
-        assert!(!table.is_hive_style_partitioning());
-        assert!(!table.is_partition_path_urlencoded());
-        assert!(table.is_partitioned());
+    async fn validate_invalid_table_props() {
+        let base_url =
+            Url::from_file_path(canonicalize(Path::new("tests/data/table_props_invalid")).unwrap())
+                .unwrap();
+        let table = Table::new(base_url.as_str(), HashMap::new()).await.unwrap();
+        let configs = table.configs;
+        assert!(
+            configs.validate(BaseFileFormat).is_err(),
+            "required config is missing"
+        );
+        assert!(configs.validate(Checksum).is_err());
+        assert!(
+            configs.validate(DatabaseName).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(configs.validate(DropsPartitionFields).is_err());
+        assert!(configs.validate(IsHiveStylePartitioning).is_err());
+        assert!(configs.validate(IsPartitionPathUrlencoded).is_err());
+        assert!(
+            configs.validate(KeyGeneratorClass).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(
+            configs.validate(PartitionFields).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(
+            configs.validate(PrecombineField).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(
+            configs.validate(PopulatesMetaFields).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(
+            configs.validate(RecordKeyFields).is_ok(),
+            "non-required config is missing"
+        );
+        assert!(
+            configs.validate(TableName).is_err(),
+            "required config is missing"
+        );
+        assert!(
+            configs.validate(TableType).is_ok(),
+            "Valid table type value"
+        );
+        assert!(configs.validate(TableVersion).is_err());
+        assert!(configs.validate(TimelineLayoutVersion).is_err());
+    }
+
+    #[tokio::test]
+    async fn get_invalid_table_props() {
+        let base_url =
+            Url::from_file_path(canonicalize(Path::new("tests/data/table_props_invalid")).unwrap())
+                .unwrap();
+        let table = Table::new(base_url.as_str(), HashMap::new()).await.unwrap();
+        let configs = table.configs;
+        assert!(configs.get(BaseFileFormat).is_err());
+        assert!(configs.get(Checksum).is_err());
+        assert!(configs.get(DatabaseName).is_err());
+        assert!(configs.get(DropsPartitionFields).is_err());
+        assert!(configs.get(IsHiveStylePartitioning).is_err());
+        assert!(configs.get(IsPartitionPathUrlencoded).is_err());
+        assert!(configs.get(KeyGeneratorClass).is_err());
+        assert!(configs.get(PartitionFields).is_err());
+        assert!(configs.get(PrecombineField).is_err());
+        assert!(configs.get(PopulatesMetaFields).is_err());
+        assert!(configs.get(RecordKeyFields).is_err());
+        assert!(configs.get(TableName).is_err());
+        assert!(configs.get(TableType).is_ok(), "Valid table type value");
+        assert!(configs.get(TableVersion).is_err());
+        assert!(configs.get(TimelineLayoutVersion).is_err());
+    }
+
+    #[tokio::test]
+    async fn get_default_for_invalid_table_props() {
+        let base_url =
+            Url::from_file_path(canonicalize(Path::new("tests/data/table_props_invalid")).unwrap())
+                .unwrap();
+        let table = Table::new(base_url.as_str(), HashMap::new()).await.unwrap();
+        let configs = table.configs;
+        assert!(panic::catch_unwind(|| configs.get_or_default(BaseFileFormat)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(Checksum)).is_err());
+        assert_eq!(
+            configs.get_or_default(DatabaseName).to::<String>(),
+            "default"
+        );
+        assert_not!(configs.get_or_default(DropsPartitionFields).to::<bool>());
+        assert!(panic::catch_unwind(|| configs.get_or_default(IsHiveStylePartitioning)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(IsPartitionPathUrlencoded)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(KeyGeneratorClass)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(PartitionFields)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(PrecombineField)).is_err());
+        assert!(configs.get_or_default(PopulatesMetaFields).to::<bool>());
+        assert!(panic::catch_unwind(|| configs.get_or_default(RecordKeyFields)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(TableName)).is_err());
+        assert_eq!(
+            configs.get_or_default(TableType).to::<String>(),
+            "COPY_ON_WRITE"
+        );
+        assert!(panic::catch_unwind(|| configs.get_or_default(TableVersion)).is_err());
+        assert!(panic::catch_unwind(|| configs.get_or_default(TimelineLayoutVersion)).is_err());
+    }
+
+    #[tokio::test]
+    async fn get_valid_table_props() {
+        let base_url =
+            Url::from_file_path(canonicalize(Path::new("tests/data/table_props_valid")).unwrap())
+                .unwrap();
+        let table = Table::new(base_url.as_str(), HashMap::new()).await.unwrap();
+        let configs = table.configs;
+        assert_eq!(
+            configs.get(BaseFileFormat).unwrap().to::<String>(),
+            "parquet"
+        );
+        assert_eq!(configs.get(Checksum).unwrap().to::<isize>(), 3761586722);
+        assert_eq!(configs.get(DatabaseName).unwrap().to::<String>(), "db");
+        assert!(!configs.get(DropsPartitionFields).unwrap().to::<bool>());
+        assert!(!configs.get(IsHiveStylePartitioning).unwrap().to::<bool>());
+        assert!(!configs.get(IsPartitionPathUrlencoded).unwrap().to::<bool>());
         assert_eq!(
-            table.key_generator_class(),
+            configs.get(KeyGeneratorClass).unwrap().to::<String>(),
             "org.apache.hudi.keygen.SimpleKeyGenerator"
         );
-        assert_eq!(table.location(), base_path.to_str().unwrap());
-        assert_eq!(table.partition_fields(), vec!["city"]);
-        assert_eq!(table.precombine_field(), "ts");
-        assert!(table.populates_meta_fields());
-        assert_eq!(table.record_key_fields(), vec!["uuid"]);
-        assert_eq!(table.table_name(), "trips");
-        assert_eq!(table.table_type(), CopyOnWrite);
-        assert_eq!(table.table_version(), 6);
-        assert_eq!(table.timeline_layout_version(), 1);
+        assert_eq!(
+            configs.get(PartitionFields).unwrap().to::<Vec<String>>(),
+            vec!["city"]
+        );
+        assert_eq!(configs.get(PrecombineField).unwrap().to::<String>(), "ts");
+        assert!(configs.get(PopulatesMetaFields).unwrap().to::<bool>());
+        assert_eq!(
+            configs.get(RecordKeyFields).unwrap().to::<Vec<String>>(),
+            vec!["uuid"]
+        );
+        assert_eq!(configs.get(TableName).unwrap().to::<String>(), "trips");
+        assert_eq!(
+            configs.get(TableType).unwrap().to::<String>(),
+            "COPY_ON_WRITE"
+        );
+        assert_eq!(configs.get(TableVersion).unwrap().to::<isize>(), 6);
+        assert_eq!(configs.get(TimelineLayoutVersion).unwrap().to::<isize>(), 1);
     }
 }
diff --git a/crates/core/src/table/timeline.rs b/crates/core/src/table/timeline.rs
index 51601a4..d2d476d 100644
--- a/crates/core/src/table/timeline.rs
+++ b/crates/core/src/table/timeline.rs
@@ -29,6 +29,7 @@ use parquet::arrow::parquet_to_arrow_schema;
 use serde_json::{Map, Value};
 use url::Url;
 
+use crate::config::HudiConfigs;
 use crate::storage::utils::split_filename;
 use crate::storage::Storage;
 
@@ -76,7 +77,7 @@ impl Instant {
 #[derive(Clone, Debug)]
 #[allow(dead_code)]
 pub struct Timeline {
-    props: Arc<HashMap<String, String>>,
+    configs: Arc<HudiConfigs>,
     storage: Arc<Storage>,
     pub instants: Vec<Instant>,
 }
@@ -85,13 +86,13 @@ impl Timeline {
     pub async fn new(
         base_url: Arc<Url>,
         storage_options: Arc<HashMap<String, String>>,
-        props: Arc<HashMap<String, String>>,
+        configs: Arc<HudiConfigs>,
     ) -> Result<Self> {
         let storage = Storage::new(base_url, storage_options)?;
         let instants = Self::load_completed_commit_instants(&storage).await?;
         Ok(Self {
             storage,
-            props,
+            configs,
             instants,
         })
     }
@@ -179,6 +180,7 @@ mod tests {
 
     use hudi_tests::TestTable;
 
+    use crate::config::HudiConfigs;
     use crate::table::timeline::{Instant, State, Timeline};
 
     #[tokio::test]
@@ -187,7 +189,7 @@ mod tests {
         let timeline = Timeline::new(
             Arc::new(base_url),
             Arc::new(HashMap::new()),
-            Arc::new(HashMap::new()),
+            Arc::new(HudiConfigs::empty()),
         )
         .await
         .unwrap();
@@ -201,7 +203,7 @@ mod tests {
         let timeline = Timeline::new(
             Arc::new(base_url),
             Arc::new(HashMap::new()),
-            Arc::new(HashMap::new()),
+            Arc::new(HudiConfigs::empty()),
         )
         .await
         .unwrap();
@@ -222,7 +224,7 @@ mod tests {
         let timeline = Timeline::new(
             Arc::new(base_url),
             Arc::new(HashMap::new()),
-            Arc::new(HashMap::new()),
+            Arc::new(HudiConfigs::empty()),
         )
         .await
         .unwrap();
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 3b0a73a..1435d94 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -38,8 +38,8 @@ use datafusion_expr::{Expr, TableType};
 use datafusion_physical_expr::create_physical_expr;
 use DataFusionError::Execution;
 
-use hudi_core::config::HudiConfig::ReadInputPartitions;
-use hudi_core::config::OptionsParser;
+use hudi_core::config::read::HudiReadConfig::InputPartitions;
+use hudi_core::config::ConfigParser;
 use hudi_core::storage::utils::{get_scheme_authority, parse_uri};
 use hudi_core::HudiTable;
 
@@ -51,9 +51,9 @@ pub struct HudiDataSource {
 
 impl HudiDataSource {
     pub async fn new(base_uri: &str, options: HashMap<String, String>) -> Result<Self> {
-        let input_partitions = ReadInputPartitions
+        let input_partitions = InputPartitions
             .parse_value_or_default(&options)
-            .cast::<usize>();
+            .to::<usize>();
         match HudiTable::new(base_uri, options).await {
             Ok(t) => Ok(Self {
                 table: Arc::new(t),
@@ -136,15 +136,13 @@ mod tests {
     use datafusion::prelude::{SessionConfig, SessionContext};
     use datafusion_common::ScalarValue;
 
-    use hudi_core::config::HudiConfig::ReadInputPartitions;
-    use hudi_tests::utils::get_bool_column;
+    use hudi_core::config::read::HudiReadConfig::InputPartitions;
     use hudi_tests::TestTable::{
-        V6Nonpartitioned, V6SimplekeygenHivestyleNoMetafields, V6SimplekeygenNonhivestyle,
-        V6TimebasedkeygenNonhivestyle,
+        V6ComplexkeygenHivestyle, V6Nonpartitioned, V6SimplekeygenHivestyleNoMetafields,
+        V6SimplekeygenNonhivestyle, V6TimebasedkeygenNonhivestyle,
     };
     use hudi_tests::{utils, TestTable};
-    use utils::{get_i32_column, get_str_column};
-    use TestTable::V6ComplexkeygenHivestyle;
+    use utils::{get_bool_column, get_i32_column, get_str_column};
 
     use crate::HudiDataSource;
 
@@ -215,7 +213,7 @@ mod tests {
             println!(">>> testing for {}", test_table.as_ref());
             let ctx = prepare_session_context(
                 test_table,
-                &[(ReadInputPartitions.as_ref().to_string(), "2".to_string())],
+                &[(InputPartitions.as_ref().to_string(), "2".to_string())],
             )
             .await;
 


Test patch:
diff --git a/crates/core/tests/data/table_props_invalid/.hoodie/hoodie.properties b/crates/core/tests/data/table_props_invalid/.hoodie/hoodie.properties
new file mode 100644
index 0000000..242d34f
--- /dev/null
+++ b/crates/core/tests/data/table_props_invalid/.hoodie/hoodie.properties
@@ -0,0 +1,27 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+hoodie.table.type=cow
+hoodie.table.base.file.format=foo
+hoodie.table.checksum=bar
+hoodie.table.version=baz
+hoodie.timeline.layout.version=_
+hoodie.datasource.write.drop.partition.columns=1
+hoodie.datasource.write.hive_style_partitioning=0
+hoodie.datasource.write.partitionpath.urlencode=-1
diff --git a/crates/core/tests/data/table_metadata/sample_table_properties/.hoodie/hoodie.properties b/crates/core/tests/data/table_props_valid/.hoodie/hoodie.properties
similarity index 96%
rename from crates/core/tests/data/table_metadata/sample_table_properties/.hoodie/hoodie.properties
rename to crates/core/tests/data/table_props_valid/.hoodie/hoodie.properties
index 9c0c7e2..d1e5ac6 100644
--- a/crates/core/tests/data/table_metadata/sample_table_properties/.hoodie/hoodie.properties
+++ b/crates/core/tests/data/table_props_valid/.hoodie/hoodie.properties
@@ -17,7 +17,7 @@
 # under the License.
 #
 
-hoodie.table.type=COPY_ON_WRITE
+hoodie.table.type=copy_on_write
 hoodie.table.metadata.partitions=files
 hoodie.table.precombine.field=ts
 hoodie.table.partition.fields=city
@@ -28,6 +28,7 @@ hoodie.table.checksum=3761586722
 hoodie.datasource.write.drop.partition.columns=false
 hoodie.table.recordkey.fields=uuid
 hoodie.table.name=trips
+hoodie.database.name=db
 hoodie.partition.metafile.use.base.format=false
 hoodie.datasource.write.hive_style_partitioning=false
 hoodie.table.metadata.partitions.inflight=
diff --git a/crates/tests/src/utils.rs b/crates/tests/src/utils.rs
index 0f7d44b..73fafd9 100644
--- a/crates/tests/src/utils.rs
+++ b/crates/tests/src/utils.rs
@@ -20,6 +20,20 @@
 use arrow::record_batch::RecordBatch;
 use arrow_array::{Array, BooleanArray, Int32Array, StringArray};
 
+#[macro_export]
+macro_rules! assert_not {
+    ($cond:expr) => {
+        if $cond {
+            panic!("assertion failed: condition is true");
+        }
+    };
+    ($cond:expr, $($arg:tt)+) => {
+        if $cond {
+            panic!("assertion failed: condition is true: {}", format_args!($($arg)+));
+        }
+    };
+}
+
 #[macro_export]
 macro_rules! assert_approx_eq {
     ($a:expr, $b:expr, $delta:expr) => {{