Add shared sparse vector support

dev/agent-skills/target-connector/SKILL.md

@@ -217,6 +217,37 @@ _shared_sink = coco.TargetActionSink.from_fn(_apply_actions)
 
 When building queries from user-provided names (table, column, index) or values (record IDs, keys), you must guard against injection and ensure correctness. See [input_safety.md](input_safety.md) for patterns on identifier validation, parameterized queries, and value escaping.
 
+### Vector Schema Resolution
+
+Table-like connectors that infer dense or sparse vector columns from annotations must use the shared connector-kit resolver:
+
+```python
+from cocoindex.connectorkits import resolve_vector_schemas
+
+schemas = await resolve_vector_schemas(
+    type_info.base_type,
+    all_annotations,
+    # Include this only when the connector supports dense but not sparse vectors.
+    reject_sparse_vectors_for="MyConnector",
+)
+```
+
+The resolver gives direct annotations precedence over `ContextKey` providers, detects dense/sparse conflicts, and validates that `SparseVectorSchema` is attached only to a `SparseVector` field. Do not duplicate these checks in the connector.
+
+Connectors that do not consume vector metadata but still need to reject sparse columns should use the non-resolving guard, which deliberately ignores `ContextKey` annotations:
+
+```python
+from cocoindex.connectorkits import reject_sparse_vectors
+
+reject_sparse_vectors(
+    type_info.base_type,
+    all_annotations,
+    connector_name="MyConnector",
+)
+```
+
+Do not add `SparseVectorSchemaProvider` to a public override union when the connector rejects sparse vectors; annotations remain runtime-guarded without advertising unsupported input.
+
 ## Completion Checklist
 
 After implementing the connector code, complete these additional steps:

docs/src/content/docs/common_resources/vector_schema.mdx

@@ -1,9 +1,8 @@
 ---
 title: "*Vector schema* annotations"
 description: >
-  Describe vector columns with VectorSchema and VectorSchemaProvider. Covers
-  the three annotation patterns (ContextKey, embedder instance, explicit
-  schema) and MultiVectorSchema for models like ColBERT.
+  Describe dense, sparse, and multi-vector columns with shared schema types and
+  provider protocols.
 ---
 The schema module (`cocoindex.resources.schema`) defines types that describe vector columns. CocoIndex connectors use these to automatically configure the correct column type (e.g., `vector(384)` in Postgres, `fixed_size_list<float32>(384)` in LanceDB).
 
@@ -119,3 +118,39 @@ multi_schema = MultiVectorSchema(
     vector_schema=VectorSchema(dtype=np.dtype(np.float32), size=128)
 )
 ```
+
+## SparseVector / SparseVectorSchema
+
+`SparseVector` is the canonical value shared by connectors that support sparse vectors. It stores parallel tuples of 0-based integer indices and finite float weights. Indices must be non-negative, sorted ascending, and unique. Direct construction accepts only that canonical tuple/int/float representation. Use `from_arrays()` to coerce parallel numeric iterables and sort them by index; `from_mapping()` delegates to the same normalization path. `as_sparse_vector()` accepts either an already-canonical value or a mapping.
+
+```python
+from cocoindex.resources.schema import SparseVector
+
+sparse = SparseVector.from_mapping({7: 0.9, 1: 0.5})
+assert sparse.indices == (1, 7)
+assert sparse.values == (0.5, 0.9)
+
+sparse = SparseVector.from_arrays(encoder.indices, encoder.values)
+assert sparse.indices == tuple(sorted(encoder.indices))
+```
+
+`SparseVectorSchema` carries optional dimensionality. Sparse values use float32 across CocoIndex connectors. When supplied, `size` must be positive. The dimension is required by Postgres `sparsevec(n)` but optional for backends such as zvec and Qdrant.
+
+```python
+from dataclasses import dataclass
+from typing import Annotated
+
+from cocoindex.resources.schema import SparseVector, SparseVectorSchema
+
+@dataclass
+class SparseDoc:
+    id: str
+    sparse: Annotated[
+        SparseVector,
+        SparseVectorSchema(size=50_000),
+    ]
+```
+
+`SparseVectorSchemaProvider` is the corresponding provider protocol. Its `__coco_sparse_vector_schema__()` method can be supplied directly or through a `ContextKey`, following the same resolution pattern as `VectorSchemaProvider`.
+
+Metrics, scoring modifiers, index methods, and quantization are connector-specific and therefore stay on connector configuration rather than `SparseVectorSchema`.

docs/src/content/docs/connectors/postgres.mdx

@@ -241,6 +241,8 @@ The actual PostgreSQL index is named `{table_name}__vector__{name}`.
 - `m` — Maximum number of connections per layer (hnsw only).
 - `ef_construction` — Size of the dynamic candidate list for construction (hnsw only).
 
+`sparsevec` columns support only `method="hnsw"`; passing the default `"ivfflat"` raises an error. pgvector limits an HNSW-indexed sparse vector to 1,000 non-zero elements.
+
 **Example:**
 
 ```python
