chore: Better error handling

docs/search/indexing/bm25.mdx

@@ -9,6 +9,7 @@ It ranks documents by considering how often a term appears and how unique that t
 documents. BM25 is especially useful when you need to find exact keywords or phrases within documents.
 
 The BM25 index enables full text search over a table and relevance scoring using the BM25 algorithm.
+This index is strongly consistent, which means that new data is immediately searchable across all connections.
 
 ## Creating a BM25 Index
 

docs/search/indexing/overview.mdx

@@ -6,8 +6,8 @@ title: Indexing Explained
 
 In Postgres, indexes are database objects that provide a fast way to look up rows in a table. If you
 imagine a Postgres table as a book, then an index is like the table of contents. Instead of flipping
-through every page to find a chapter, the table of contents stores additional information that enables
-readers to locate a chapter much faster.
+through every page to find a chapter, the table of contents stores additional information that can help
+readers find chapters faster.
 
 ## ParadeDB Indexes
 
@@ -16,4 +16,3 @@ the HNSW index for dense vector search, and the sparse HNSW index for sparse vec
 
 Once an index is created, it will automatically stay in sync with the underlying table data, and
 does not need to be re-created unless the table schema changes (for instance, if a column is renamed or deleted).
-The bm25 index is strongly consistent. This means that new data is immediately searchable across all connections.

pg_analytics/src/api/init.rs

@@ -11,5 +11,7 @@
 #[pg_guard]
 #[no_mangle]
 pub extern "C" fn init() {
-    let _ = DatafusionContext::init().expect("Failed to initialize context");
+    let _ = DatafusionContext::init().unwrap_or_else(|err| {
+        panic!("{}", err);
+    });
 }

pg_analytics/src/datafusion/catalog.rs

@@ -3,7 +3,7 @@
 use deltalake::datafusion::common::DataFusionError;
 use parking_lot::RwLock;
 use pgrx::*;
-use std::{any::Any, collections::HashMap, ffi::CStr, sync::Arc};
+use std::{any::type_name, any::Any, collections::HashMap, ffi::CStr, ffi::OsStr, sync::Arc};
 
 use crate::datafusion::directory::ParadeDirectory;
 use crate::datafusion::schema::ParadeSchemaProvider;
@@ -36,9 +36,9 @@
             if path.is_dir() {
                 let schema_oid = path
                     .file_name()
-                    .ok_or_else(|| ParadeError::NotFound)?
+                    .ok_or(ParadeError::NoneError(type_name::<OsStr>().to_string()))?
                     .to_str()
-                    .ok_or_else(|| ParadeError::NotFound)?
+                    .ok_or(ParadeError::NoneError(type_name::<str>().to_string()))?
                     .parse::<u32>()?;
 
                 let pg_oid = pg_sys::Oid::from(schema_oid);

pg_analytics/src/datafusion/context.rs

@@ -11,6 +11,7 @@
 use deltalake::datafusion::sql::TableReference;
 use lazy_static::lazy_static;
 use parking_lot::{RwLock, RwLockWriteGuard};
+use std::any::type_name;
 use std::sync::Arc;
 
 use crate::datafusion::catalog::{ParadeCatalog, ParadeCatalogList, PARADE_CATALOG};
@@ -56,14 +57,16 @@
 
         let schema_provider = context
             .catalog(PARADE_CATALOG)
-            .ok_or_else(|| ParadeError::NotFound)?
+            .ok_or(ParadeError::CatalogNotFound(PARADE_CATALOG.to_string()))?
             .schema(schema_name)
-            .ok_or_else(|| ParadeError::NotFound)?;
+            .ok_or(ParadeError::SchemaNotFound(schema_name.to_string()))?;
 
         let parade_provider = schema_provider
             .as_any()
             .downcast_ref::<ParadeSchemaProvider>()
-            .ok_or_else(|| ParadeError::NotFound)?;
+            .ok_or(ParadeError::NoneError(
+                type_name::<ParadeSchemaProvider>().to_string(),
+            ))?;
 
         f(parade_provider)
     }
@@ -83,12 +86,14 @@
 
         let catalog_provider = context
             .catalog(PARADE_CATALOG)
-            .ok_or_else(|| ParadeError::NotFound)?;
+            .ok_or(ParadeError::CatalogNotFound(PARADE_CATALOG.to_string()))?;
 
         let parade_catalog = catalog_provider
             .as_any()
             .downcast_ref::<ParadeCatalog>()
