feat: add location for all error (#1475)

Closed #1275

python/src/dataset/commit.rs

@@ -14,6 +14,8 @@
 
 use std::fmt::Debug;
 
+use snafu::{location, Location};
+
 use lance_core::io::commit::{CommitError, CommitLease, CommitLock};
 use lance_core::Error;
 
@@ -36,6 +38,7 @@ fn handle_error(py_err: PyErr, py: Python) -> CommitError {
         Err(import_error) => {
             return CommitError::OtherError(Error::Internal {
                 message: format!("Error importing from pylance {}", import_error),
+                location: location!(),
             })
         }
     };
@@ -45,6 +48,7 @@ fn handle_error(py_err: PyErr, py: Python) -> CommitError {
     } else {
         CommitError::OtherError(Error::Internal {
             message: format!("Error from commit handler: {}", py_err),
+            location: location!(),
         })
     }
 }

rust/lance-core/src/datatypes.rs

@@ -77,6 +77,7 @@ fn parse_timeunit(unit: &str) -> Result<TimeUnit> {
         "ns" => Ok(TimeUnit::Nanosecond),
         _ => Err(Error::Arrow {
             message: format!("Unsupported TimeUnit: {unit}"),
+            location: location!(),
         }),
     }
 }

rust/lance-core/src/datatypes/field.rs

@@ -310,6 +310,7 @@ impl Field {
                     "Attempt to intersect different fields: {} and {}",
                     self.name, other.name,
                 ),
+                location: location!(),
             });
         }
         let self_type = self.data_type();
@@ -347,6 +348,7 @@ impl Field {
                     "Attempt to intersect different fields: ({}, {}) and ({}, {})",
                     self.name, self_type, other.name, other_type
                 ),
+                location: location!(),
             });
         }
 

rust/lance-core/src/encodings/dictionary.rs

@@ -143,6 +143,7 @@ impl<'a> DictionaryDecoder<'a> {
         } else {
             return Err(Error::Arrow {
                 message: format!("Not a dictionary type: {}", self.data_type),
+                location: location!(),
             });
         };
 
@@ -160,6 +161,7 @@ impl<'a> DictionaryDecoder<'a> {
             DataType::UInt64 => self.make_dict_array::<UInt64Type>(keys).await,
             _ => Err(Error::Arrow {
                 message: format!("Dictionary encoding does not support index type: {index_type}",),
+                location: location!(),
             }),
         }
     }
@@ -196,6 +198,7 @@ impl<'a> AsyncIndex<usize> for DictionaryDecoder<'a> {
             source: "DictionaryDecoder does not support get()"
                 .to_string()
                 .into(),
+            location: location!(),
         })
     }
 }

rust/lance-core/src/error.rs

