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

Issue:
Define Hudi error types across hudi-core
Use `thiserror` to define error enums to fit different error handling flows, and replace `anyhow::Error` usage.
feat: define Hudi error types across hudi-core
## Description

Use `thiserror` instead of `anyhow` to define the error for hudi-core crate.



<!--- Describe your changes in detail -->

<!--- If it fixes an open issue, please link to the issue here. -->

#72 

<!--- Please link any related issues and PRs as well. -->

## How are the changes test-covered

- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
  - [ ] Details are described below



Discussion / hints:
@codope any plan to resume the work? Updated the milestone/release date.
does this still wip? while i try to add some create table methods, just found this is a little chaos.

maybe i will try to refact that.

@codope @xushiyan 
@gohalo Please go ahead with your refactoring. I can help with review. I had opened https://github.com/apache/hudi-rs/pull/59 to refactor error handling but it is incomplete and preferably we shall use [thiserror](https://docs.rs/thiserror/latest/thiserror/). You can open a separate PR, and I will close mine in your favour.
Submit a pr, just replaced the anyhow package, but still have some problems to fix.

First, how to classified the errors.
1. User configurations, invalid file/object path, unsupported features.
2. Communication failed, mostly from the obejct_store crate.
3. Invalid hoodie table, including file layout, file format, crashed data file and so on. Including from parquet arrow crates.
4. and more.

Second, how to simplify the error generation or mapping process. `thiserror` provide many macros for this, such as `error` `from` etc. But still have many situation to simplify, through some trait like `Into`.




