From da6afdafc1b14324ac3077af2429134a9b17f006 Mon Sep 17 00:00:00 2001
From: codedump <lichuang1982@gmail.com>
Date: Fri, 2 Jan 2026 17:56:23 +0800
Subject: [PATCH] feat: refactor use of Error::io

---
 java/lance-jni/src/ffi.rs                     |  8 ++--
 java/lance-jni/src/sql.rs                     |  2 +-
 python/src/dataset.rs                         | 21 +++++-----
 rust/lance-core/src/error.rs                  | 20 ++++++++++
 rust/lance-datafusion/src/projection.rs       |  4 +-
 rust/lance-datafusion/src/sql.rs              |  5 ++-
 .../src/previous/format/metadata.rs           |  2 +-
 rust/lance-file/src/previous/reader.rs        |  6 +--
 rust/lance-file/src/previous/writer/mod.rs    |  8 ++--
 rust/lance-file/src/reader.rs                 |  4 +-
 .../src/scalar/inverted/tokenizer/jieba.rs    | 17 ++++----
 .../src/scalar/inverted/tokenizer/lindera.rs  |  4 +-
 rust/lance-index/src/vector/utils.rs          |  4 +-
 rust/lance-io/src/encodings/binary.rs         |  4 +-
 rust/lance-io/src/encodings/plain.rs          |  2 +-
 rust/lance-io/src/utils.rs                    |  2 +-
 rust/lance-table/src/format.rs                |  2 +-
 rust/lance-table/src/format/index.rs          |  2 +-
 rust/lance-table/src/format/manifest.rs       |  2 +-
 rust/lance-table/src/io/commit/dynamodb.rs    | 40 +++++++++----------
 rust/lance-table/src/io/manifest.rs           | 10 +++--
 rust/lance/src/dataset/blob.rs                | 24 +++++------
 rust/lance/src/dataset/fragment.rs            | 31 +++++++-------
 rust/lance/src/dataset/hash_joiner.rs         | 22 +++++++---
 rust/lance/src/dataset/scanner.rs             |  4 +-
 rust/lance/src/dataset/transaction.rs         |  4 +-
 rust/lance/src/dataset/updater.rs             |  4 +-
 rust/lance/src/index/vector/builder.rs        | 12 ++++--
 28 files changed, 151 insertions(+), 119 deletions(-)

diff --git a/java/lance-jni/src/ffi.rs b/java/lance-jni/src/ffi.rs
index 371e44563ed..d1f656873d8 100644
--- a/java/lance-jni/src/ffi.rs
+++ b/java/lance-jni/src/ffi.rs
@@ -296,7 +296,7 @@ impl JNIEnvExt for JNIEnv<'_> {
             .l()?;
         let operator_str = self.get_string_from_method(&operator_obj, "name")?;
         Operator::try_from(operator_str.as_str())
-            .map_err(|e| Error::io_error(format!("Invalid operator: {:?}", e)))
+            .map_err(|e| Error::input_error(format!("Invalid operator: {:?}", e)))
     }
 
     fn get_occur_from_method(&mut self, obj: &JObject) -> Result<Occur> {
@@ -310,7 +310,7 @@ impl JNIEnvExt for JNIEnv<'_> {
             .l()?;
         let occur_str = self.get_string_from_method(&occur_obj, "name")?;
         Occur::try_from(occur_str.as_str())
-            .map_err(|e| Error::io_error(format!("Invalid occur: {:?}", e)))
+            .map_err(|e| Error::input_error(format!("Invalid occur: {:?}", e)))
     }
 
     fn get_string_from_method(&mut self, obj: &JObject, method_name: &str) -> Result<String> {
@@ -388,7 +388,7 @@ impl JNIEnvExt for JNIEnv<'_> {
         self.get_optional_from_method(obj, method_name, |env, inner_jobj| {
             let inner_value = env.call_method(&inner_jobj, "intValue", "()I", &[])?.i()?;
             T::try_from(inner_value).map_err(|e| {
-                Error::io_error(format!("Failed to convert from i32 to rust type: {:?}", e))
+                Error::input_error(format!("Failed to convert from i32 to rust type: {:?}", e))
             })
         })
     }
@@ -421,7 +421,7 @@ impl JNIEnvExt for JNIEnv<'_> {
         self.get_optional_from_method(obj, method_name, |env, inner_jobj| {
             let inner_value = env.call_method(&inner_jobj, "longValue", "()J", &[])?.j()?;
             T::try_from(inner_value).map_err(|e| {
-                Error::io_error(format!("Failed to convert from i32 to rust type: {:?}", e))
+                Error::input_error(format!("Failed to convert from i32 to rust type: {:?}", e))
             })
         })
     }
diff --git a/java/lance-jni/src/sql.rs b/java/lance-jni/src/sql.rs
index e667f6c7128..07bc3d8e2fa 100644
--- a/java/lance-jni/src/sql.rs
+++ b/java/lance-jni/src/sql.rs
@@ -35,7 +35,7 @@ pub extern "system" fn Java_org_lance_SqlQuery_intoBatchRecords(
             with_row_addr,
             stream_addr,
         )
-        .map_err(|e| Error::io_error(e.to_string()))
+        .map_err(|e| Error::input_error(e.to_string()))
     )
 }
 
diff --git a/python/src/dataset.rs b/python/src/dataset.rs
index 80be4d98b35..b85c110f01d 100644
--- a/python/src/dataset.rs
+++ b/python/src/dataset.rs
@@ -344,11 +344,14 @@ pub fn transforms_from_python(transforms: &Bound<'_, PyAny>) -> PyResult<NewColu
                 let result = udf_obj
                     .call_method1(py, "_call", (py_batch,))
                     .map_err(|err| {
-                        lance::Error::io(format_python_error(err, py).unwrap(), location!())
+                        lance::Error::invalid_input(
+                            format_python_error(err, py).unwrap(),
+                            location!(),
+                        )
                     })?;
                 let result_batch: PyArrowType<RecordBatch> = result
                     .extract(py)
-                    .map_err(|err| lance::Error::io(err.to_string(), location!()))?;
+                    .map_err(|err| lance::Error::invalid_input(err.to_string(), location!()))?;
                 Ok(result_batch.0)
             })
         };
@@ -3419,7 +3422,7 @@ impl WriteFragmentProgress for PyWriteProgress {
             Ok(())
         })
         .map_err(|e| {
-            lance::Error::io(
+            lance::Error::invalid_input(
                 format!("Failed to call begin() on WriteFragmentProgress: {}", e),
                 location!(),
             )
@@ -3436,7 +3439,7 @@ impl WriteFragmentProgress for PyWriteProgress {
             Ok(())
         })
         .map_err(|e| {
-            lance::Error::io(
+            lance::Error::invalid_input(
                 format!("Failed to call complete() on WriteFragmentProgress: {}", e),
                 location!(),
             )
@@ -3480,7 +3483,7 @@ impl UDFCheckpointStore for PyBatchUDFCheckpointWrapper {
             Ok(batch.map(|b| b.0))
         })
         .map_err(|err: PyErr| {
-            lance_core::Error::io(
+            lance_core::Error::invalid_input(
                 format!("Failed to call get_batch() on UDFCheckpointer: {}", err),
                 location!(),
             )
@@ -3496,7 +3499,7 @@ impl UDFCheckpointStore for PyBatchUDFCheckpointWrapper {
             Ok(fragment)
         })
         .map_err(|err: PyErr| {
-            lance_core::Error::io(
+            lance_core::Error::invalid_input(
                 format!("Failed to call get_fragment() on UDFCheckpointer: {}", err),
                 location!(),
             )
@@ -3504,7 +3507,7 @@ impl UDFCheckpointStore for PyBatchUDFCheckpointWrapper {
         fragment_data
             .map(|data| {
                 serde_json::from_str(&data).map_err(|err| {
-                    lance::Error::io(
+                    lance_core::Error::invalid_input(
                         format!("Failed to deserialize fragment data: {}", err),
                         location!(),
                     )
@@ -3521,7 +3524,7 @@ impl UDFCheckpointStore for PyBatchUDFCheckpointWrapper {
             Ok(())
         })
         .map_err(|err: PyErr| {
-            lance_core::Error::io(
+            lance_core::Error::invalid_input(
                 format!("Failed to call insert_batch() on UDFCheckpointer: {}", err),
                 location!(),
             )
@@ -3541,7 +3544,7 @@ impl UDFCheckpointStore for PyBatchUDFCheckpointWrapper {
             Ok(())
         })
         .map_err(|err: PyErr| {
-            lance_core::Error::io(
+            lance_core::Error::invalid_input(
                 format!(
                     "Failed to call insert_fragment() on UDFCheckpointer: {}",
                     err
diff --git a/rust/lance-core/src/error.rs b/rust/lance-core/src/error.rs
index f80dbca4a7b..128789f9e73 100644
--- a/rust/lance-core/src/error.rs
+++ b/rust/lance-core/src/error.rs
@@ -164,6 +164,26 @@ impl Error {
             location,
         }
     }
+
+    pub fn not_found(uri: impl Into<String>) -> Self {
+        Self::NotFound {
+            uri: uri.into(),
+            location: std::panic::Location::caller().to_snafu_location(),
+        }
+    }
+
+    pub fn schema(message: impl Into<String>, location: Location) -> Self {
+        let message: String = message.into();
+        Self::Schema { message, location }
+    }
+
+    pub fn not_supported(message: impl Into<String>, location: Location) -> Self {
+        let message: String = message.into();
+        Self::NotSupported {
+            source: message.into(),
+            location,
+        }
+    }
 }
 
 pub trait LanceOptionExt<T> {
diff --git a/rust/lance-datafusion/src/projection.rs b/rust/lance-datafusion/src/projection.rs
index 1c504615ed5..f586ac4bb20 100644
--- a/rust/lance-datafusion/src/projection.rs
+++ b/rust/lance-datafusion/src/projection.rs
@@ -70,7 +70,7 @@ impl ProjectionBuilder {
 
     fn check_duplicate_column(&self, name: &str) -> Result<()> {
         if self.output.contains_key(name) {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!("Duplicate column name: {}", name),
                 location!(),
             ));
@@ -280,7 +280,7 @@ impl ProjectionPlan {
             } else {
                 // Regular data column - validate it exists in base schema
                 if base.schema().field(&field.name).is_none() {
-                    return Err(Error::io(
+                    return Err(Error::invalid_input(
                         format!("Column '{}' not found in schema", field.name),
                         location!(),
                     ));
diff --git a/rust/lance-datafusion/src/sql.rs b/rust/lance-datafusion/src/sql.rs
index 0f9e342c138..547badfdbfa 100644
--- a/rust/lance-datafusion/src/sql.rs
+++ b/rust/lance-datafusion/src/sql.rs
@@ -80,8 +80,9 @@ pub(crate) fn parse_sql_expr(expr: &str) -> Result<Expr> {
     } else {
         None
     };
-    let expr = selection
-        .ok_or_else(|| Error::io(format!("Expression is not valid: {expr}"), location!()))?;
+    let expr = selection.ok_or_else(|| {
+        Error::invalid_input(format!("Expression is not valid: {expr}"), location!())
+    })?;
     Ok(expr.clone())
 }
 
diff --git a/rust/lance-file/src/previous/format/metadata.rs b/rust/lance-file/src/previous/format/metadata.rs
index 025ed33d427..4724506218e 100644
--- a/rust/lance-file/src/previous/format/metadata.rs
+++ b/rust/lance-file/src/previous/format/metadata.rs
@@ -169,7 +169,7 @@ impl Metadata {
     // TODO: pub(crate)
     pub fn range_to_batches(&self, range: Range<usize>) -> Result<Vec<(i32, Range<usize>)>> {
         if range.end > *(self.batch_offsets.last().unwrap()) as usize {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "Range {:?} is out of bounds {}",
                     range,
diff --git a/rust/lance-file/src/previous/reader.rs b/rust/lance-file/src/previous/reader.rs
index 985906698b2..b6d70aafad4 100644
--- a/rust/lance-file/src/previous/reader.rs
+++ b/rust/lance-file/src/previous/reader.rs
@@ -516,7 +516,7 @@ fn get_page_info<'a>(
     batch_id: i32,
 ) -> Result<&'a PageInfo> {
     page_table.get(field.id, batch_id).ok_or_else(|| {
-        Error::io(
+        Error::invalid_input(
             format!(
                 "No page info found for field: {}, field_id={} batch={}",
                 field.name, field.id, batch_id
@@ -560,7 +560,7 @@ fn read_null_array(
             } else {
                 let idx_max = *indices.values().iter().max().unwrap() as u64;
                 if idx_max >= page_info.length as u64 {
-                    return Err(Error::io(
+                    return Err(Error::invalid_input(
                         format!(
                             "NullArray Reader: request([{}]) out of range: [0..{}]",
                             idx_max, page_info.length
@@ -580,7 +580,7 @@ fn read_null_array(
                 _ => unreachable!(),
             };
             if idx_end > page_info.length {
-                return Err(Error::io(
+                return Err(Error::invalid_input(
                     format!(
                         "NullArray Reader: request([{}..{}]) out of range: [0..{}]",
                         // and wrap it in here.
diff --git a/rust/lance-file/src/previous/writer/mod.rs b/rust/lance-file/src/previous/writer/mod.rs
index 3bef0a73455..7006442e08c 100644
--- a/rust/lance-file/src/previous/writer/mod.rs
+++ b/rust/lance-file/src/previous/writer/mod.rs
@@ -204,7 +204,7 @@ impl<M: ManifestProvider + Send + Sync> FileWriter<M> {
                 .iter()
                 .map(|batch| {
                     batch.column_by_name(&field.name).ok_or_else(|| {
-                        Error::io(
+                        Error::invalid_input(
                             format!("FileWriter::write: Field '{}' not found", field.name),
                             location!(),
                         )
@@ -639,9 +639,9 @@ impl<M: ManifestProvider + Send + Sync> FileWriter<M> {
                     })?;
 
                     let value_arr = dict_info.values.as_ref().ok_or_else(|| {
-                        Error::io(
+                        Error::invalid_input(
                             format!(
-                        "Lance field {} is dictionary type, but misses the dictionary value array", 
+                        "Lance field {} is dictionary type, but misses the dictionary value array",
                         field.name),
                             location!(),
                         )
@@ -658,7 +658,7 @@ impl<M: ManifestProvider + Send + Sync> FileWriter<M> {
                             encoder.encode(&[value_arr]).await?
                         }
                         _ => {
-                            return Err(Error::io(
+                            return Err(Error::schema(
                                 format!(
                                     "Does not support {} as dictionary value type",
                                     value_arr.data_type()
diff --git a/rust/lance-file/src/reader.rs b/rust/lance-file/src/reader.rs
index 354e0f920d0..ddc866e36db 100644
--- a/rust/lance-file/src/reader.rs
+++ b/rust/lance-file/src/reader.rs
@@ -445,7 +445,7 @@ impl FileReader {
     fn decode_footer(footer_bytes: &Bytes) -> Result<Footer> {
         let len = footer_bytes.len();
         if len < FOOTER_LEN {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "does not have sufficient data, len: {}, bytes: {:?}",
                     len, footer_bytes
@@ -474,7 +474,7 @@ impl FileReader {
 
         let magic_bytes = footer_bytes.slice(len - 4..);
         if magic_bytes.as_ref() != MAGIC {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "file does not appear to be a Lance file (invalid magic: {:?})",
                     MAGIC
diff --git a/rust/lance-index/src/scalar/inverted/tokenizer/jieba.rs b/rust/lance-index/src/scalar/inverted/tokenizer/jieba.rs
index bc3c6469321..d450658ee02 100644
--- a/rust/lance-index/src/scalar/inverted/tokenizer/jieba.rs
+++ b/rust/lance-index/src/scalar/inverted/tokenizer/jieba.rs
@@ -5,7 +5,6 @@ use std::{fs::File, io::BufReader, path::Path, path::PathBuf};
 
 use lance_core::{Error, Result};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
-use snafu::location;
 
 #[derive(Serialize, Deserialize, Default)]
 pub struct JiebaConfig {
@@ -20,8 +19,8 @@ pub trait JiebaTokenizerBuilder: Sized {
 
     fn load(p: &Path) -> Result<Self> {
         if !p.is_dir() {
-            return Err(Error::io(
-                format!("{} is not a valid directory", p.display()),
+            return Err(Error::invalid_input(
+                format!("Invalid directory path: {}", p.display()),
                 snafu::location!(),
             ));
         }
@@ -77,26 +76,26 @@ impl JiebaTokenizerBuilder for JiebaBuilder {
         let file = std::fs::File::open(main_dict_path)?;
         let mut f = std::io::BufReader::new(file);
         let mut jieba = jieba_rs::Jieba::with_dict(&mut f).map_err(|e| {
-            Error::io(
+            Error::invalid_input(
                 format!(
-                    "load jieba tokenizer dictionary {}, error: {}",
+                    "Failed to load Jieba dictionary from {}: {}",
                     main_dict_path.display(),
                     e
                 ),
-                location!(),
+                snafu::location!(),
             )
         })?;
         for user_dict_path in &self.user_dict_paths() {
             let file = std::fs::File::open(user_dict_path)?;
             let mut f = std::io::BufReader::new(file);
             jieba.load_dict(&mut f).map_err(|e| {
-                Error::io(
+                Error::invalid_input(
                     format!(
-                        "load jieba tokenizer user dictionary {},  error: {}",
+                        "Failed to load Jieba user dictionary from {}: {}",
                         user_dict_path.display(),
                         e
                     ),
-                    location!(),
+                    snafu::location!(),
                 )
             })?
         }
diff --git a/rust/lance-index/src/scalar/inverted/tokenizer/lindera.rs b/rust/lance-index/src/scalar/inverted/tokenizer/lindera.rs
index e7ea7ca6c09..ec492a43513 100644
--- a/rust/lance-index/src/scalar/inverted/tokenizer/lindera.rs
+++ b/rust/lance-index/src/scalar/inverted/tokenizer/lindera.rs
@@ -12,8 +12,8 @@ pub const LINDERA_LANGUAGE_MODEL_CONFIG_FILE: &str = "config.yml";
 pub trait LinderaTokenizerBuilder: Sized {
     fn load(p: &Path) -> Result<Self> {
         if !p.is_dir() {
-            return Err(Error::io(
-                format!("{} is not a valid directory", p.display()),
+            return Err(Error::invalid_input(
+                format!("Invalid directory path: {}", p.display()),
                 snafu::location!(),
             ));
         }
diff --git a/rust/lance-index/src/vector/utils.rs b/rust/lance-index/src/vector/utils.rs
index ac7772d4009..9f1a80476da 100644
--- a/rust/lance-index/src/vector/utils.rs
+++ b/rust/lance-index/src/vector/utils.rs
@@ -128,8 +128,8 @@ pub(crate) fn prefetch_arrow_array(array: &dyn Array) -> Result<()> {
             do_prefetch(array.values().as_ptr_range())
         }
         _ => {
-            return Err(Error::io(
-                format!("unsupported prefetch on {} type", array.data_type()),
+            return Err(Error::invalid_input(
+                format!("Unsupported data type for prefetch: {}", array.data_type()),
                 location!(),
             ));
         }
diff --git a/rust/lance-io/src/encodings/binary.rs b/rust/lance-io/src/encodings/binary.rs
index 34f4b05f80b..1d2c23030fe 100644
--- a/rust/lance-io/src/encodings/binary.rs
+++ b/rust/lance-io/src/encodings/binary.rs
@@ -99,8 +99,8 @@ impl Encoder for BinaryEncoder<'_> {
             DataType::LargeUtf8 => self.encode_typed_arr::<LargeUtf8Type>(arrs).await,
             DataType::LargeBinary => self.encode_typed_arr::<LargeBinaryType>(arrs).await,
             _ => {
-                return Err(lance_core::Error::io(
-                    format!("Binary encoder does not support {}", data_type),
+                return Err(lance_core::Error::invalid_input(
+                    format!("Unsupported data type for binary encoding: {}", data_type),
                     location!(),
                 ));
             }
diff --git a/rust/lance-io/src/encodings/plain.rs b/rust/lance-io/src/encodings/plain.rs
index 5f18ffcf947..9966d8b33ec 100644
--- a/rust/lance-io/src/encodings/plain.rs
+++ b/rust/lance-io/src/encodings/plain.rs
@@ -241,7 +241,7 @@ impl<'a> PlainDecoder<'a> {
     ///
     async fn decode_primitive(&self, start: usize, end: usize) -> Result<ArrayRef> {
         if end > self.length {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "PlainDecoder: request([{}..{}]) out of range: [0..{}]",
                     start, end, self.length
diff --git a/rust/lance-io/src/utils.rs b/rust/lance-io/src/utils.rs
index c63947803a1..776231916fb 100644
--- a/rust/lance-io/src/utils.rs
+++ b/rust/lance-io/src/utils.rs
@@ -50,7 +50,7 @@ pub async fn read_binary_array(
             reader, position, length, nullable,
         )),
         _ => {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!("Unsupported binary type: {}", data_type),
                 location!(),
             ));
diff --git a/rust/lance-table/src/format.rs b/rust/lance-table/src/format.rs
index 2e504a5aa28..ee58a736ba1 100644
--- a/rust/lance-table/src/format.rs
+++ b/rust/lance-table/src/format.rs
@@ -52,7 +52,7 @@ impl TryFrom<&pb::Uuid> for Uuid {
 
     fn try_from(p: &pb::Uuid) -> Result<Self> {
         if p.uuid.len() != 16 {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 "Protobuf UUID is malformed".to_string(),
                 location!(),
             ));
diff --git a/rust/lance-table/src/format/index.rs b/rust/lance-table/src/format/index.rs
index fdb2e2b4b90..bf1201fe129 100644
--- a/rust/lance-table/src/format/index.rs
+++ b/rust/lance-table/src/format/index.rs
@@ -97,7 +97,7 @@ impl TryFrom<pb::IndexMetadata> for IndexMetadata {
 
         Ok(Self {
             uuid: proto.uuid.as_ref().map(Uuid::try_from).ok_or_else(|| {
-                Error::io(
+                Error::invalid_input(
                     "uuid field does not exist in Index metadata".to_string(),
                     location!(),
                 )
diff --git a/rust/lance-table/src/format/manifest.rs b/rust/lance-table/src/format/manifest.rs
index b556dbdd468..d50e59d1bc7 100644
--- a/rust/lance-table/src/format/manifest.rs
+++ b/rust/lance-table/src/format/manifest.rs
@@ -441,7 +441,7 @@ impl Manifest {
     /// Note this does not support recycling of fragment ids.
     pub fn fragments_since(&self, since: &Self) -> Result<Vec<Fragment>> {
         if since.version >= self.version {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "fragments_since: given version {} is newer than manifest version {}",
                     since.version, self.version
diff --git a/rust/lance-table/src/io/commit/dynamodb.rs b/rust/lance-table/src/io/commit/dynamodb.rs
index 9db7ce5027e..c4c0614e7d6 100644
--- a/rust/lance-table/src/io/commit/dynamodb.rs
+++ b/rust/lance-table/src/io/commit/dynamodb.rs
@@ -275,11 +275,11 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
 
         let path = item
             .get(path!())
-            .ok_or_else(|| Error::io(format!("key {} is not present", path!()), location!()))?;
+            .ok_or_else(|| Error::not_found(format!("key {} is not present", path!())))?;
 
         match path {
             AttributeValue::S(path) => Ok(path.clone()),
-            _ => Err(Error::io(
+            _ => Err(Error::invalid_input(
                 format!("key {} is not a string", path!()),
                 location!(),
             )),
@@ -309,9 +309,11 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
 
         let path = item
             .get(path!())
-            .ok_or_else(|| Error::io(format!("key {} is not present", path!()), location!()))?
+            .ok_or_else(|| Error::not_found(format!("key {} is not present", path!())))?
             .as_s()
-            .map_err(|_| Error::io(format!("key {} is not a string", path!()), location!()))?
+            .map_err(|_| {
+                Error::invalid_input(format!("key {} is not a string", path!()), location!())
+            })?
             .as_str();
         let path = Path::from(path);
 
@@ -362,9 +364,9 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
                     return Ok(None);
                 }
                 if items.len() > 1 {
-                    return Err(Error::io(
+                    return Err(Error::invalid_input(
                         format!(
-                            "dynamodb table: {} return unexpected number of items",
+                            "dynamodb table: {} returned unexpected number of items",
                             self.table_name
                         ),
                         location!(),
@@ -373,22 +375,16 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
 
                 let item = items.pop().expect("length checked");
                 let version_attribute = item
-                .get(version!())
-                .ok_or_else(||
-                    Error::io(
-                        format!("dynamodb error: found entries for {} but the returned data does not contain {} column", base_uri, version!()),
-                        location!(),
-                    )
-                )?;
+                    .get(version!())
+                    .ok_or_else(|| Error::not_found(
+                        format!("dynamodb error: found entries for {} but the returned data does not contain {} column", base_uri, version!())
+                    ))?;
 
                 let path_attribute = item
-                .get(path!())
-                .ok_or_else(||
-                    Error::io(
-                        format!("dynamodb error: found entries for {} but the returned data does not contain {} column", base_uri, path!()),
-                        location!(),
-                    )
-                )?;
+                    .get(path!())
+                    .ok_or_else(|| Error::not_found(
+                        format!("dynamodb error: found entries for {} but the returned data does not contain {} column", base_uri, path!())
+                    ))?;
 
                 let size = item.get("size").and_then(|attr| match attr {
                     AttributeValue::N(size) => size.parse().ok(),
@@ -399,7 +395,7 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
 
                 match (version_attribute, path_attribute) {
                     (AttributeValue::N(version), AttributeValue::S(path)) => {
-                        let version = version.parse().map_err(|e| Error::io(
+                        let version = version.parse().map_err(|e| Error::invalid_input(
                             format!("dynamodb error: could not parse the version number returned {}, error: {}", version, e),
                             location!(),
                         ))?;
@@ -414,7 +410,7 @@ impl ExternalManifestStore for DynamoDBExternalManifestStore {
                         };
                         Ok(Some(location))
                     },
-                    _ => Err(Error::io(
+                    _ => Err(Error::invalid_input(
                         format!("dynamodb error: found entries for {base_uri} but the returned data is not number type"),
                         location!(),
                     ))
diff --git a/rust/lance-table/src/io/manifest.rs b/rust/lance-table/src/io/manifest.rs
index 1612800f201..2f938d19ece 100644
--- a/rust/lance-table/src/io/manifest.rs
+++ b/rust/lance-table/src/io/manifest.rs
@@ -57,13 +57,15 @@ pub async fn read_manifest(
     }
 
     if buf.len() < 16 {
-        return Err(Error::io(
+        return Err(Error::corrupt_file(
+            path.clone(),
             "Invalid format: file size is smaller than 16 bytes".to_string(),
             location!(),
         ));
     }
     if !buf.ends_with(MAGIC) {
-        return Err(Error::io(
+        return Err(Error::corrupt_file(
+            path.clone(),
             "Invalid format: magic number does not match".to_string(),
             location!(),
         ));
@@ -98,7 +100,7 @@ pub async fn read_manifest(
     let buf = buf.slice(4..buf.len() - 16);
 
     if buf.len() != recorded_length {
-        return Err(Error::io(
+        return Err(Error::invalid_input(
             format!(
                 "Invalid format: manifest length does not match. Expected {}, got {}",
                 recorded_length,
@@ -206,7 +208,7 @@ pub async fn write_manifest(
                         encoder.encode(&[value_arr]).await?
                     }
                     _ => {
-                        return Err(Error::io(
+                        return Err(Error::schema(
                             format!(
                                 "Does not support {} as dictionary value type",
                                 value_arr.data_type()
diff --git a/rust/lance/src/dataset/blob.rs b/rust/lance/src/dataset/blob.rs
index 7238cad866f..3debbd7298f 100644
--- a/rust/lance/src/dataset/blob.rs
+++ b/rust/lance/src/dataset/blob.rs
@@ -480,10 +480,10 @@ impl BlobFile {
                 *cursor = new_cursor;
                 Ok(data)
             }
-            ReaderState::Closed => Err(Error::IO {
-                location: location!(),
-                source: "Blob file is already closed".into(),
-            }),
+            ReaderState::Closed => Err(Error::invalid_input(
+                "Blob file is already closed".to_string(),
+                location!(),
+            )),
             _ => unreachable!(),
         }
     }
@@ -529,10 +529,10 @@ impl BlobFile {
                 *cursor = new_cursor;
                 Ok(())
             }
-            ReaderState::Closed => Err(Error::IO {
-                location: location!(),
-                source: "Blob file is already closed".into(),
-            }),
+            ReaderState::Closed => Err(Error::invalid_input(
+                "Blob file is already closed".to_string(),
+                location!(),
+            )),
             ReaderState::Uninitialized(cursor) => {
                 *cursor = new_cursor;
                 Ok(())
@@ -545,10 +545,10 @@ impl BlobFile {
         let reader = self.reader.lock().await;
         match *reader {
             ReaderState::Open((cursor, _)) => Ok(cursor),
-            ReaderState::Closed => Err(Error::IO {
-                location: location!(),
-                source: "Blob file is already closed".into(),
-            }),
+            ReaderState::Closed => Err(Error::invalid_input(
+                "Blob file is already closed".to_string(),
+                location!(),
+            )),
             ReaderState::Uninitialized(cursor) => Ok(cursor),
         }
     }
diff --git a/rust/lance/src/dataset/fragment.rs b/rust/lance/src/dataset/fragment.rs
index 099b883d355..cfe003a7233 100644
--- a/rust/lance/src/dataset/fragment.rs
+++ b/rust/lance/src/dataset/fragment.rs
@@ -724,7 +724,7 @@ impl FileFragment {
             determine_file_version(dataset.object_store.as_ref(), &filepath, None).await?;
 
         if file_version != dataset.manifest.data_storage_format.lance_file_version()? {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "File version mismatch. Dataset version: {:?} Fragment version: {:?}",
                     dataset.manifest.data_storage_format.lance_file_version()?,
@@ -876,14 +876,11 @@ impl FileFragment {
         let row_id_sequence = row_id_sequence?;
 
         if opened_files.is_empty() && !read_config.has_system_cols() {
-            return Err(Error::io(
-                format!(
-                    "Did not find any data files for schema: {}\nfragment_id={}",
-                    projection,
-                    self.id()
-                ),
-                location!(),
-            ));
+            return Err(Error::not_found(format!(
+                "No data files found for schema: {}, fragment_id={}",
+                projection,
+                self.id()
+            )));
         }
 
         let num_physical_rows = self.physical_rows().await?;
@@ -1158,10 +1155,10 @@ impl FileFragment {
     /// fragment.
     pub async fn physical_rows(&self) -> Result<usize> {
         if self.metadata.files.is_empty() {
-            return Err(Error::io(
-                format!("Fragment {} does not contain any data", self.id()),
-                location!(),
-            ));
+            return Err(Error::not_found(format!(
+                "Fragment {} does not contain any data",
+                self.id()
+            )));
         };
 
         // Early versions that did not write the writer version also could write
@@ -1996,7 +1993,7 @@ impl std::fmt::Display for FragmentReader {
 
 fn merge_batches(batches: &[RecordBatch]) -> Result<RecordBatch> {
     if batches.is_empty() {
-        return Err(Error::io(
+        return Err(Error::invalid_input(
             "Cannot merge empty batches".to_string(),
             location!(),
         ));
@@ -2026,14 +2023,14 @@ impl FragmentReader {
             for reader in readers.iter().skip(1) {
                 if let Some(other_legacy) = reader.as_legacy_opt() {
                     if other_legacy.num_batches() != num_batches {
-                        return Err(Error::io(
+                        return Err(Error::invalid_input(
                                 "Cannot create FragmentReader from data files with different number of batches"
                                     .to_string(),
                             location!(),
                         ));
                     }
                 } else {
-                    return Err(Error::io(
+                    return Err(Error::invalid_input(
                         "Cannot mix legacy and non-legacy readers".to_string(),
                         location!(),
                     ));
@@ -3833,7 +3830,7 @@ mod tests {
                 FragReadConfig::default(),
             )
             .await;
-        assert!(matches!(res, Err(Error::IO { .. })));
+        assert!(matches!(res, Err(Error::NotFound { .. })));
 
         Ok(())
     }
diff --git a/rust/lance/src/dataset/hash_joiner.rs b/rust/lance/src/dataset/hash_joiner.rs
index 8c93a8d7bcf..e9f8c14d9bb 100644
--- a/rust/lance/src/dataset/hash_joiner.rs
+++ b/rust/lance/src/dataset/hash_joiner.rs
@@ -53,7 +53,10 @@ impl HashJoiner {
         .await
         .unwrap()?;
         if batches.is_empty() {
-            return Err(Error::io("HashJoiner: No data".to_string(), location!()));
+            return Err(Error::invalid_input(
+                "HashJoiner: No data".to_string(),
+                location!(),
+            ));
         };
 
         let map = DashMap::new();
@@ -95,7 +98,10 @@ impl HashJoiner {
                     match task_result {
                         Ok(Ok(_)) => Ok(()),
                         Ok(Err(err)) => Err(err),
-                        Err(err) => Err(Error::io(format!("HashJoiner: {}", err), location!())),
+                        Err(err) => Err(Error::invalid_input(
+                            format!("HashJoiner: {}", err),
+                            location!(),
+                        )),
                     }
                 }
             })
@@ -175,7 +181,7 @@ impl HashJoiner {
                     let task_result = task::spawn_blocking(move || {
                         let array_refs = arrays.iter().map(|x| x.as_ref()).collect::<Vec<_>>();
                         interleave(array_refs.as_ref(), indices.as_ref())
-                            .map_err(|err| Error::io(
+                            .map_err(|err| Error::invalid_input(
                                 format!("HashJoiner: {}", err),
                                 location!(),
                             ))
@@ -258,8 +264,9 @@ impl HashJoiner {
                 async move {
                     let task_result = task::spawn_blocking(move || {
                         let array_refs = arrays.iter().map(|x| x.as_ref()).collect::<Vec<_>>();
-                        interleave(array_refs.as_ref(), indices.as_ref())
-                            .map_err(|err| Error::io(format!("HashJoiner: {}", err), location!()))
+                        interleave(array_refs.as_ref(), indices.as_ref()).map_err(|err| {
+                            Error::invalid_input(format!("HashJoiner: {}", err), location!())
+                        })
                     })
                     .await;
                     match task_result {
@@ -286,7 +293,10 @@ impl HashJoiner {
                             Ok(array)
                         }
                         Ok(Err(err)) => Err(err),
-                        Err(err) => Err(Error::io(format!("HashJoiner: {}", err), location!())),
+                        Err(err) => Err(Error::invalid_input(
+                            format!("HashJoiner: {}", err),
+                            location!(),
+                        )),
                     }
                 }
             })
diff --git a/rust/lance/src/dataset/scanner.rs b/rust/lance/src/dataset/scanner.rs
index fc5f636f5b1..16a5fe4d565 100644
--- a/rust/lance/src/dataset/scanner.rs
+++ b/rust/lance/src/dataset/scanner.rs
@@ -815,7 +815,7 @@ impl Scanner {
 
     fn ensure_not_fragment_scan(&self) -> Result<()> {
         if self.is_fragment_scan() {
-            Err(Error::io(
+            Err(Error::not_supported(
                 "This operation is not supported for fragment scan".to_string(),
                 location!(),
             ))
@@ -1730,7 +1730,7 @@ impl Scanner {
                     .column(0)
                     .as_any()
                     .downcast_ref::<Int64Array>()
-                    .ok_or(Error::io(
+                    .ok_or(Error::invalid_input(
                         "Count plan did not return a UInt64Array".to_string(),
                         location!(),
                     ))?;
diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs
index f954115ada3..e5916b7bae1 100644
--- a/rust/lance/src/dataset/transaction.rs
+++ b/rust/lance/src/dataset/transaction.rs
@@ -3073,7 +3073,7 @@ impl TryFrom<&pb::transaction::rewrite::RewrittenIndex> for RewrittenIndex {
                 .as_ref()
                 .map(Uuid::try_from)
                 .ok_or_else(|| {
-                    Error::io(
+                    Error::invalid_input(
                         "required field (old_id) missing from message".to_string(),
                         location!(),
                     )
@@ -3083,7 +3083,7 @@ impl TryFrom<&pb::transaction::rewrite::RewrittenIndex> for RewrittenIndex {
                 .as_ref()
                 .map(Uuid::try_from)
                 .ok_or_else(|| {
-                    Error::io(
+                    Error::invalid_input(
                         "required field (new_id) missing from message".to_string(),
                         location!(),
                     )
diff --git a/rust/lance/src/dataset/updater.rs b/rust/lance/src/dataset/updater.rs
index 7aa0fe41fe4..7197f877943 100644
--- a/rust/lance/src/dataset/updater.rs
+++ b/rust/lance/src/dataset/updater.rs
@@ -155,14 +155,14 @@ impl Updater {
     /// Update one batch.
     pub async fn update(&mut self, batch: RecordBatch) -> Result<()> {
         let Some(last) = self.last_input.as_ref() else {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 "Fragment Updater: no input data is available before update".to_string(),
                 location!(),
             ));
         };
 
         if last.num_rows() != batch.num_rows() {
-            return Err(Error::io(
+            return Err(Error::invalid_input(
                 format!(
                     "Fragment Updater: new batch has different size with the source batch: {} != {}",
                     last.num_rows(),
diff --git a/rust/lance/src/index/vector/builder.rs b/rust/lance/src/index/vector/builder.rs
index e05d54c2540..26fed852b86 100644
--- a/rust/lance/src/index/vector/builder.rs
+++ b/rust/lance/src/index/vector/builder.rs
@@ -943,10 +943,14 @@ impl<S: IvfSubIndex + 'static, Q: Quantization + 'static> IvfIndexBuilder<S, Q>
         // This can happen after a split creates a new partition
         if let Some(reader) = reader {
             if reader.partition_size(part_id)? > 0 {
-                let mut partition_data = reader.read_partition(part_id).await?.ok_or(Error::io(
-                    format!("partition {} is empty", part_id).as_str(),
-                    location!(),
-                ))?;
+                let mut partition_data =
+                    reader
+                        .read_partition(part_id)
+                        .await?
+                        .ok_or(Error::invalid_input(
+                            format!("partition {} is empty", part_id),
+                            location!(),
+                        ))?;
                 while let Some(batch) = partition_data.try_next().await? {
                     loss += batch
                         .metadata()