feat: for Postgres and Qdrant, show notes when vector is stored as json (#794)

Author: georgeh0

examples/face_recognition/main.py

@@ -75,7 +75,7 @@ def extract_faces(content: bytes) -> list[FaceBase]:
 @cocoindex.op.function(cache=True, behavior_version=1, gpu=True)
 def extract_face_embedding(
     face: bytes,
-) -> cocoindex.Vector[cocoindex.Float32, typing.Literal[128]]:
+) -> cocoindex.Vector[cocoindex.Float32]:
     """Extract the embedding of a face."""
     img = Image.open(io.BytesIO(face)).convert("RGB")
     embedding = face_recognition.face_encodings(

src/ops/targets/postgres.rs

@@ -426,12 +426,41 @@ pub struct TableSetupAction {
 pub struct SetupStatus {
     create_pgvector_extension: bool,
     actions: TableSetupAction,
+    vector_as_jsonb_columns: Vec<(String, ValueType)>,
 }
 
 impl SetupStatus {
     fn new(desired_state: Option<SetupState>, existing: setup::CombinedState<SetupState>) -> Self {
         let table_action =
             TableMainSetupAction::from_states(desired_state.as_ref(), &existing, false);
+        let vector_as_jsonb_columns = desired_state
+            .as_ref()
+            .iter()
+            .flat_map(|s| {
+                s.columns.value_columns.iter().filter_map(|(name, schema)| {
+                    if let ValueType::Basic(BasicValueType::Vector(vec_schema)) = schema
+                        && !convertible_to_pgvector(vec_schema)
+                    {
+                        let is_touched = match &table_action.table_upsertion {
+                            Some(TableUpsertionAction::Create { values, .. }) => {
+                                values.contains_key(name)
+                            }
+                            Some(TableUpsertionAction::Update {
+                                columns_to_upsert, ..
+                            }) => columns_to_upsert.contains_key(name),
+                            None => false,
+                        };
+                        if is_touched {
+                            Some((name.clone(), schema.clone()))
+                        } else {
+                            None
+                        }
+                    } else {
+                        None
+                    }
+                })
+            })
+            .collect::<Vec<_>>();
         let (indexes_to_delete, indexes_to_create) = desired_state
             .as_ref()
             .map(|desired| {
@@ -469,6 +498,7 @@ impl SetupStatus {
                 indexes_to_delete,
                 indexes_to_create,
             },
+            vector_as_jsonb_columns,
         }
     }
 }
@@ -506,6 +536,13 @@ fn describe_index_spec(index_name: &str, index_spec: &VectorIndexDef) -> String
 impl setup::ResourceSetupStatus for SetupStatus {
     fn describe_changes(&self) -> Vec<setup::ChangeDescription> {
         let mut descriptions = self.actions.table_action.describe_changes();
+        for (column_name, schema) in self.vector_as_jsonb_columns.iter() {
+            descriptions.push(setup::ChangeDescription::Note(format!(
+                "Field `{}` has type `{}`. Only number vector with fixed size is supported by pgvector. It will be stored as `jsonb`.",
+                column_name,
+                schema
+            )));
+        }
         if self.create_pgvector_extension {
             descriptions.push(setup::ChangeDescription::Action(
                 "Create pg_vector extension (if not exists)".to_string(),

src/ops/targets/qdrant.rs

@@ -95,6 +95,9 @@ struct VectorDef {
 struct SetupState {
     #[serde(default)]
     vectors: BTreeMap<String, VectorDef>,
+
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    unsupported_vector_fields: Vec<(String, ValueType)>,
 }
 
 #[derive(Debug)]
@@ -131,6 +134,12 @@ impl setup::ResourceSetupStatus for SetupStatus {
                     format!(" with vectors: {vector_descriptions}")
                 }
             )));
+            for (name, schema) in add_collection.unsupported_vector_fields.iter() {
+                result.push(setup::ChangeDescription::Note(format!(
+                    "Field `{}` has type `{}`. Only number vector with fixed size is supported by Qdrant. It will be stored in payload.",
+                    name, schema
+                )));
+            }
         }
         result
     }
@@ -311,6 +320,7 @@ impl StorageFactoryBase for Factory {
 
                 let mut fields_info = Vec::<FieldInfo>::new();
                 let mut vector_def = BTreeMap::<String, VectorDef>::new();
+                let mut unsupported_vector_fields = Vec::<(String, ValueType)>::new();
 
                 for field in d.value_fields_schema.iter() {
                     let vector_size = parse_supported_vector_size(&field.value_type.typ);
@@ -326,6 +336,12 @@ impl StorageFactoryBase for Factory {
                                 metric: DEFAULT_VECTOR_SIMILARITY_METRIC,
                             },
                         );
+                    } else if matches!(
+                        &field.value_type.typ,
+                        schema::ValueType::Basic(schema::BasicValueType::Vector(_))
+                    ) {
+                        // This is a vector field but not supported by Qdrant
+                        unsupported_vector_fields.push((field.name.clone(), field.value_type.typ.clone()));
                     }
                 }
 
@@ -370,6 +386,7 @@ impl StorageFactoryBase for Factory {
                     },
                     desired_setup_state: SetupState {
                         vectors: vector_def,
+                        unsupported_vector_fields,
                     },
                 })
             })
@@ -398,16 +415,12 @@ impl StorageFactoryBase for Factory {
         _auth_registry: &Arc<AuthRegistry>,
     ) -> Result<Self::SetupStatus> {
         let desired_exists = desired.is_some();
-        let add_collection = desired
-            .filter(|state| {
-                !existing.always_exists()
-                    || existing
-                        .possible_versions()
-                        .any(|v| v.vectors != state.vectors)
-            })
-            .map(|state| SetupState {
-                vectors: state.vectors,
-            });
+        let add_collection = desired.filter(|state| {
+            !existing.always_exists()
+                || existing
+                    .possible_versions()
+                    .any(|v| v.vectors != state.vectors)
+        });
         let delete_collection = existing.possible_versions().next().is_some()
             && (!desired_exists || add_collection.is_some());
         Ok(SetupStatus {

src/setup/states.rs

@@ -305,16 +305,21 @@ impl<K, S, C: ResourceSetupStatus> std::fmt::Display for ResourceSetupInfo<K, S,
             let changes = setup_status.describe_changes();
             if !changes.is_empty() {
                 let mut f = indented(f).with_str(INDENT);
-                writeln!(f, "{}", "TODO:".color(AnsiColors::BrightBlack))?;
+                writeln!(f, "")?;
                 for change in changes {
                     match change {
                         ChangeDescription::Action(action) => {
-                            writeln!(f, "  - {}", action.color(AnsiColors::BrightBlack))?;
+                            writeln!(
+                                f,
+                                "{} {}",
+                                "TODO:".color(AnsiColors::BrightBlack).bold(),
+                                action.color(AnsiColors::BrightBlack)
+                            )?;
                         }
                         ChangeDescription::Note(note) => {
                             writeln!(
                                 f,
-                                "  {} {}",
+                                "{} {}",
                                 "NOTE:".color(AnsiColors::Yellow).bold(),
                                 note.color(AnsiColors::Yellow)
                             )?;