-            .ok_or_else(|| ParadeError::NotFound)?;
+            .ok_or(ParadeError::NoneError(
+                type_name::<ParadeCatalog>().to_string(),
+            ))?;
 
         f(parade_catalog)
     }
@@ -151,8 +156,8 @@
         let schema_name = reference.schema().unwrap_or("public");
 
         DatafusionContext::with_schema_provider(schema_name, |provider| {
-            let table =
-                task::block_on(provider.table(table_name)).ok_or_else(|| ParadeError::NotFound)?;
+            let table = task::block_on(provider.table(table_name))
+                .ok_or(ParadeError::TableNotFound(table_name.to_string()))?;
 
             Ok(provider_as_source(table))
         })

pg_analytics/src/datafusion/datatype.rs

@@ -13,6 +13,7 @@
 };
 use pgrx::pg_sys::{Datum, VARHDRSZ};
 use pgrx::*;
+use std::any::type_name;
 use std::sync::Arc;
 
 use crate::errors::ParadeError;
@@ -39,19 +40,21 @@
                 ExactNumberInfo::PrecisionAndScale(*precision as u64, *scale as u64),
             ),
             DataType::Time64(TimeUnit::Microsecond) => {
-                return Err(ParadeError::NotSupported("time not supported".into()))
+                return Err(ParadeError::NotSupported("time not supported".to_string()))
             }
             DataType::Timestamp(TimeUnit::Microsecond, timestamp) => SQLDataType::Timestamp(
                 None,
                 match timestamp {
                     None => TimezoneInfo::WithoutTimeZone,
                     Some(_) => {
-                        return Err(ParadeError::NotSupported("timestamp with time zone".into()))
+                        return Err(ParadeError::NotSupported(
+                            "timestamp with time zone".to_string(),
+                        ))
                     }
                 },
             ),
             DataType::Date32 => SQLDataType::Date,
-            _ => return Err(ParadeError::NotFound),
+            _ => return Err(ParadeError::DataTypeNotSupported(self.to_string())),
         };
 
         Ok(result)
@@ -90,15 +93,17 @@
 
                 DataType::Decimal128(casted_precision, casted_scale)
             }
-            SQLDataType::Time(_, _) => return Err(ParadeError::NotSupported("time".into())),
+            SQLDataType::Time(_, _) => return Err(ParadeError::NotSupported("time".to_string())),
             SQLDataType::Timestamp(_, TimezoneInfo::WithoutTimeZone) => {
                 DataType::Timestamp(TimeUnit::Microsecond, None)
             }
             SQLDataType::Timestamp(_, TimezoneInfo::WithTimeZone) => {
-                return Err(ParadeError::NotSupported("timestamp with time zone".into()))
+                return Err(ParadeError::NotSupported(
+                    "timestamp with time zone".to_string(),
+                ))
             }
             SQLDataType::Date => DataType::Date32,
-            _ => return Err(ParadeError::NotFound),
+            _ => return Err(ParadeError::DataTypeNotSupported(sql_data_type.to_string())),
         };
 
         Ok(result)
@@ -126,26 +131,38 @@
                 PgBuiltInOids::OIDOID | PgBuiltInOids::XIDOID => SQLDataType::UnsignedInt4(None),
                 PgBuiltInOids::FLOAT4OID => SQLDataType::Float4,
                 PgBuiltInOids::FLOAT8OID => SQLDataType::Float8,
-                PgBuiltInOids::TIMEOID => return Err(ParadeError::NotSupported("time".into())),
+                PgBuiltInOids::TIMEOID => {
+                    return Err(ParadeError::NotSupported("time".to_string()))
+                }
                 PgBuiltInOids::TIMESTAMPOID => {
                     SQLDataType::Timestamp(None, TimezoneInfo::WithoutTimeZone)
                 }
                 PgBuiltInOids::DATEOID => SQLDataType::Date,
                 PgBuiltInOids::TIMETZOID => {
-                    return Err(ParadeError::NotSupported("time with time zone".into()))
+                    return Err(ParadeError::NotSupported("time with time zone".to_string()))
                 }
                 PgBuiltInOids::TIMESTAMPTZOID => {
-                    return Err(ParadeError::NotSupported("timestamp with time zone".into()))
+                    return Err(ParadeError::NotSupported(
+                        "timestamp with time zone".to_string(),
+                    ))
+                }
+                PgBuiltInOids::JSONOID => {
+                    return Err(ParadeError::NotSupported("json".to_string()))
+                }
+                PgBuiltInOids::JSONBOID => {
+                    return Err(ParadeError::NotSupported("jsonb".to_string()))
+                }
+                PgBuiltInOids::BOXOID => return Err(ParadeError::NotSupported("box".to_string())),
+                PgBuiltInOids::POINTOID => {
+                    return Err(ParadeError::NotSupported("point".to_string()))
                 }