@@ -27,56 +27,77 @@ pub fn box_error(e: impl std::error::Error + Send + Sync + 'static) -> BoxedErro
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]
 pub enum Error {
-    #[snafu(display("Invalid user input: {source}"))]
-    InvalidInput { source: BoxedError },
-    #[snafu(display("Dataset already exists: {uri}"))]
-    DatasetAlreadyExists { uri: String },
+    #[snafu(display("Invalid user input: {source}, {location}"))]
+    InvalidInput {
+        source: BoxedError,
+        location: Location,
+    },
+    #[snafu(display("Dataset already exists: {uri}, {location}"))]
+    DatasetAlreadyExists { uri: String, location: Location },
     // #[snafu(display("Append with different schema: original={original} new={new}"))]
     #[snafu(display("Append with different schema:"))]
     SchemaMismatch {},
-    #[snafu(display("Dataset at path {path} was not found: {source}"))]
-    DatasetNotFound { path: String, source: BoxedError },
-    #[snafu(display("Encountered corrupt file {path}: {source}"))]
+    #[snafu(display("Dataset at path {path} was not found: {source}, {location}"))]
+    DatasetNotFound {
+        path: String,
+        source: BoxedError,
+        location: Location,
+    },
+    #[snafu(display("Encountered corrupt file {path}: {source}, {location}"))]
     CorruptFile {
         path: object_store::path::Path,
         source: BoxedError,
+        location: Location,
         // TODO: add backtrace?
     },
-    #[snafu(display("Not supported: {source}"))]
-    NotSupported { source: BoxedError },
-    #[snafu(display("Commit conflict for version {version}: {source}"))]
-    CommitConflict { version: u64, source: BoxedError },
-    #[snafu(display("Encountered internal error. Please file a bug report at https://github.com/lancedb/lance/issues. {message}"))]
-    Internal { message: String },
-    #[snafu(display("A prerequisite task failed: {message}"))]
-    PrerequisiteFailed { message: String },
-    #[snafu(display("LanceError(Arrow): {message}"))]
-    Arrow { message: String },
+    #[snafu(display("Not supported: {source}, {location}"))]
+    NotSupported {
+        source: BoxedError,
+        location: Location,
+    },
+    #[snafu(display("Commit conflict for version {version}: {source}, {location}"))]
+    CommitConflict {
+        version: u64,
+        source: BoxedError,
+        location: Location,
+    },
+    #[snafu(display("Encountered internal error. Please file a bug report at https://github.com/lancedb/lance/issues. {message}, {location}"))]
+    Internal { message: String, location: Location },
+    #[snafu(display("A prerequisite task failed: {message}, {location}"))]
+    PrerequisiteFailed { message: String, location: Location },
+    #[snafu(display("LanceError(Arrow): {message}, {location}"))]
+    Arrow { message: String, location: Location },
     #[snafu(display("LanceError(Schema): {message}, {location}"))]
     Schema { message: String, location: Location },
     #[snafu(display("Not found: {uri}, {location}"))]
     NotFound { uri: String, location: Location },
     #[snafu(display("LanceError(IO): {message}, {location}"))]
     IO { message: String, location: Location },
-    #[snafu(display("LanceError(Index): {message}"))]
-    Index { message: String },
+    #[snafu(display("LanceError(Index): {message}, {location}"))]
+    Index { message: String, location: Location },
     /// Stream early stop
     Stop,
 }
 
 impl Error {
-    pub fn corrupt_file(path: object_store::path::Path, message: impl Into<String>) -> Self {
+    pub fn corrupt_file(
+        path: object_store::path::Path,
+        message: impl Into<String>,
+        location: Location,
+    ) -> Self {
         let message: String = message.into();
         Self::CorruptFile {
             path,
             source: message.into(),
+            location,
         }
     }
 
-    pub fn invalid_input(message: impl Into<String>) -> Self {
+    pub fn invalid_input(message: impl Into<String>, location: Location) -> Self {
         let message: String = message.into();
         Self::InvalidInput {
             source: message.into(),
+            location,
         }
     }
 }
@@ -87,6 +108,7 @@ impl From<ArrowError> for Error {
     fn from(e: ArrowError) -> Self {
         Self::Arrow {
             message: e.to_string(),
+            location: location!(),
         }
     }
 }
@@ -95,6 +117,7 @@ impl From<&ArrowError> for Error {
     fn from(e: &ArrowError) -> Self {
         Self::Arrow {
             message: e.to_string(),
+            location: location!(),
         }
     }
 }
@@ -157,6 +180,7 @@ impl From<serde_json::Error> for Error {
     fn from(e: serde_json::Error) -> Self {
         Self::Arrow {
             message: e.to_string(),
+            location: location!(),
         }
     }
 }
@@ -171,10 +195,10 @@ fn arrow_io_error_from_msg(message: String) -> ArrowError {
 impl From<Error> for ArrowError {
     fn from(value: Error) -> Self {
         match value {
-            Error::Arrow { message } => arrow_io_error_from_msg(message), // we lose the error type converting to LanceError
+            Error::Arrow { message, .. } => arrow_io_error_from_msg(message), // we lose the error type converting to LanceError
             Error::IO { message, .. } => arrow_io_error_from_msg(message),
             Error::Schema { message, .. } => Self::SchemaError(message),
-            Error::Index { message } => arrow_io_error_from_msg(message),
+            Error::Index { message, .. } => arrow_io_error_from_msg(message),
             Error::Stop => arrow_io_error_from_msg("early stop".to_string()),
             e => arrow_io_error_from_msg(e.to_string()), // Find a more scalable way of doing this
         }

rust/lance-core/src/format/page_table.rs

@@ -15,6 +15,7 @@
 use arrow_array::builder::Int64Builder;
 use arrow_array::{Array, Int64Array};
 use arrow_schema::DataType;
+use snafu::{location, Location};
 use std::collections::BTreeMap;
 use tokio::io::AsyncWriteExt;
 
@@ -96,6 +97,7 @@ impl PageTable {
         if self.pages.is_empty() {
             return Err(Error::InvalidInput {
                 source: "empty page table".into(),
+                location: location!(),
             });
         }
 

rust/lance-core/src/io/commit.rs

@@ -138,6 +138,7 @@ pub fn parse_version_from_path(path: &Path) -> Result<u64> {
         .and_then(|(version, _)| version.parse::<u64>().ok())
         .ok_or(crate::Error::Internal {
             message: format!("Expected manifest file, but found {}", path),
+            location: location!(),
         })
 }
 
@@ -248,6 +249,7 @@ impl From<CommitError> for Error {
         match e {
             CommitError::CommitConflict => Self::Internal {
                 message: "Commit conflict".to_string(),
+                location: location!(),
             },
             CommitError::OtherError(e) => e,
         }

rust/lance-core/src/io/deletion.rs

@@ -25,7 +25,7 @@ use bytes::Buf;
 use object_store::path::Path;
 use rand::Rng;
 use roaring::bitmap::RoaringBitmap;
-use snafu::ResultExt;
+use snafu::{location, Location, ResultExt};
 
 use super::object_store::ObjectStore;
 use crate::error::{box_error, CorruptFileSnafu};
@@ -315,6 +315,7 @@ pub async fn read_deletion_file(
                         "Expected exactly one batch in deletion file, got {}",
                         batches.len()
                     ),
+                    location!(),
                 ));
             }
 
@@ -327,6 +328,7 @@ pub async fn read_deletion_file(
                         deletion_arrow_schema(),
                         batch.schema()
                     ),
+                    location!(),
                 ));
             }
 
@@ -343,6 +345,7 @@ pub async fn read_deletion_file(
                     return Err(Error::corrupt_file(
                         path,
                         "Null values are not allowed in deletion files",
+                        location!(),
                     ));
                 }
             }

rust/lance-core/src/io/object_store.rs

@@ -462,6 +462,7 @@ impl ObjectStore {
                         source:
                             "`s3+ddb://` scheme and custom commit handler are mutually exclusive"
                                 .into(),
+                        location: location!(),
                     });
                 }
 
@@ -473,6 +474,7 @@ impl ObjectStore {
                             source:
                                 "`s3+ddb://` scheme and expects exactly one query `ddbTableName`"
                                     .into(),
+                            location: location!(),
                         });
                     }
                     match url.query_pairs().next() {
@@ -484,13 +486,15 @@ impl ObjectStore {
                                     source:
                                         "`s3+ddb://` scheme requires non empty dynamodb table name"
                                             .into(),
+                                    location: location!(),
                                 });
                             }
                             Some(table_name)
                         }
                         _ => {
                             return Err(Error::InvalidInput {
-                                source: "`s3+ddb://` scheme and expects exactly one query `ddbTableName`".into()
+                                source: "`s3+ddb://` scheme and expects exactly one query `ddbTableName`".into(),
+                                location: location!(),
                             });
                         }
                     }
@@ -519,6 +523,7 @@ impl ObjectStore {
                         return Err(Error::InvalidInput {
                             source: "`s3+ddb://` scheme requires `dynamodb` feature to be enabled"
                                 .into(),
+                            location: location!(),
                         });
                     }
                     None => params
@@ -530,6 +535,7 @@ impl ObjectStore {
                 // before creating the OSObjectStore we need to rewrite the url to drop ddb related parts
                 url.set_scheme("s3").map_err(|()| Error::Internal {
                     message: "could not set scheme".into(),
+                    location: location!(),
                 })?;
                 url.set_query(None);
 

rust/lance-core/src/io/reader.rs

@@ -268,6 +268,7 @@ impl FileReader {
                                     "File {} not found in fragment {:?}",
                                     path, fragment
                                 ),
+                                location: location!(),
                             })?
                             .fields,
                     )

rust/lance-index/src/vector/ivf.rs

@@ -164,13 +164,15 @@ impl Ivf {
     ) -> Result<RecordBatch> {
         let vector_arr = batch.column_by_name(column).ok_or(Error::Index {
             message: format!("Column {} does not exist.", column),
+            location: location!(),
         })?;
         let data = vector_arr.as_fixed_size_list_opt().ok_or(Error::Index {
             message: format!(
                 "Column {} is not a vector type: {}",
                 column,
                 vector_arr.data_type()
             ),
+            location: location!(),
         })?;
         let matrix = MatrixView::<Float32Type>::try_from(data)?;
         let part_ids = self.compute_partitions(&matrix).await;

rust/lance-index/src/vector/kmeans.rs

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use arrow_array::{builder::Float32Builder, FixedSizeListArray, Float32Array};
 use lance_arrow::FixedSizeListArrayExt;
 use log::info;
 use rand::{seq::IteratorRandom, Rng};
+use snafu::{location, Location};
+use std::sync::Arc;
 
 use lance_core::{Error, Result};
 use lance_linalg::{
@@ -42,7 +42,7 @@ pub async fn train_kmeans(
     if num_rows < k {
         return Err(Error::Index{message: format!(
             "KMeans: can not train {k} centroids with {num_rows} vectors, choose a smaller K (< {num_rows}) instead"
-        )});
+        ),location: location!()});
     }
     // Ony sample sample_rate * num_clusters. See Faiss
     let data = if num_rows > sample_rate * k {

rust/lance-index/src/vector/pq.rs

@@ -25,6 +25,7 @@ use lance_linalg::distance::{cosine_distance_batch, dot_distance_batch, l2_dista
 use lance_linalg::kernels::argmin;
 use lance_linalg::{distance::MetricType, MatrixView};
 use rand::SeedableRng;
+use snafu::{location, Location};
 pub mod transform;
 
 use super::kmeans::train_kmeans;
@@ -357,6 +358,7 @@ impl ProductQuantizer {
                     data.num_columns(),
                     params.num_sub_vectors
                 ),
+                location: location!(),
             });
         }
         assert_eq!(data.data().null_count(), 0);