Finalize CTable varlen scalar support

 - Add final validation for vlstring/vlbytes schema options
 - Tag BatchArray backends with ctable_varlen_scalar metadata
 - Validate varlen scalar backend metadata on reopen
 - Fix CTable constructor reopen path for vlstring/vlbytes columns
 - Add explicit lazy/query guard errors for varlen scalar columns
 - Extend vlstring/vlbytes persistence and guard tests
 - Confirm full pytest suite passes

Author: FrancescAlted

src/blosc2/ctable.py

@@ -936,9 +936,19 @@ def size(self) -> int:
         """Number of live values in the column."""
         return len(self)
 
+    def _ensure_queryable(self) -> None:
+        if self.is_varlen_scalar:
+            raise NotImplementedError(
+                f"Column {self._col_name!r} is a vlstring/vlbytes column; "
+                "lazy expressions and vectorized comparisons are not supported yet."
+            )
+
     @staticmethod
     def _unwrap_operand(other):
-        return other._raw_col if isinstance(other, Column) else other
+        if isinstance(other, Column):
+            other._ensure_queryable()
+            return other._raw_col
+        return other
 
     @property
     def _is_nullable_bool(self) -> bool:
@@ -950,99 +960,129 @@ def _is_nullable_bool(self) -> bool:
         )
 
     def __neg__(self):
+        self._ensure_queryable()
         return -self._raw_col
 
     def __pos__(self):
+        self._ensure_queryable()
         return +self._raw_col
 
     def __abs__(self):
+        self._ensure_queryable()
         return abs(self._raw_col)
 
     def __add__(self, other):
+        self._ensure_queryable()
         return self._raw_col + self._unwrap_operand(other)
 
     def __radd__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) + self._raw_col
 
     def __sub__(self, other):
+        self._ensure_queryable()
         return self._raw_col - self._unwrap_operand(other)
 
     def __rsub__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) - self._raw_col
 
     def __mul__(self, other):
+        self._ensure_queryable()
         return self._raw_col * self._unwrap_operand(other)
 
     def __rmul__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) * self._raw_col
 
     def __truediv__(self, other):
+        self._ensure_queryable()
         return self._raw_col / self._unwrap_operand(other)
 
     def __rtruediv__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) / self._raw_col
 
     def __floordiv__(self, other):
+        self._ensure_queryable()
         return self._raw_col // self._unwrap_operand(other)
 
     def __rfloordiv__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) // self._raw_col
 
     def __mod__(self, other):
+        self._ensure_queryable()
         return self._raw_col % self._unwrap_operand(other)
 
     def __rmod__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) % self._raw_col
 
     def __pow__(self, other):
+        self._ensure_queryable()
         return self._raw_col ** self._unwrap_operand(other)
 
     def __rpow__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) ** self._raw_col
 
     def __and__(self, other):
+        self._ensure_queryable()
         return self._raw_col & self._unwrap_operand(other)
 
     def __rand__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) & self._raw_col
 
     def __or__(self, other):
+        self._ensure_queryable()
         return self._raw_col | self._unwrap_operand(other)
 
     def __ror__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) | self._raw_col
 
     def __xor__(self, other):
+        self._ensure_queryable()
         return self._raw_col ^ self._unwrap_operand(other)
 
     def __rxor__(self, other):
+        self._ensure_queryable()
         return self._unwrap_operand(other) ^ self._raw_col
 
     def __invert__(self):
+        self._ensure_queryable()
         if self._is_nullable_bool:
             return self._raw_col == 0
         return ~self._raw_col
 
     def __lt__(self, other):
+        self._ensure_queryable()
         return self._raw_col < self._unwrap_operand(other)
 
     def __le__(self, other):
+        self._ensure_queryable()
         return self._raw_col <= self._unwrap_operand(other)
 
     def __eq__(self, other):
+        self._ensure_queryable()
         if self._is_nullable_bool and isinstance(other, (bool, np.bool_)):
             return self._raw_col == int(other)
         return self._raw_col == self._unwrap_operand(other)
 
     def __ne__(self, other):
+        self._ensure_queryable()
         if self._is_nullable_bool and isinstance(other, (bool, np.bool_)):
             return self._raw_col == int(not other)
         return self._raw_col != self._unwrap_operand(other)
 
     def __gt__(self, other):
+        self._ensure_queryable()
         return self._raw_col > self._unwrap_operand(other)
 
     def __ge__(self, other):
+        self._ensure_queryable()
         return self._raw_col >= self._unwrap_operand(other)
 
     @property
@@ -1598,6 +1638,8 @@ def __init__(
                 cc = self._schema.columns_by_name[name]
                 if self._is_list_column(cc):
                     col = storage.open_list_column(name)
+                elif self._is_varlen_scalar_column(cc):
+                    col = storage.open_varlen_scalar_column(name, cc.spec)
                 else:
                     col = storage.open_column(name)
                 self._cols[name] = col
@@ -5656,6 +5698,16 @@ def _where_expression_operands(self) -> dict[str, blosc2.NDArray | blosc2.LazyEx
         operands.update({name: cc["lazy"] for name, cc in self._computed_cols.items()})
         return operands
 
+    def _guard_varlen_scalar_expression(self, expr: str) -> None:
+        for col in self._schema.columns:
+            if self._is_varlen_scalar_column(col) and re.search(
+                rf"(?<!\w){re.escape(col.name)}(?!\w)", expr
+            ):
+                raise NotImplementedError(
+                    f"Column {col.name!r} is a vlstring/vlbytes column; "
+                    "lazy expressions are not supported yet."
+                )
+
     def where(
         self,
         expr_result: str | np.ndarray | blosc2.NDArray | blosc2.LazyExpr | Column,
@@ -5735,6 +5787,7 @@ def where(
             t.where(not t.returned)
         """
         if isinstance(expr_result, str):
+            self._guard_varlen_scalar_expression(expr_result)
             expr_result = blosc2.lazyexpr(expr_result, self._where_expression_operands())
         if isinstance(expr_result, np.ndarray) and expr_result.dtype == np.bool_:
             expr_result = blosc2.asarray(expr_result)

src/blosc2/ctable_storage.py

@@ -33,6 +33,7 @@
     _make_persistent_backend,
     _open_persistent_backend,
     _ScalarVarLenArray,
+    _validate_role_metadata,
 )
 from blosc2.schunk import process_opened_object
 
@@ -396,7 +397,8 @@ def open_varlen_scalar_column(self, name: str, spec) -> _ScalarVarLenArray:
                 )
             )
         else:
-            backend = _open_persistent_backend(path, self._mode)
+            backend = _open_persistent_backend(path, self._mode, spec=spec)
+        _validate_role_metadata(backend, spec)
         return _ScalarVarLenArray(spec, backend)
 
     def create_valid_rows(self, *, shape, chunks, blocks):

src/blosc2/scalar_array.py

@@ -33,35 +33,74 @@
 _CTABLE_VARLEN_SCALAR_META_KEY = "ctable_varlen_scalar"
 
 
+def _role_metadata_for_spec(spec) -> dict[str, Any]:
+    """Return the fixed metadata role tag for a CTable varlen scalar backend."""
+    return {
+        "version": 1,
+        "py_type": "str" if spec.python_type is str else "bytes",
+        "nullable": bool(getattr(spec, "nullable", False)),
+        "batch_rows": getattr(spec, "batch_rows", 2048),
+    }
+
+
+def _storage_with_role_meta(spec, **storage_kwargs: Any):
+    import blosc2
+
+    storage = blosc2.Storage(**storage_kwargs)
+    fixed_meta = dict(storage.meta or {})
+    fixed_meta[_CTABLE_VARLEN_SCALAR_META_KEY] = _role_metadata_for_spec(spec)
+    storage.meta = fixed_meta
+    return storage
+
+
+def _validate_role_metadata(backend: BatchArray, spec) -> None:
+    """Validate the optional CTable varlen scalar role tag on an opened backend."""
+    meta = backend.schunk.meta
+    if _CTABLE_VARLEN_SCALAR_META_KEY not in meta:
+        # Older local artifacts may only have the BatchArray tag; the CTable schema
+        # still identifies the logical role, so keep reopen tolerant.
+        return
+    role = meta[_CTABLE_VARLEN_SCALAR_META_KEY]
+    expected_py_type = "str" if spec.python_type is str else "bytes"
+    if role.get("py_type") != expected_py_type:
+        raise ValueError(
+            f"Varlen scalar backend type mismatch: expected {expected_py_type!r}, "
+            f"found {role.get('py_type')!r}."
+        )
+
+
 def _make_backend(spec) -> BatchArray:
     """Create a fresh in-memory BatchArray for a varlen scalar spec."""
+    storage = _storage_with_role_meta(spec)
     return BatchArray(
+        storage=storage,
         items_per_block=getattr(spec, "items_per_block", None),
         serializer=getattr(spec, "serializer", "msgpack"),
     )
 
 
 def _make_persistent_backend(spec, urlpath: str, mode: str, *, cparams=None, dparams=None) -> BatchArray:
     """Create or open a persistent BatchArray for a varlen scalar spec."""
-    kwargs: dict[str, Any] = {
-        "urlpath": urlpath,
-        "mode": mode,
-        "contiguous": True,
-    }
+    kwargs: dict[str, Any] = {}
     if cparams is not None:
         kwargs["cparams"] = cparams
     if dparams is not None:
         kwargs["dparams"] = dparams
+    storage = _storage_with_role_meta(spec, urlpath=urlpath, mode=mode, contiguous=True)
     return BatchArray(
+        storage=storage,
         items_per_block=getattr(spec, "items_per_block", None),
         serializer=getattr(spec, "serializer", "msgpack"),
         **kwargs,
     )
 
 
-def _open_persistent_backend(urlpath: str, mode: str) -> BatchArray:
+def _open_persistent_backend(urlpath: str, mode: str, spec=None) -> BatchArray:
     """Reopen an existing persistent BatchArray (any mode)."""
-    return BatchArray(urlpath=urlpath, mode=mode)
+    backend = BatchArray(urlpath=urlpath, mode=mode)
+    if spec is not None:
+        _validate_role_metadata(backend, spec)
+    return backend
 
 
 class _ScalarVarLenArray:

src/blosc2/schema.py

@@ -515,6 +515,12 @@ def __init__(
         batch_rows: int | None = 2048,
         items_per_block: int | None = None,
     ):
+        if serializer != "msgpack":
+            raise ValueError("vlstring currently only supports serializer='msgpack'")
+        if batch_rows is not None and batch_rows <= 0:
+            raise ValueError("batch_rows must be positive or None")
+        if items_per_block is not None and items_per_block <= 0:
+            raise ValueError("items_per_block must be positive or None")
         self.nullable = nullable
         self.serializer = serializer
         self.batch_rows = batch_rows
@@ -568,6 +574,12 @@ def __init__(
         batch_rows: int | None = 2048,
         items_per_block: int | None = None,
     ):
+        if serializer != "msgpack":
+            raise ValueError("vlbytes currently only supports serializer='msgpack'")
+        if batch_rows is not None and batch_rows <= 0:
+            raise ValueError("batch_rows must be positive or None")
+        if items_per_block is not None and items_per_block <= 0:
+            raise ValueError("items_per_block must be positive or None")
         self.nullable = nullable
         self.serializer = serializer
         self.batch_rows = batch_rows

tests/ctable/test_vlstring_vlbytes.py

@@ -31,6 +31,11 @@ def test_vlstring_spec_defaults():
     assert spec.python_type is str
 
 
+def test_vlstring_rejects_unsupported_serializer():
+    with pytest.raises(ValueError, match="serializer='msgpack'"):
+        blosc2.vlstring(serializer="arrow")
+
+
 def test_vlbytes_spec_defaults():
     spec = blosc2.vlbytes()
     assert spec.nullable is False
@@ -382,6 +387,32 @@ def test_ctable_vlstring_save_load(tmp_path):
     ct2.close()
 
 
+def test_ctable_vlstring_backend_role_metadata(tmp_path):
+    urlpath = str(tmp_path / "vl_meta.b2d")
+    ct = blosc2.CTable(VLRow, new_data=ROWS[:2], urlpath=urlpath, mode="w")
+    ct.close()
+
+    backend = blosc2.open(str(tmp_path / "vl_meta.b2d" / "_cols" / "text.b2b"), mode="r")
+    assert backend.schunk.meta["ctable_varlen_scalar"] == {
+        "version": 1,
+        "py_type": "str",
+        "nullable": False,
+        "batch_rows": 2048,
+    }
+
+
+def test_ctable_constructor_reopens_vlstring_persistent_table(tmp_path):
+    urlpath = str(tmp_path / "vl_ctor_reopen.b2d")
+    ct = blosc2.CTable(VLRow, new_data=ROWS[:2], urlpath=urlpath, mode="w")
+    ct.close()
+
+    reopened = blosc2.CTable(VLRow, urlpath=urlpath, mode="a")
+    assert reopened.text[1] == ROWS[1][1]
+    reopened.append((42, "ctor append", b"ctor bytes"))
+    assert reopened.text[2] == "ctor append"
+    reopened.close()
+
+
 def test_ctable_vlstring_save_reload_b2z(tmp_path):
     urlpath = str(tmp_path / "vl_test.b2z")
     ct = blosc2.CTable(VLRow, new_data=ROWS, urlpath=urlpath, mode="w")
@@ -479,6 +510,14 @@ def test_ctable_vlstring_sort_raises():
         next(gen)
 
 
+def test_ctable_vlstring_lazy_expression_raises():
+    ct = blosc2.CTable(VLRow, new_data=ROWS)
+    with pytest.raises(NotImplementedError, match="vlstring/vlbytes"):
+        ct.where('text == "hello world"')
+    with pytest.raises(NotImplementedError, match="vlstring/vlbytes"):
+        _ = ct.text == "hello world"
+
+
 def test_ctable_vlstring_build_index_raises(tmp_path):
     urlpath = str(tmp_path / "vl_idx.b2d")
     ct = blosc2.CTable(VLRow, new_data=ROWS, urlpath=urlpath, mode="w")