-                PgBuiltInOids::JSONOID => return Err(ParadeError::NotSupported("json".into())),
-                PgBuiltInOids::JSONBOID => return Err(ParadeError::NotSupported("jsonb".into())),
-                PgBuiltInOids::BOXOID => return Err(ParadeError::NotSupported("box".into())),
-                PgBuiltInOids::POINTOID => return Err(ParadeError::NotSupported("point".into())),
-                PgBuiltInOids::TIDOID => return Err(ParadeError::NotSupported("tid".into())),
+                PgBuiltInOids::TIDOID => return Err(ParadeError::NotSupported("tid".to_string())),
                 PgBuiltInOids::CSTRINGOID => {
-                    return Err(ParadeError::NotSupported("cstring".into()))
+                    return Err(ParadeError::NotSupported("cstring".to_string()))
+                }
+                PgBuiltInOids::INETOID => {
+                    return Err(ParadeError::NotSupported("inet".to_string()))
                 }
-                PgBuiltInOids::INETOID => return Err(ParadeError::NotSupported("inet".into())),
                 PgBuiltInOids::NUMERICOID => {
                     let scale: i32 = (((typmod - VARHDRSZ as i32) & 0x7ff) ^ 1024) - 1024;
                     let precision: i32 = ((typmod - VARHDRSZ as i32) >> 16) & 0xffff;
@@ -168,31 +185,36 @@
                         scale as u64,
                     ))
                 }
-                PgBuiltInOids::VOIDOID => return Err(ParadeError::NotSupported("void".into())),
+                PgBuiltInOids::VOIDOID => {
+                    return Err(ParadeError::NotSupported("void".to_string()))
+                }
                 PgBuiltInOids::INT4RANGEOID => {
-                    return Err(ParadeError::NotSupported("int4 range".into()))
+                    return Err(ParadeError::NotSupported("int4 range".to_string()))
                 }
                 PgBuiltInOids::INT8RANGEOID => {
-                    return Err(ParadeError::NotSupported("int8 range".into()))
+                    return Err(ParadeError::NotSupported("int8 range".to_string()))
                 }
                 PgBuiltInOids::NUMRANGEOID => {
-                    return Err(ParadeError::NotSupported("numeric range".into()))
+                    return Err(ParadeError::NotSupported("numeric range".to_string()))
                 }
                 PgBuiltInOids::DATERANGEOID => {
-                    return Err(ParadeError::NotSupported("date range".into()))
+                    return Err(ParadeError::NotSupported("date range".to_string()))
                 }
                 PgBuiltInOids::TSRANGEOID => {
-                    return Err(ParadeError::NotSupported("timestamp range".into()))
+                    return Err(ParadeError::NotSupported("timestamp range".to_string()))
                 }
                 PgBuiltInOids::TSTZRANGEOID => {
                     return Err(ParadeError::NotSupported(
-                        "timestamp with time zone range".into(),
+                        "timestamp with time zone range".to_string(),
                     ))
                 }
                 PgBuiltInOids::UUIDOID => SQLDataType::Uuid,
-                _ => return Err(ParadeError::NotFound),
+                _ => return Err(ParadeError::DataTypeNotSupported("unknown".to_string())),
             },
-            _ => return Err(ParadeError::NotFound),
+            PgOid::Invalid => return Err(ParadeError::DataTypeNotSupported("invalid".to_string())),
+            PgOid::Custom(_) => {
+                return Err(ParadeError::DataTypeNotSupported("custom".to_string()))
+            }
         };
 
         Ok(result)
@@ -212,13 +234,15 @@
             SQLDataType::Numeric(ExactNumberInfo::PrecisionAndScale(_precision, _scale)) => {
                 PgBuiltInOids::NUMERICOID
             }
-            SQLDataType::Time(_, _) => return Err(ParadeError::NotSupported("time".into())),
+            SQLDataType::Time(_, _) => return Err(ParadeError::NotSupported("time".to_string())),
             SQLDataType::Timestamp(_, TimezoneInfo::WithoutTimeZone) => PgBuiltInOids::TIMESTAMPOID,
             SQLDataType::Date => PgBuiltInOids::DATEOID,
             SQLDataType::Timestamp(_, TimezoneInfo::WithTimeZone) => {
-                return Err(ParadeError::NotSupported("timestamp with time zone".into()))
+                return Err(ParadeError::NotSupported(
+                    "timestamp with time zone".to_string(),
+                ))
             }