@@ -295,7 +297,9 @@ async def TableSchema.from_class(
     record_type: type[RowT],
     primary_key: list[str],
     *,
-    column_overrides: dict[str, PgType | VectorSchemaProvider] | None = None,
+    column_overrides: dict[
+        str, PgType | VectorSchemaProvider | SparseVectorSchemaProvider
+    ] | None = None,
 ) -> TableSchema[RowT]
 ```
 
@@ -338,6 +342,7 @@ Python types are automatically mapped to PostgreSQL types:
 | `datetime.timedelta` | `interval` |
 | `list`, `dict`, nested structs | `jsonb` |
 | `NDArray` (with vector schema) | `vector(n)` or `halfvec(n)` |
+| `SparseVector` (with sparse vector schema) | `sparsevec(n)` |
 
 :::note[U+0000 (NUL) in strings]
 U+0000 (NUL) is a valid Unicode codepoint, but Postgres cannot store it — neither in `text`-family columns nor inside strings in `jsonb` (the `\u0000` escape is rejected at parse time). CocoIndex automatically strips U+0000 from strings before writing to Postgres, recursively for nested strings and dict keys in `jsonb` payloads. For example, `"Hello\0World"` is written as `"HelloWorld"`.
@@ -379,6 +384,31 @@ schema = postgres.TableSchema(
 
 For `NDArray` fields, a [`VectorSchemaProvider`](../common_resources/vector_schema#vectorschemaprovider) annotation specifies the vector dimension and dtype. The connector has built-in pgvector support and automatically creates the extension when needed. See [Vector Schema](../common_resources/vector_schema#vectorschemaprovider) for the full list of annotation options (`ContextKey`, embedder instance, or explicit `VectorSchema`).
 
+#### Sparse vectors
+
+Postgres requires a dimension for `sparsevec(n)`, and its sparse values are float32. Annotate the shared `SparseVector` type with `SparseVectorSchema(size=...)`; CocoIndex converts its canonical 0-based indices to pgvector's 1-based wire format.
+
+```python
+from dataclasses import dataclass
+from typing import Annotated
+
+from cocoindex.resources.schema import SparseVector, SparseVectorSchema
+
+@dataclass
+class SparseDoc:
+    id: str
+    sparse: Annotated[
+        SparseVector,
+        SparseVectorSchema(size=50_000),
+    ]
+
+schema = await postgres.TableSchema.from_class(SparseDoc, primary_key=["id"])
+table = await postgres.mount_table_target(PG_DB, "sparse_docs", schema)
+table.declare_vector_index(column="sparse", metric="ip", method="hnsw")
+```
+
+Rows may provide a `SparseVector` or an integer-to-float mapping at declaration time; mappings are validated and sorted before encoding. A bare `SparseVector` field or a schema without `size` raises an actionable error.
+
 ### Table schema: explicit column definitions
 
 Define columns directly using `ColumnDef`:

docs/src/content/docs/connectors/qdrant.mdx

@@ -231,7 +231,6 @@ point = qdrant.PointStruct(
 For hybrid retrieval, declare a named dense vector and a named sparse vector in the same collection:
 
 ```python
-from qdrant_client.http import models as qdrant_models
 from cocoindex.resources.schema import VectorSchema
 import numpy as np
 
@@ -246,23 +245,22 @@ schema = await qdrant.CollectionSchema.create(
 )
 ```
 
-Points put both vector types in `PointStruct.vector`. The sparse vector is a native Qdrant `SparseVector`, not payload:
+Points put both vector types in `PointStruct.vector`. Use `qdrant.sparse_vector()` to convert the shared canonical `SparseVector` (or an integer-to-float mapping) to Qdrant's native model:
 
 ```python
 point = qdrant.PointStruct(
-    id=123,
+    id=doc_id,
     vector={
         "dense": dense_embedding.tolist(),
-        "sparse": qdrant_models.SparseVector(
-            indices=sparse_indices,
-            values=sparse_values,
-        ),
+        "sparse": qdrant.sparse_vector({7: 0.9, 1: 0.5}),
     },
     payload={"text": text},
 )
 target.declare_point(point)
 ```
 
+Qdrant does not require sparse dimensionality, so `SparseVectorSchema` is not needed here; a shared [`SparseVector`](../common_resources/vector_schema#sparsevector--sparsevectorschema) annotation on your row type remains useful when the same rows also target Postgres or zvec. Sparse scoring configuration, including `modifier="idf"`, stays on `QdrantSparseVectorDef`.
+
 #### VectorSchemaProvider
 
 The `schema` field of `QdrantVectorDef` accepts a [`VectorSchemaProvider`](../common_resources/vector_schema#vectorschemaprovider), a `ContextKey`, or an explicit `VectorSchema` to specify the vector dimension and dtype. See [Vector Schema](../common_resources/vector_schema#vectorschemaprovider) for details.
@@ -391,8 +389,7 @@ async def app_main() -> None:
 ### Example: dense + sparse hybrid vectors
 
 ```python
-from qdrant_client.http import models as qdrant_models
-from cocoindex.resources.schema import VectorSchema
+from cocoindex.resources.schema import SparseVector, VectorSchema
 import numpy as np
 
 @coco.fn