@xushiyan @codope 
Hope this codes could be reviewed as soon as possible, because it refact the core crate, and almost every new merged code should refact again. 😊
## [Codecov](https://app.codecov.io/gh/apache/hudi-rs/pull/124?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `71.92982%` with `32 lines` in your changes missing coverage. Please review.
> Project coverage is 90.53%. Comparing base [(`a2738da`)](https://app.codecov.io/gh/apache/hudi-rs/commit/a2738da6c15378cae1774a917957a404eb17b242?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`02c659b`)](https://app.codecov.io/gh/apache/hudi-rs/commit/02c659b3399cef272d09edb2acc860424f03a69a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 2 commits behind head on main.

| [Files with missing lines](https://app.codecov.io/gh/apache/hudi-rs/pull/124?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [crates/core/src/storage/utils.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fstorage%2Futils.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3N0b3JhZ2UvdXRpbHMucnM=) | 57.14% | [6 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/table/builder.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ftable%2Fbuilder.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3RhYmxlL2J1aWxkZXIucnM=) | 37.50% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [python/src/internal.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=python%2Fsrc%2Finternal.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHl0aG9uL3NyYy9pbnRlcm5hbC5ycw==) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/table/mod.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ftable%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3RhYmxlL21vZC5ycw==) | 20.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/table/partition.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ftable%2Fpartition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3RhYmxlL3BhcnRpdGlvbi5ycw==) | 55.55% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/storage/mod.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fstorage%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3N0b3JhZ2UvbW9kLnJz) | 76.92% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/config/internal.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fconfig%2Finternal.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL2NvbmZpZy9pbnRlcm5hbC5ycw==) | 60.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/config/table.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fconfig%2Ftable.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL2NvbmZpZy90YWJsZS5ycw==) | 93.54% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/table/timeline.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ftable%2Ftimeline.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3RhYmxlL3RpbWVsaW5lLnJz) | 91.66% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/124?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #124      +/-   ##
==========================================
- Coverage   91.78%   90.53%   -1.26%     
==========================================
  Files          20       21       +1     
  Lines         962     1004      +42     
==========================================
+ Hits          883      909      +26     
- Misses         79       95      +16     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/hudi-rs/pull/124?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).

@gohalo any plan to resume the work here by resolving conflicts first? we've cut release branch so master won't have much changes for some time.
OK, will fix that next week
@gohalo do you have time to resume work on this? in case you're occupied, maybe @TheR1sing3un could take this over? as this is more of a foundational work, we want to move forward sooner, ideally land this within the next 7 days. thanks!
> @gohalo do you have time to resume work on this? in case you're occupied, maybe @TheR1sing3un could take this over? as this is more of a foundational work, we want to move forward sooner, ideally land this within the next 7 days. thanks!

working on that, still have some conflicts to fix now

@xushiyan the conflicts have fixed, i will check the coverage recently, you could review the code first.
> @xushiyan the conflicts have fixed, i will check the coverage recently, you could review the code first.

Thanks for the fix! will go through this today.
I'll work on a follow up PR to get the coverage level back. merging it now.

Fix patch:
diff --git a/crates/core/src/config/error.rs b/crates/core/src/config/error.rs
new file mode 100644
index 00000000..6d852f6c
--- /dev/null
+++ b/crates/core/src/config/error.rs
@@ -0,0 +1,47 @@
+/*
+ * 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::num::ParseIntError;
+use std::str::ParseBoolError;
+use thiserror::Error;
+
+pub type Result<T, E = ConfigError> = std::result::Result<T, E>;
+
+#[derive(Error, Debug)]
+pub enum ConfigError {
+    /// ParseBool(key, value, source)
+    #[error("Failed to parse '{1}' for config '{0}': {2}")]
+    ParseBool(String, String, ParseBoolError),
+
+    /// ParseInt(key, value, source)
+    #[error("Failed to parse '{1}' for config '{0}': {2}")]
+    ParseInt(String, String, ParseIntError),
+
+    /// ParseLine(message)
+    #[error("Failed to parse line: {0}'")]
+    ParseLine(String),
+
+    #[error("Config value '{0}' is not valid")]
+    InvalidValue(String),
+
+    #[error("Config '{0}' not found")]
+    NotFound(String),
+
+    #[error("Config value '{0}' is not supported")]
+    UnsupportedValue(String),
+}
diff --git a/crates/core/src/config/internal.rs b/crates/core/src/config/internal.rs
index 4fdc2765..61e1a880 100644
--- a/crates/core/src/config/internal.rs
+++ b/crates/core/src/config/internal.rs
@@ -23,11 +23,9 @@ use std::str::FromStr;
 
 use strum_macros::EnumIter;
 
-use crate::{
-    config::{ConfigParser, HudiConfigValue},
-    CoreError::{ConfigNotFound, InvalidConfig},
-    Result,
-};
+use crate::config::error::ConfigError::{NotFound, ParseBool};
+use crate::config::Result;
+use crate::config::{ConfigParser, HudiConfigValue};
 
 /// Configurations for internal use.
 ///
@@ -67,15 +65,12 @@ impl ConfigParser for HudiInternalConfig {
         let get_result = configs
             .get(self.as_ref())
             .map(|v| v.as_str())
-            .ok_or(ConfigNotFound(self.as_ref().to_string()));
+            .ok_or(NotFound(self.key()));
 
         match self {
             Self::SkipConfigValidation => get_result
                 .and_then(|v| {
-                    bool::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::SkipConfigValidation.as_ref(),
-                        source: Box::new(e),
-                    })
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
         }
diff --git a/crates/core/src/config/mod.rs b/crates/core/src/config/mod.rs
index f586a896..bc554bdb 100644
--- a/crates/core/src/config/mod.rs
+++ b/crates/core/src/config/mod.rs
@@ -21,14 +21,18 @@ use std::any::type_name;
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use crate::{storage::utils::parse_uri, Result};
+use crate::config::error::{ConfigError, Result};
+use crate::storage::error::Result as StorageResult;
+use crate::storage::util::parse_uri;
 use serde::{Deserialize, Serialize};
 use url::Url;
+use ConfigError::NotFound;
 
+pub mod error;
 pub mod internal;
 pub mod read;
 pub mod table;
-pub mod utils;
+pub mod util;
 
 pub const HUDI_CONF_DIR: &str = "HUDI_CONF_DIR";
 
@@ -40,6 +44,10 @@ pub trait ConfigParser: AsRef<str> {
     /// Supplies the default value of the configuration.
     fn default_value(&self) -> Option<Self::Output>;
 
+    fn key(&self) -> String {
+        self.as_ref().to_string()
+    }
+
     /// To indicate if the configuration is required or not, this helps in validation.
     fn is_required(&self) -> bool {
         false
@@ -50,8 +58,7 @@ pub trait ConfigParser: AsRef<str> {
         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"
+                if !self.is_required() && matches!(e, NotFound(_)) {
                     Ok(())
                 } else {
                     Err(e)
@@ -98,7 +105,7 @@ impl HudiConfigValue {
 
     /// A convenience method to convert [HudiConfigValue] to [Url] when the value is a [String] and is intended to be a URL.
     /// Panic if the value is not a [String].
-    pub fn to_url(self) -> Result<Url> {
+    pub fn to_url(self) -> StorageResult<Url> {
         match self {
             HudiConfigValue::String(v) => parse_uri(&v),
             _ => panic!(
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
index a9389538..ece692ef 100644
--- a/crates/core/src/config/read.rs
+++ b/crates/core/src/config/read.rs
@@ -23,11 +23,9 @@ use std::str::FromStr;
 
 use strum_macros::EnumIter;
 
-use crate::{
-    config::{ConfigParser, HudiConfigValue},
-    CoreError::{ConfigNotFound, InvalidConfig},
-    Result,
-};
+use crate::config::error::ConfigError::{NotFound, ParseInt};
+use crate::config::Result;
+use crate::config::{ConfigParser, HudiConfigValue};
 
 /// Configurations for reading Hudi tables.
 ///
@@ -79,15 +77,12 @@ impl ConfigParser for HudiReadConfig {
         let get_result = configs
             .get(self.as_ref())
             .map(|v| v.as_str())
-            .ok_or(ConfigNotFound(self.as_ref().to_string()));
+            .ok_or(NotFound(self.key()));
 
         match self {
             Self::InputPartitions => get_result
                 .and_then(|v| {
-                    usize::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::InputPartitions.as_ref(),
-                        source: Box::new(e),
-                    })
+                    usize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::UInteger),
             Self::AsOfTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
@@ -97,10 +92,8 @@ impl ConfigParser for HudiReadConfig {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use crate::config::read::HudiReadConfig::InputPartitions;
-    use crate::config::ConfigParser;
-    use crate::CoreError::InvalidConfig;
-    use std::collections::HashMap;
 
     #[test]
     fn parse_valid_config_value() {
@@ -113,10 +106,7 @@ mod tests {
     fn parse_invalid_config_value() {
         let options = HashMap::from([(InputPartitions.as_ref().to_string(), "foo".to_string())]);
         let value = InputPartitions.parse_value(&options);
-        assert!(matches!(
-            value.unwrap_err(),
-            InvalidConfig { item: _, source: _ }
-        ));
+        assert!(matches!(value.unwrap_err(), ParseInt(_, _, _)));
         assert_eq!(
             InputPartitions
                 .parse_value_or_default(&options)
diff --git a/crates/core/src/config/table.rs b/crates/core/src/config/table.rs
index bb19066c..5b82b4cd 100644
--- a/crates/core/src/config/table.rs
+++ b/crates/core/src/config/table.rs
@@ -23,11 +23,12 @@ use std::str::FromStr;
 
 use strum_macros::{AsRefStr, EnumIter};
 
-use crate::{
-    config::{ConfigParser, HudiConfigValue},
-    CoreError::{self, ConfigNotFound, InvalidConfig, Unsupported},
-    Result,
+use crate::config::error::ConfigError;
+use crate::config::error::ConfigError::{
+    InvalidValue, NotFound, ParseBool, ParseInt, UnsupportedValue,
 };
+use crate::config::Result;
+use crate::config::{ConfigParser, HudiConfigValue};
 
 /// Configurations for Hudi tables, most of them are persisted in `hoodie.properties`.
 ///
@@ -148,7 +149,7 @@ impl ConfigParser for HudiTableConfig {
         let get_result = configs
             .get(self.as_ref())
             .map(|v| v.as_str())
-            .ok_or(ConfigNotFound(self.as_ref().to_string()));
+            .ok_or(NotFound(self.key()));
 
         match self {
             Self::BaseFileFormat => get_result
@@ -157,35 +158,23 @@ impl ConfigParser for HudiTableConfig {
             Self::BasePath => get_result.map(|v| HudiConfigValue::String(v.to_string())),
             Self::Checksum => get_result
                 .and_then(|v| {
-                    isize::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::Checksum.as_ref(),
-                        source: Box::new(e),
-                    })
+                    isize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), 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| InvalidConfig {
-                        item: Self::DropsPartitionFields.as_ref(),
-                        source: Box::new(e),
-                    })
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
             Self::IsHiveStylePartitioning => get_result
                 .and_then(|v| {
-                    bool::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::IsHiveStylePartitioning.as_ref(),
-                        source: Box::new(e),
-                    })
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
             Self::IsPartitionPathUrlencoded => get_result
                 .and_then(|v| {
-                    bool::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::IsPartitionPathUrlencoded.as_ref(),
-                        source: Box::new(e),
-                    })
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
             Self::KeyGeneratorClass => get_result.map(|v| HudiConfigValue::String(v.to_string())),
@@ -194,10 +183,7 @@ impl ConfigParser for HudiTableConfig {
             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| InvalidConfig {
-                        item: Self::PopulatesMetaFields.as_ref(),
-                        source: Box::new(e),
-                    })
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
             Self::RecordKeyFields => get_result
@@ -208,18 +194,12 @@ impl ConfigParser for HudiTableConfig {
                 .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
             Self::TableVersion => get_result
                 .and_then(|v| {
-                    isize::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::TableVersion.as_ref(),
-                        source: Box::new(e),
-                    })
+                    isize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Integer),
             Self::TimelineLayoutVersion => get_result
                 .and_then(|v| {
-                    isize::from_str(v).map_err(|e| InvalidConfig {
-                        item: Self::TimelineLayoutVersion.as_ref(),
-                        source: Box::new(e),
-                    })
+                    isize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Integer),
         }
@@ -236,13 +216,13 @@ pub enum TableTypeValue {
 }
 
 impl FromStr for TableTypeValue {
-    type Err = CoreError;
+    type Err = ConfigError;
 
-    fn from_str(s: &str) -> Result<Self> {
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
         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),
-            v => Err(Unsupported(format!("Unsupported table type {}", v))),
+            v => Err(InvalidValue(v.to_string())),
         }
     }
 }
@@ -255,21 +235,20 @@ pub enum BaseFileFormatValue {
 }
 
 impl FromStr for BaseFileFormatValue {
-    type Err = CoreError;
+    type Err = ConfigError;
 
-    fn from_str(s: &str) -> Result<Self> {
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_ascii_lowercase().as_str() {
             "parquet" => Ok(Self::Parquet),
-            v => Err(Unsupported(format!("Unsupported base file format {}", v))),
+            "orc" => Err(UnsupportedValue(s.to_string())),
+            v => Err(InvalidValue(v.to_string())),
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use std::str::FromStr;
-
-    use crate::config::table::{BaseFileFormatValue, TableTypeValue};
+    use super::*;
 
     #[test]
     fn create_table_type() {
@@ -297,10 +276,22 @@ mod tests {
             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());
+        assert!(matches!(
+            TableTypeValue::from_str("").unwrap_err(),
+            InvalidValue(_)
+        ));
+        assert!(matches!(
+            TableTypeValue::from_str("copyonwrite").unwrap_err(),
+            InvalidValue(_)
+        ));
+        assert!(matches!(
+            TableTypeValue::from_str("MERGEONREAD").unwrap_err(),
+            InvalidValue(_)
+        ));
+        assert!(matches!(
+            TableTypeValue::from_str("foo").unwrap_err(),
+            InvalidValue(_)
+        ));
     }
 
     #[test]
@@ -313,10 +304,13 @@ mod tests {
             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."
-        );
+        assert!(matches!(
+            BaseFileFormatValue::from_str("").unwrap_err(),
+            InvalidValue(_)
+        ));
+        assert!(matches!(
+            BaseFileFormatValue::from_str("orc").unwrap_err(),
+            UnsupportedValue(_)
+        ));
     }
 }
diff --git a/crates/core/src/config/utils.rs b/crates/core/src/config/util.rs
similarity index 95%
rename from crates/core/src/config/utils.rs
rename to crates/core/src/config/util.rs
index 85784ec4..33921c65 100644
--- a/crates/core/src/config/utils.rs
+++ b/crates/core/src/config/util.rs
@@ -18,12 +18,12 @@
  */
 //! Config utilities.
 
+use crate::config::error::ConfigError::ParseLine;
+use crate::config::Result;
 use bytes::Bytes;
 use std::collections::HashMap;
 use std::io::{BufRead, BufReader, Cursor};
 
-use crate::{CoreError, Result};
-
 /// Returns an empty iterator to represent an empty set of options.
 pub fn empty_options<'a>() -> std::iter::Empty<(&'a str, &'a str)> {
     std::iter::empty::<(&str, &str)>()
@@ -58,7 +58,7 @@ pub fn parse_data_for_options(data: &Bytes, split_chars: &str) -> Result<HashMap
     let mut options = HashMap::new();
 
     for line in lines {
-        let line = line.map_err(|e| CoreError::Internal(format!("Failed to read line {e}")))?;
+        let line = line.map_err(|e| ParseLine(format!("Failed to parse config data: {:?}", e)))?;
         let trimmed_line = line.trim();
         if trimmed_line.is_empty() || trimmed_line.starts_with('#') {
             continue;
@@ -66,9 +66,10 @@ pub fn parse_data_for_options(data: &Bytes, split_chars: &str) -> Result<HashMap
         let mut parts = trimmed_line.splitn(2, |c| split_chars.contains(c));
         let key = parts
             .next()
-            .ok_or(CoreError::Internal(
-                "Missing key in config line".to_string(),
-            ))?
+            .ok_or(ParseLine(format!(
+                "Missing key in config line: {}",
+                trimmed_line
+            )))?
             .trim()
             .to_owned();
         let value = parts.next().unwrap_or("").trim().to_owned();
diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs
new file mode 100644
index 00000000..6d4f0b45
--- /dev/null
+++ b/crates/core/src/error.rs
@@ -0,0 +1,56 @@
+/*
+ * 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::config::error::ConfigError;
+use crate::storage::error::StorageError;
+use thiserror::Error;
+
+pub type Result<T, E = CoreError> = std::result::Result<T, E>;
+
+#[derive(Error, Debug)]
+pub enum CoreError {
+    #[error(transparent)]
+    ArrowError(#[from] arrow::error::ArrowError),
+
+    #[error("Config error: {0}")]
+    Config(#[from] ConfigError),
+
+    #[error("File group error: {0}")]
+    FileGroup(String),
+
+    #[error("{0}: {1:?}")]
+    ReadFileSliceError(String, StorageError),
+
+    #[error("{0}")]
+    InvalidPartitionPath(String),
+
+    #[error(transparent)]
+    ParquetError(#[from] parquet::errors::ParquetError),
+
+    #[error("Storage error: {0}")]
+    Storage(#[from] StorageError),
+
+    #[error("Timeline error: {0}")]
+    Timeline(String),
+
+    #[error("{0}")]
+    Unsupported(String),
+
+    #[error(transparent)]
+    Utf8Error(#[from] std::str::Utf8Error),
+}
diff --git a/crates/core/src/file_group/mod.rs b/crates/core/src/file_group/mod.rs
index 2fde1bfc..58fb1446 100644
--- a/crates/core/src/file_group/mod.rs
+++ b/crates/core/src/file_group/mod.rs
@@ -22,17 +22,17 @@
 
 pub mod reader;
 
+use crate::error::CoreError;
+use crate::storage::file_info::FileInfo;
+use crate::storage::file_stats::FileStats;
+use crate::storage::Storage;
+use crate::Result;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::fmt::Formatter;
 use std::hash::{Hash, Hasher};
 use std::path::PathBuf;
 
-use crate::storage::file_info::FileInfo;
-use crate::storage::file_stats::FileStats;
-use crate::storage::Storage;
-use crate::{CoreError::Internal, Result};
-
 /// Represents common metadata about a Hudi Base File.
 #[derive(Clone, Debug)]
 pub struct BaseFile {
@@ -52,10 +52,16 @@ impl BaseFile {
         let err_msg = format!("Failed to parse file name '{}' for base file.", file_name);
         let (name, _) = file_name
             .rsplit_once('.')
-            .ok_or(Internal(err_msg.clone()))?;
+            .ok_or(CoreError::FileGroup(err_msg.clone()))?;
         let parts: Vec<&str> = name.split('_').collect();
-        let file_group_id = parts.first().ok_or(Internal(err_msg.clone()))?.to_string();
-        let commit_time = parts.get(2).ok_or(Internal(err_msg.clone()))?.to_string();
+        let file_group_id = parts
+            .first()
+            .ok_or(CoreError::FileGroup(err_msg.clone()))?
+            .to_string();
+        let commit_time = parts
+            .get(2)
+            .ok_or(CoreError::FileGroup(err_msg.clone()))?
+            .to_string();
         Ok((file_group_id, commit_time))
     }
 
@@ -190,7 +196,7 @@ impl FileGroup {
     pub fn add_base_file(&mut self, base_file: BaseFile) -> Result<&Self> {
         let commit_time = base_file.commit_time.as_str();
         if self.file_slices.contains_key(commit_time) {
-            Err(Internal(format!(
+            Err(CoreError::FileGroup(format!(
                 "Commit time {0} is already present in File Group {1}",
                 commit_time.to_owned(),
                 self.id,
@@ -243,6 +249,18 @@ mod tests {
         assert_eq!(base_file.commit_time, "20240402144910683");
     }
 
+    #[test]
+    fn create_a_base_file_returns_error() {
+        let result = BaseFile::from_file_name("no_file_extension");
+        assert!(matches!(result.unwrap_err(), CoreError::FileGroup(_)));
+
+        let result = BaseFile::from_file_name(".parquet");
+        assert!(matches!(result.unwrap_err(), CoreError::FileGroup(_)));
+
+        let result = BaseFile::from_file_name("no-valid-delimiter.parquet");
+        assert!(matches!(result.unwrap_err(), CoreError::FileGroup(_)));
+    }
+
     #[test]
     fn load_a_valid_file_group() {
         let mut fg = FileGroup::new("5a226868-2934-4f84-a16f-55124630c68d-0".to_owned(), None);
@@ -277,7 +295,7 @@ mod tests {
             "5a226868-2934-4f84-a16f-55124630c68d-0_2-10-0_20240402144910683.parquet",
         );
         assert!(res2.is_err());
-        assert_eq!(res2.unwrap_err().to_string(), "Commit time 20240402144910683 is already present in File Group 5a226868-2934-4f84-a16f-55124630c68d-0");
+        assert_eq!(res2.unwrap_err().to_string(), "File group error: Commit time 20240402144910683 is already present in File Group 5a226868-2934-4f84-a16f-55124630c68d-0");
     }
 
     #[test]
diff --git a/crates/core/src/file_group/reader.rs b/crates/core/src/file_group/reader.rs
index eeb283ae..165ed9ce 100644
--- a/crates/core/src/file_group/reader.rs
+++ b/crates/core/src/file_group/reader.rs
@@ -17,12 +17,14 @@
  * under the License.
  */
 use crate::config::table::HudiTableConfig;
-use crate::config::utils::split_hudi_options_from_others;
+use crate::config::util::split_hudi_options_from_others;
 use crate::config::HudiConfigs;
+use crate::error::CoreError::ReadFileSliceError;
 use crate::file_group::FileSlice;
 use crate::storage::Storage;
 use crate::Result;
 use arrow_array::RecordBatch;
+use futures::TryFutureExt;
 use std::sync::Arc;
 
 /// File group reader handles all read operations against a file group.
@@ -58,7 +60,15 @@ impl FileGroupReader {
         &self,
         relative_path: &str,
     ) -> Result<RecordBatch> {
-        self.storage.get_parquet_file_data(relative_path).await
+        self.storage
+            .get_parquet_file_data(relative_path)
+            .map_err(|e| {
+                ReadFileSliceError(
+                    format!("Failed to read file slice at path '{}'", relative_path),
+                    e,
+                )
+            })
+            .await
     }
 
     pub async fn read_file_slice(&self, file_slice: &FileSlice) -> Result<RecordBatch> {
@@ -91,4 +101,16 @@ mod tests {
             .contains(HudiTableConfig::BasePath));
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_read_file_slice_returns_error() {
+        let storage =
+            Storage::new_with_base_url(Url::parse("file:///non-existent-path/table").unwrap())
+                .unwrap();
+        let reader = FileGroupReader::new(storage);
+        let result = reader
+            .read_file_slice_by_base_file_path("non_existent_file")
+            .await;
+        assert!(matches!(result.unwrap_err(), ReadFileSliceError(_, _)));
+    }
 }
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index 6485c1df..42079a17 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -44,53 +44,10 @@
 //! ```
 
 pub mod config;
+pub mod error;
 pub mod file_group;
 pub mod storage;
 pub mod table;
 pub mod util;
 
-use thiserror::Error;
-
-#[derive(Error, Debug)]
-pub enum CoreError {
-    #[error("Config '{0}' not found")]
-    ConfigNotFound(String),
-
-    #[error("Invalid config item '{item}', {source:?}")]
-    InvalidConfig {
-        item: &'static str,
-        source: Box<dyn std::error::Error + Sync + Send + 'static>,
-    },
-
-    #[error("Parse url '{url}' failed, {source}")]
-    UrlParse {
-        url: String,
-        source: url::ParseError,
-    },
-
-    #[error("Invalid file path '{name}', {detail}")]
-    InvalidPath { name: String, detail: String },
-
-    #[error("{0}")]
-    Unsupported(String),
-
-    #[error("{0}")]
-    Internal(String),
-
-    #[error(transparent)]
-    Utf8Error(#[from] std::str::Utf8Error),
-
-    #[error(transparent)]
-    ObjectStore(#[from] object_store::Error),
-
-    #[error(transparent)]
-    ObjectStorePath(#[from] object_store::path::Error),
-
-    #[error(transparent)]
-    Parquet(#[from] parquet::errors::ParquetError),
-
-    #[error(transparent)]
-    Arrow(#[from] arrow::error::ArrowError),
-}
-
-type Result<T, E = CoreError> = std::result::Result<T, E>;
+use error::Result;
diff --git a/crates/core/src/storage/error.rs b/crates/core/src/storage/error.rs
new file mode 100644
index 00000000..09c749dc
--- /dev/null
+++ b/crates/core/src/storage/error.rs
@@ -0,0 +1,45 @@
+/*
+ * 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 thiserror::Error;
+
+pub type Result<T, E = StorageError> = std::result::Result<T, E>;
+
+#[derive(Error, Debug)]
+pub enum StorageError {
+    #[error(transparent)]
+    ArrowError(#[from] arrow::error::ArrowError),
+
+    #[error("Failed to create storage: {0}")]
+    Creation(String),
+
+    #[error("Invalid path: {0}")]
+    InvalidPath(String),
+
+    #[error(transparent)]
+    ObjectStoreError(#[from] object_store::Error),
+
+    #[error(transparent)]
+    ObjectStorePathError(#[from] object_store::path::Error),
+
+    #[error(transparent)]
+    ParquetError(#[from] parquet::errors::ParquetError),
+
+    #[error(transparent)]
+    UrlParseError(#[from] url::ParseError),
+}
diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs
index 0a9f0e15..7f310115 100644
--- a/crates/core/src/storage/mod.rs
+++ b/crates/core/src/storage/mod.rs
@@ -36,13 +36,15 @@ use url::Url;
 
 use crate::config::table::HudiTableConfig;
 use crate::config::HudiConfigs;
+use crate::storage::error::Result;
+use crate::storage::error::StorageError::{Creation, InvalidPath};
 use crate::storage::file_info::FileInfo;
-use crate::storage::utils::join_url_segments;
-use crate::{CoreError, Result};
+use crate::storage::util::join_url_segments;
 
+pub mod error;
 pub mod file_info;
 pub mod file_stats;
-pub mod utils;
+pub mod util;
 
 #[allow(dead_code)]
 #[derive(Clone, Debug)]
@@ -60,14 +62,15 @@ impl Storage {
         options: Arc<HashMap<String, String>>,
         hudi_configs: Arc<HudiConfigs>,
     ) -> Result<Arc<Storage>> {
-        if !hudi_configs.contains(HudiTableConfig::BasePath) {
-            return Err(CoreError::Internal(format!(
-                "Failed to create storage: {} is required.",
-                HudiTableConfig::BasePath.as_ref()
-            )));
-        }
-
-        let base_url = hudi_configs.get(HudiTableConfig::BasePath)?.to_url()?;
+        let base_url = match hudi_configs.try_get(HudiTableConfig::BasePath) {
+            Some(v) => v.to_url()?,
+            None => {
+                return Err(Creation(format!(
+                    "{} is required.",
+                    HudiTableConfig::BasePath.as_ref()
+                )))
+            }
+        };
 
         match parse_url_opts(&base_url, options.as_ref()) {
             Ok((object_store, _)) => Ok(Arc::new(Storage {
@@ -76,7 +79,7 @@ impl Storage {
                 options,
                 hudi_configs,
             })),
-            Err(e) => Err(CoreError::ObjectStore(e)),
+            Err(e) => Err(Creation(format!("Failed to create storage: {}", e))),
         }
     }
 
@@ -109,10 +112,10 @@ impl Storage {
         let uri = obj_url.to_string();
         let name = obj_path
             .filename()
-            .ok_or(CoreError::InvalidPath {
-                name: obj_path.to_string(),
-                detail: "failed to get file name".to_string(),
-            })?
+            .ok_or(InvalidPath(format!(
+                "Failed to get file name from {:?}",
+                obj_path
+            )))?
             .to_string();
         Ok(FileInfo {
             uri,
@@ -176,10 +179,10 @@ impl Storage {
         for dir in dir_paths {
             dirs.push(
                 dir.filename()
-                    .ok_or(CoreError::InvalidPath {
-                        name: dir.to_string(),
-                        detail: "failed to get file name".to_string(),
-                    })?
+                    .ok_or(InvalidPath(format!(
+                        "Failed to get file name from: {:?}",
+                        dir
+                    )))?
                     .to_string(),
             )
         }
@@ -208,10 +211,10 @@ impl Storage {
             let name = obj_meta
                 .location
                 .filename()
-                .ok_or(CoreError::InvalidPath {
-                    name: obj_meta.location.to_string(),
-                    detail: "failed to get file name".to_string(),
-                })?
+                .ok_or(InvalidPath(format!(
+                    "Failed to get file name from {:?}",
+                    obj_meta.location
+                )))?
                 .to_string();
             let uri = join_url_segments(&prefix_url, &[&name])?.to_string();
             file_info.push(FileInfo {
@@ -246,10 +249,10 @@ pub async fn get_leaf_dirs(storage: &Storage, subdir: Option<&str>) -> Result<Ve
                 next_subdir.push(curr);
             }
             next_subdir.push(child_dir);
-            let next_subdir = next_subdir.to_str().ok_or(CoreError::InvalidPath {
-                name: format!("{:?}", next_subdir),
-                detail: "failed to convert path".to_string(),
-            })?;
+            let next_subdir = next_subdir.to_str().ok_or(InvalidPath(format!(
+                "Failed to convert path: {:?}",
+                next_subdir
+            )))?;
             let curr_leaf_dir = get_leaf_dirs(storage, Some(next_subdir)).await?;
             leaf_dirs.extend(curr_leaf_dir);
         }
@@ -264,12 +267,6 @@ mod tests {
     use std::fs::canonicalize;
     use std::path::Path;
 
-    use crate::storage::file_info::FileInfo;
-    use crate::storage::utils::join_url_segments;
-    use crate::storage::{get_leaf_dirs, Storage};
-    use object_store::path::Path as ObjPath;
-    use url::Url;
-
     #[test]
     fn test_storage_new_error_no_base_path() {
         let options = Arc::new(HashMap::new());
@@ -299,7 +296,7 @@ mod tests {
             result.is_err(),
             "Should return error when no base path is invalid."
         );
-        assert!(matches!(result.unwrap_err(), CoreError::ObjectStore(_)));
+        assert!(matches!(result.unwrap_err(), Creation(_)));
     }
 
     #[tokio::test]
diff --git a/crates/core/src/storage/utils.rs b/crates/core/src/storage/util.rs
similarity index 82%
rename from crates/core/src/storage/utils.rs
rename to crates/core/src/storage/util.rs
index 10198a15..e3f32b95 100644
--- a/crates/core/src/storage/utils.rs
+++ b/crates/core/src/storage/util.rs
@@ -19,12 +19,10 @@
 
 //! Utility functions for storage.
 use std::path::Path;
-use url::{ParseError, Url};
+use url::Url;
 
-use crate::{
-    CoreError::{InvalidPath, UrlParse},
-    Result,
-};
+use crate::storage::error::StorageError::{InvalidPath, UrlParseError};
+use crate::storage::Result;
 
 /// Splits a filename into a stem and an extension.
 pub fn split_filename(filename: &str) -> Result<(String, String)> {
@@ -33,10 +31,7 @@ pub fn split_filename(filename: &str) -> Result<(String, String)> {
     let stem = path
         .file_stem()
         .and_then(|s| s.to_str())
-        .ok_or_else(|| InvalidPath {
-            name: filename.to_string(),
-            detail: "no file stem found".to_string(),
-        })?
+        .ok_or(InvalidPath(format!("No file stem found in {}", filename)))?
         .to_string();
 
     let extension = path
@@ -52,19 +47,12 @@ pub fn split_filename(filename: &str) -> Result<(String, String)> {
 pub fn parse_uri(uri: &str) -> Result<Url> {
     let mut url = match Url::parse(uri) {
         Ok(url) => url,
-        Err(source) => Url::from_directory_path(uri).map_err(|_| UrlParse {
-            url: uri.to_string(),
-            source,
-        })?,
+        Err(e) => Url::from_directory_path(uri).map_err(|_| UrlParseError(e))?,
     };
 
     if url.path().ends_with('/') {
-        url.path_segments_mut()
-            .map_err(|_| InvalidPath {
-                name: uri.to_string(),
-                detail: "parse uri failed".to_string(),
-            })?
-            .pop();
+        let err = InvalidPath(format!("Url {:?} cannot be a base", url));
+        url.path_segments_mut().map_err(|_| err)?.pop();
     }
 
     Ok(url)
@@ -85,12 +73,8 @@ pub fn join_url_segments(base_url: &Url, segments: &[&str]) -> Result<Url> {
 
     for &seg in segments {
         let segs: Vec<_> = seg.split('/').filter(|&s| !s.is_empty()).collect();
-        url.path_segments_mut()
-            .map_err(|_| UrlParse {
-                url: base_url.to_string(),
-                source: ParseError::RelativeUrlWithoutBase,
-            })?
-            .extend(segs);
+        let err = InvalidPath(format!("Url {:?} cannot be a base", url));
+        url.path_segments_mut().map_err(|_| err)?.extend(segs);
     }
 
     Ok(url)
@@ -98,12 +82,9 @@ pub fn join_url_segments(base_url: &Url, segments: &[&str]) -> Result<Url> {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use std::str::FromStr;
 
-    use url::Url;
-
-    use crate::storage::utils::{join_url_segments, parse_uri};
-
     #[test]
     fn parse_valid_uri_in_various_forms() {
         let urls = vec![
diff --git a/crates/core/src/table/builder.rs b/crates/core/src/table/builder.rs
index eca7d2b5..20df88e2 100644
--- a/crates/core/src/table/builder.rs
+++ b/crates/core/src/table/builder.rs
@@ -31,13 +31,14 @@ use crate::config::read::HudiReadConfig;
 use crate::config::table::HudiTableConfig::{DropsPartitionFields, TableType, TableVersion};
 use crate::config::table::TableTypeValue::CopyOnWrite;
 use crate::config::table::{HudiTableConfig, TableTypeValue};
-use crate::config::utils::{parse_data_for_options, split_hudi_options_from_others};
+use crate::config::util::{parse_data_for_options, split_hudi_options_from_others};
 use crate::config::{HudiConfigs, HUDI_CONF_DIR};
+use crate::error::CoreError;
 use crate::storage::Storage;
 use crate::table::fs_view::FileSystemView;
 use crate::table::timeline::Timeline;
 use crate::table::Table;
-use crate::{CoreError, Result};
+use crate::Result;
 
 /// Builder for creating a [Table] instance.
 #[derive(Debug, Clone)]
@@ -185,7 +186,9 @@ impl TableBuilder {
         let hudi_options = &mut self.hudi_options;
         Self::imbue_table_properties(hudi_options, storage.clone()).await?;
 
-        // TODO load Hudi configs from env vars here before loading global configs
+        // TODO support imbuing Hudi options from env vars HOODIE_ENV.*
+        // (see https://hudi.apache.org/docs/next/s3_hoodie)
+        // before loading global configs
 
         Self::imbue_global_hudi_configs_if_absent(hudi_options, storage.clone()).await
     }
diff --git a/crates/core/src/table/fs_view.rs b/crates/core/src/table/fs_view.rs
index ee68eecd..092b352a 100644
--- a/crates/core/src/table/fs_view.rs
+++ b/crates/core/src/table/fs_view.rs
@@ -25,8 +25,9 @@ use crate::file_group::{BaseFile, FileGroup, FileSlice};
 use crate::storage::file_info::FileInfo;
 use crate::storage::{get_leaf_dirs, Storage};
 
+use crate::error::CoreError;
 use crate::table::partition::PartitionPruner;
-use crate::{CoreError, Result};
+use crate::Result;
 use dashmap::DashMap;
 use futures::stream::{self, StreamExt, TryStreamExt};
 
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 4836d1c9..da32a109 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -60,7 +60,7 @@
 //! ```rust
 //! use url::Url;
 //! use hudi_core::table::Table;
-//! use hudi_core::storage::utils::parse_uri;
+//! use hudi_core::storage::util::parse_uri;
 //!
 //! pub async fn test() {
 //!     let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
@@ -95,13 +95,14 @@ use crate::config::read::HudiReadConfig::AsOfTimestamp;
 use crate::config::table::HudiTableConfig;
 use crate::config::table::HudiTableConfig::PartitionFields;
 use crate::config::HudiConfigs;
+use crate::error::CoreError;
 use crate::file_group::reader::FileGroupReader;
 use crate::file_group::FileSlice;
 use crate::table::builder::TableBuilder;
 use crate::table::fs_view::FileSystemView;
 use crate::table::partition::PartitionPruner;
 use crate::table::timeline::Timeline;
-use crate::{CoreError, Result};
+use crate::Result;
 
 pub mod builder;
 mod fs_view;
@@ -137,7 +138,10 @@ impl Table {
     }
 
     pub fn base_url(&self) -> Result<Url> {
-        self.hudi_configs.get(HudiTableConfig::BasePath)?.to_url()
+        self.hudi_configs
+            .get(HudiTableConfig::BasePath)?
+            .to_url()
+            .map_err(CoreError::from)
     }
 
     pub fn hudi_options(&self) -> HashMap<String, String> {
@@ -161,6 +165,7 @@ impl Table {
             .register_object_store(runtime_env.clone());
     }
 
+    /// Get the latest [Schema] of the table.
     pub async fn get_schema(&self) -> Result<Schema> {
         self.timeline.get_latest_schema().await
     }
@@ -259,19 +264,10 @@ impl Table {
         filters: &[(&str, &str, &str)],
     ) -> Result<Vec<RecordBatch>> {
         let file_slices = self.get_file_slices_as_of(timestamp, filters).await?;
-        let mut batches = Vec::new();
         let fg_reader = self.create_file_group_reader();
-        for f in file_slices {
-            match fg_reader.read_file_slice(&f).await {
-                Ok(batch) => batches.push(batch),
-                Err(e) => {
-                    return Err(CoreError::Internal(format!(
-                        "Failed to read file slice {:?} - {}",
-                        f, e
-                    )))
-                }
-            }
-        }
+        let batches =
+            futures::future::try_join_all(file_slices.iter().map(|f| fg_reader.read_file_slice(f)))
+                .await?;
         Ok(batches)
     }
 
@@ -294,16 +290,14 @@ impl Table {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use arrow_array::StringArray;
-    use std::collections::HashSet;
     use std::fs::canonicalize;
     use std::path::PathBuf;
     use std::{env, panic};
-    use url::Url;
 
     use hudi_tests::{assert_not, TestTable};
 
-    use crate::config::read::HudiReadConfig::AsOfTimestamp;
     use crate::config::table::HudiTableConfig::{
         BaseFileFormat, Checksum, DatabaseName, DropsPartitionFields, IsHiveStylePartitioning,
         IsPartitionPathUrlencoded, KeyGeneratorClass, PartitionFields, PopulatesMetaFields,
@@ -311,9 +305,8 @@ mod tests {
         TimelineLayoutVersion,
     };
     use crate::config::HUDI_CONF_DIR;
-    use crate::storage::utils::join_url_segments;
+    use crate::storage::util::join_url_segments;
     use crate::storage::Storage;
-    use crate::table::Table;
 
     /// Test helper to create a new `Table` instance without validating the configuration.
     ///
diff --git a/crates/core/src/table/partition.rs b/crates/core/src/table/partition.rs
index 8f035213..ca41f731 100644
--- a/crates/core/src/table/partition.rs
+++ b/crates/core/src/table/partition.rs
@@ -18,7 +18,9 @@
  */
 use crate::config::table::HudiTableConfig;
 use crate::config::HudiConfigs;
-use crate::{CoreError, Result};
+use crate::error::CoreError;
+use crate::error::CoreError::{InvalidPartitionPath, Unsupported};
+use crate::Result;
 use arrow_array::{ArrayRef, Scalar, StringArray};
 use arrow_cast::{cast_with_options, CastOptions};
 use arrow_ord::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
@@ -119,7 +121,7 @@ impl PartitionPruner {
         let parts: Vec<&str> = partition_path.split('/').collect();
 
         if parts.len() != self.schema.fields().len() {
-            return Err(CoreError::Internal(format!(
+            return Err(InvalidPartitionPath(format!(
                 "Partition path should have {} part(s) but got {}",
                 self.schema.fields().len(),
                 parts.len()
@@ -132,14 +134,11 @@ impl PartitionPruner {
             .zip(parts)
             .map(|(field, part)| {
                 let value = if self.is_hive_style {
-                    let (name, value) = part.split_once('=').ok_or_else(|| {
-                        CoreError::Internal(format!(
-                            "Partition path should be hive-style but got {}",
-                            part
-                        ))
-                    })?;
+                    let (name, value) = part.split_once('=').ok_or(InvalidPartitionPath(
+                        format!("Partition path should be hive-style but got {}", part),
+                    ))?;
                     if name != field.name() {
-                        return Err(CoreError::Internal(format!(
+                        return Err(InvalidPartitionPath(format!(
                             "Partition path should contain {} but got {}",
                             field.name(),
                             name
@@ -179,13 +178,13 @@ impl Operator {
 }
 
 impl FromStr for Operator {
-    type Err = crate::CoreError;
+    type Err = CoreError;
 
-    fn from_str(s: &str) -> Result<Self> {
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
         Operator::TOKEN_OP_PAIRS
             .iter()
             .find_map(|&(token, op)| if token == s { Some(op) } else { None })
-            .ok_or_else(|| CoreError::Internal(format!("Unsupported operator: {}", s)))
+            .ok_or(Unsupported(format!("Unsupported operator: {}", s)))
     }
 }
 
@@ -198,9 +197,11 @@ pub struct PartitionFilter {
 }
 
 impl TryFrom<((&str, &str, &str), &Schema)> for PartitionFilter {
-    type Error = crate::CoreError;
+    type Error = CoreError;
 
-    fn try_from((filter, partition_schema): ((&str, &str, &str), &Schema)) -> Result<Self> {
+    fn try_from(
+        (filter, partition_schema): ((&str, &str, &str), &Schema),
+    ) -> Result<Self, Self::Error> {
         let (field_name, operator_str, value_str) = filter;
 
         let field: &Field = partition_schema.field_with_name(field_name)?;
@@ -435,6 +436,13 @@ mod tests {
         let configs = create_hudi_configs(true, false);
         let pruner = PartitionPruner::new(&[], &schema, &configs).unwrap();
 
-        assert!(pruner.parse_segments("invalid/path").is_err());
+        let result = pruner.parse_segments("date=2023-02-01/category=A/count=10/extra");
+        assert!(matches!(result.unwrap_err(), InvalidPartitionPath(_)));
+
+        let result = pruner.parse_segments("date=2023-02-01/category=A/10");
+        assert!(matches!(result.unwrap_err(), InvalidPartitionPath(_)));
+
+        let result = pruner.parse_segments("date=2023-02-01/category=A/non_exist_field=10");
+        assert!(matches!(result.unwrap_err(), InvalidPartitionPath(_)));
     }
 }
diff --git a/crates/core/src/table/timeline.rs b/crates/core/src/table/timeline.rs
index d147abf4..d48aa556 100644
--- a/crates/core/src/table/timeline.rs
+++ b/crates/core/src/table/timeline.rs
@@ -28,10 +28,12 @@ use parquet::arrow::parquet_to_arrow_schema;
 use serde_json::{Map, Value};
 
 use crate::config::HudiConfigs;
+use crate::error::CoreError;
 use crate::file_group::FileGroup;
-use crate::storage::utils::split_filename;
+use crate::storage::error::StorageError;
+use crate::storage::util::split_filename;
 use crate::storage::Storage;
-use crate::{CoreError, Result};
+use crate::Result;
 
 /// The [State] of an [Instant] represents the status of the action performed on the table.
 #[allow(dead_code)]
@@ -80,10 +82,11 @@ impl Instant {
         commit_file_path.push(self.file_name());
         commit_file_path
             .to_str()
-            .ok_or(CoreError::Internal(format!(
+            .ok_or(StorageError::InvalidPath(format!(
                 "Failed to get file path for {:?}",
                 self
             )))
+            .map_err(CoreError::Storage)
             .map(|s| s.to_string())
     }
 
@@ -143,11 +146,15 @@ impl Timeline {
             .storage
             .get_file_data(instant.relative_path()?.as_str())
             .await?;
+        let err_msg = format!("Failed to get commit metadata for {:?}", instant);
         let json: Value = serde_json::from_slice(&bytes)
-            .map_err(|e| CoreError::Internal(format!("Invalid instant {:?}", e)))?;
+            .map_err(|e| CoreError::Timeline(format!("{}: {:?}", err_msg, e)))?;
         let commit_metadata = json
             .as_object()
-            .ok_or_else(|| CoreError::Internal("Expected JSON object".to_string()))?
+            .ok_or(CoreError::Timeline(format!(
+                "{}: not a JSON object",
+                err_msg
+            )))?
             .clone();
         Ok(commit_metadata)
     }
@@ -178,7 +185,7 @@ impl Timeline {
                 None,
             )?)
         } else {
-            Err(CoreError::Internal(
+            Err(CoreError::Timeline(
                 "Failed to resolve the latest schema: no file path found".to_string(),
             ))
         }
@@ -216,6 +223,7 @@ impl Timeline {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use std::collections::HashMap;
     use std::fs::canonicalize;
     use std::path::Path;
@@ -226,8 +234,6 @@ mod tests {
     use hudi_tests::TestTable;
 
     use crate::config::table::HudiTableConfig;
-    use crate::config::HudiConfigs;
-    use crate::table::timeline::{Instant, State, Timeline};
 
     async fn create_test_timeline(base_url: Url) -> Timeline {
         Timeline::new(
@@ -254,7 +260,7 @@ mod tests {
         assert!(table_schema.is_err());
         assert_eq!(
             table_schema.err().unwrap().to_string(),
-            "Failed to resolve the latest schema: no file path found"
+            "Timeline error: Failed to resolve the latest schema: no file path found"
         )
     }
 
@@ -281,4 +287,41 @@ mod tests {
             ]
         )
     }
+
+    #[tokio::test]
+    async fn get_commit_metadata_returns_error() {
+        let base_url = Url::from_file_path(
+            canonicalize(Path::new(
+                "tests/data/timeline/commits_with_invalid_content",
+            ))
+            .unwrap(),
+        )
+        .unwrap();
+        let timeline = create_test_timeline(base_url).await;
+        let instant = Instant {
+            state: State::Completed,
+            action: "commit".to_owned(),
+            timestamp: "20240402123035233".to_owned(),
+        };
+
+        // Test error when reading empty commit metadata file
+        let result = timeline.get_commit_metadata(&instant).await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert!(matches!(err, CoreError::Timeline(_)));
+        assert!(err.to_string().contains("Failed to get commit metadata"));
+
+        let instant = Instant {
+            state: State::Completed,
+            action: "commit".to_owned(),
+            timestamp: "20240402144910683".to_owned(),
+        };
+
+        // Test error when reading a commit metadata file with invalid JSON
+        let result = timeline.get_commit_metadata(&instant).await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert!(matches!(err, CoreError::Timeline(_)));
+        assert!(err.to_string().contains("not a JSON object"));
+    }
 }
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 1bf17eaf..b5790b8e 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -40,8 +40,8 @@ use datafusion_expr::{CreateExternalTable, Expr, TableType};
 use datafusion_physical_expr::create_physical_expr;
 
 use hudi_core::config::read::HudiReadConfig::InputPartitions;
-use hudi_core::config::utils::empty_options;
-use hudi_core::storage::utils::{get_scheme_authority, parse_uri};
+use hudi_core::config::util::empty_options;
+use hudi_core::storage::util::{get_scheme_authority, parse_uri};
 use hudi_core::table::Table as HudiTable;
 
 /// Create a `HudiDataSource`.
diff --git a/python/src/internal.rs b/python/src/internal.rs
index 549a503b..a6a35524 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -26,13 +26,14 @@ use arrow::pyarrow::ToPyArrow;
 use pyo3::{pyclass, pyfunction, pymethods, PyErr, PyObject, PyResult, Python};
 use tokio::runtime::Runtime;
 
+use hudi::error::CoreError;
 use hudi::file_group::reader::FileGroupReader;
 use hudi::file_group::FileSlice;
+use hudi::storage::error::StorageError;
 use hudi::table::builder::TableBuilder;
 use hudi::table::Table;
 use hudi::util::convert_vec_to_slice;
 use hudi::util::vec_to_slice;
-use hudi::CoreError;
 use pyo3::create_exception;
 use pyo3::exceptions::PyException;
 
@@ -45,7 +46,7 @@ fn convert_to_py_err(err: CoreError) -> PyErr {
 
 #[derive(thiserror::Error, Debug)]
 pub enum PythonError {
-    #[error("Error in Hudi core")]
+    #[error("Error in Hudi core: {0}")]
     HudiCore(#[from] CoreError),
 }
 
@@ -110,17 +111,17 @@ pub struct HudiFileSlice {
 #[pymethods]
 impl HudiFileSlice {
     fn base_file_relative_path(&self) -> PyResult<String> {
-        Ok(PathBuf::from(&self.partition_path)
+        let path = PathBuf::from(&self.partition_path)
             .join(&self.base_file_name)
             .to_str()
             .map(String::from)
-            .ok_or_else(|| {
-                CoreError::Internal(format!(
-                    "Failed to get base file relative path for file slice: {:?}",
-                    self
-                ))
-            })
-            .map_err(PythonError::from)?)
+            .ok_or(StorageError::InvalidPath(format!(
+                "Failed to get base file relative path for file slice: {:?}",
+                self
+            )))
+            .map_err(CoreError::from)
+            .map_err(PythonError::from)?;
+        Ok(path)
     }
 }
 


Test patch:
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.commit b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.commit
new file mode 100644
index 00000000..e69de29b
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.commit.requested b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.commit.requested
new file mode 100644
index 00000000..e69de29b
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.inflight b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402123035233.inflight
new file mode 100644
index 00000000..e69de29b
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.commit b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.commit
new file mode 100644
index 00000000..cb64946b
--- /dev/null
+++ b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.commit
@@ -0,0 +1,7 @@
+[
+    "should",
+    "not",
+    "be",
+    "an",
+    "array"
+]
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.commit.requested b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.commit.requested
new file mode 100644
index 00000000..e69de29b
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.inflight b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/20240402144910683.inflight
new file mode 100644
index 00000000..e69de29b
diff --git a/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/hoodie.properties b/crates/core/tests/data/timeline/commits_with_invalid_content/.hoodie/hoodie.properties
new file mode 100644
index 00000000..e69de29b