-            _ => return Err(ParadeError::NotFound),
+            _ => return Err(ParadeError::DataTypeNotSupported(sql_data_type.to_string())),
         };
 
         let typmod: i32 = match sql_data_type {
@@ -248,7 +272,7 @@
             pg_sys::numeric,
             &[anynumeric.into_datum(), original_typmod.into_datum()],
         )
-        .ok_or_else(|| ParadeError::Generic("numeric direct function call failed".into()))?
+        .ok_or_else(|| ParadeError::Generic("numeric direct function call failed".to_string()))?
     };
 
     // Scale the anynumeric up or down
@@ -271,7 +295,7 @@
             pg_sys::numeric,
             &[scaled_anynumeric.into_datum(), target_typmod.into_datum()],
         )
-        .ok_or_else(|| ParadeError::Generic("numeric direct function call failed".into()))
+        .ok_or_else(|| ParadeError::Generic("numeric direct function call failed".to_string()))
     }
 }
 
@@ -410,24 +434,13 @@
                     if is_null {
                         vec.push(None);
                     } else {
-                        vec.push(unsafe {
-                            match AnyNumeric::from_datum(*datum, false) {
-                                Some(numeric) => {
-                                    let ret = scale_anynumeric(
-                                        numeric,
-                                        precision as i32,
-                                        scale as i32,
-                                        true,
-                                    )?;
-                                    let val = i128::try_from(ret);
-                                    match val {
-                                        Ok(val) => Some(val),
-                                        Err(e) => return Err(ParadeError::Generic(e.to_string())),
-                                    }
-                                }
-                                None => return Err(ParadeError::NotFound),
-                            }
-                        });
+                        let numeric = unsafe {
+                            AnyNumeric::from_datum(*datum, false)
+                                .ok_or(ParadeError::DatumNotFound("numeric".to_string()))?
+                        };
+                        let numeric_with_scale =
+                            scale_anynumeric(numeric, precision as i32, scale as i32, true)?;
+                        vec.push(Some(i128::try_from(numeric_with_scale)?));
                     }
                 }
                 Ok(Arc::new(
@@ -480,7 +493,9 @@
                 }
                 Ok(Arc::new(Date32Array::from(vec)))
             }
-            _ => Err(ParadeError::NotFound),
+            _ => Err(ParadeError::DataTypeNotSupported(
+                datafusion_type.to_string(),
+            )),
         }
     }
 
@@ -495,13 +510,17 @@
             DataType::Boolean => array
                 .as_any()
                 .downcast_ref::<BooleanArray>()
-                .ok_or_else(|| ParadeError::NotFound)?
+                .ok_or(ParadeError::NoneError(
+                    type_name::<BooleanArray>().to_string(),
+                ))?
                 .value(index)
                 .into_datum(),
             DataType::Utf8 => array
                 .as_any()
                 .downcast_ref::<StringArray>()
-                .ok_or_else(|| ParadeError::NotFound)?
+                .ok_or(ParadeError::NoneError(
+                    type_name::<StringArray>().to_string(),
+                ))?
                 .value(index)
                 .into_datum(),
             DataType::Int16 => array.as_primitive::<Int16Type>().value(index).into_datum(),
@@ -523,18 +542,24 @@
                 ret.into_datum()
             }
             DataType::Time64(TimeUnit::Microsecond) => {
-                return Err(ParadeError::NotSupported("time".into()));
+                return Err(ParadeError::NotSupported("time".to_string()));
             }
             DataType::Timestamp(TimeUnit::Microsecond, None) => array
                 .as_primitive::<TimestampMicrosecondType>()
                 .value(index)
                 .into_datum(),
             DataType::Timestamp(TimeUnit::Microsecond, Some(_)) => {
-                return Err(ParadeError::NotSupported("timestamp with time zone".into()));
+                return Err(ParadeError::NotSupported(
+                    "timestamp with time zone".to_string(),
+                ));
             }
             DataType::Date32 => array.as_primitive::<Date32Type>().value(index).into_datum(),
-            _ => return Err(ParadeError::NotFound),
+            _ => {
+                return Err(ParadeError::DataTypeNotSupported(
+                    datafusion_type.to_string(),
+                ))
+            }
         }
-        .ok_or_else(|| ParadeError::NotFound)
+        .ok_or(ParadeError::DatumNotFound(datafusion_type.to_string()))
     }
 }