@@ -416,9 +413,11 @@ async def app_main() -> None:
             id=chunk.id,
             vector={
                 "dense": chunk.dense_embedding.tolist(),
-                "sparse": qdrant_models.SparseVector(
-                    indices=chunk.sparse_indices,
-                    values=chunk.sparse_values,
+                "sparse": qdrant.sparse_vector(
+                    SparseVector.from_arrays(
+                        chunk.sparse_indices,
+                        chunk.sparse_values,
+                    )
                 ),
             },
             payload={"text": chunk.text},
@@ -508,9 +507,8 @@ results = client.query_points(
             limit=100,
         ),
         qdrant_models.Prefetch(
-            query=qdrant_models.SparseVector(
-                indices=query_sparse_indices,
-                values=query_sparse_values,
+            query=qdrant.sparse_vector(
+                dict(zip(query_sparse_indices, query_sparse_values, strict=True))
             ),
             using="sparse",
             limit=100,

docs/src/content/docs/connectors/zvec.mdx

@@ -122,7 +122,11 @@ async def CollectionSchema.from_class(
     record_type: type[RowT],
     primary_key: list[str],
     *,
-    column_overrides: dict[str, ZvecType | ZvecVectorDef | ZvecFtsType | VectorSchemaProvider] | None = None,
+    column_overrides: dict[
+        str,
+        ZvecType | ZvecVectorDef | ZvecFtsType
+        | VectorSchemaProvider | SparseVectorSchemaProvider,
+    ] | None = None,
 ) -> CollectionSchema[RowT]
 ```
 
@@ -175,6 +179,7 @@ Scalar Python types map to zvec field types as follows:
 | `list[str]` / `list[int]` / `list[float]` / `list[bool]` | `ARRAY_STRING` / `ARRAY_INT64` / `ARRAY_DOUBLE` / `ARRAY_BOOL` |
 | other `list`, `dict`, nested structs | `STRING` (JSON) |
 | `NDArray` (with vector schema) | `VECTOR_FP32` (float32) or `VECTOR_FP16` (float16) |
+| `SparseVector` | `SPARSE_VECTOR_FP32` |
 
 Scalar fields get an invert index by default so they can be used in query filters. The primary-key column maps to the document `id` and is not stored as a separate field.
 
@@ -237,15 +242,27 @@ class Doc:
 
 #### Sparse vectors
 
-Mark a `dict[int, float]` field (mapping dimension → weight) as sparse with `ZvecVectorDef(sparse=True)`:
+Use the shared `SparseVector` type. Sparse values use float32. `SparseVectorSchema` is optional; zvec does not require a dimension, so `size` is accepted but ignored.
 
 ```python
+from dataclasses import dataclass
+from typing import Annotated
+
+from cocoindex.resources.schema import SparseVector, SparseVectorSchema
+
 @dataclass
-class Doc:
+class SparseDoc:
     id: str
-    sparse: Annotated[dict[int, float], ZvecVectorDef(sparse=True)]
+    sparse: Annotated[
+        SparseVector,
+        SparseVectorSchema(size=50_000),
+    ]
+
+schema = await zvec.CollectionSchema.from_class(SparseDoc, primary_key=["id"])
 ```
 
+At declaration time, the value may also be an integer-to-float mapping; CocoIndex validates and sorts it before writing. The legacy `Annotated[dict[int, float], ZvecVectorDef(sparse=True)]` spelling remains supported, but `SparseVector` is preferred for portable pipeline code. `ZvecVectorDef` can still accompany the shared type when a connector-specific metric is needed.
+
 ## Full example
 
 ```python

python/cocoindex/_internal/datatype.py

@@ -252,13 +252,15 @@ def analyze_type_info(t: Any, *, nullable: bool = False) -> DataTypeInfo:
         variant = MappingType(key_type=key_type, value_type=elem_type)
     elif base_type in (types.UnionType, typing.Union):
         non_none_types = [arg for arg in type_args if arg not in (None, types.NoneType)]
-        if len(non_none_types) == 0:
-            return analyze_type_info(None)
-
-        if len(non_none_types) == 1:
-            return analyze_type_info(
-                non_none_types[0],
-                nullable=nullable or len(non_none_types) < len(type_args),
+        if len(non_none_types) <= 1:
+            nested_info = analyze_type_info(
+                non_none_types[0] if non_none_types else None,
+                nullable=nullable
+                or bool(non_none_types)
+                and len(non_none_types) < len(type_args),
+            )
+            return nested_info._replace(
+                annotations=nested_info.annotations + annotations
             )
 
         variant = UnionType(variant_types=non_none_types)

python/cocoindex/connectorkits/__init__.py

@@ -4,7 +4,19 @@
 
 from typing import Any
 
-__all__ = ["SingleWatcherGuard", "default_subpath_name"]
+from .vector_schema import (
+    VectorSchemas,
+    reject_sparse_vectors,
+    resolve_vector_schemas,
+)
+
+__all__ = [
+    "SingleWatcherGuard",
+    "VectorSchemas",
+    "default_subpath_name",
+    "reject_sparse_vectors",
+    "resolve_vector_schemas",
+]
 
 
 class SingleWatcherGuard: