feat: refactor use of Error::io

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))
             })
         })
     }

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()))
     )
 }
 

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,15 +3499,15 @@ 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!(),
             )
         })?;
         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

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> {

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!(),
                     ));

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())
 }
 

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,

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.

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()

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

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!(),
                 )
             })?
         }

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!(),
             ));
         }

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!(),
             ));
         }

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!(),
                 ));
             }

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

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!(),
             ));

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!(),
             ));

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!(),
                 )

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