numpy_parser: reject undersized vector payloads too, add test coverage

The memcpy fast-path guard in unpack_row() only rejected oversized
payloads (buf.size > arr.stride), which prevents the buffer-overflow
case but let undersized payloads slip through, leaving the remainder
of the row filled with uninitialized/stale heap bytes instead of
raising. Tighten the check to buf.size != arr.stride so both
directions are rejected, and update the error message accordingly,
including the column name for easier debugging.

Add tests covering both the oversized and undersized payload cases,
since neither had test coverage before.

Also revert the arrays = [] / .append() -> [None]*n / index-assign
change in make_arrays(), which was functionally identical diff noise
unrelated to the VectorType/2D-array feature.

Restore ShortType (smallint) to _cqltype_to_numpy: an earlier fix on
this PR removed it from that table entirely to stop the VectorType
fast-path from assuming a fixed 2-byte stride for vint-prefixed vector
elements, but the table is shared with the plain scalar top-level
column dispatch in make_array(), where smallint genuinely *is*
fixed-width (no length prefix). Removing the entry regressed scalar
smallint columns onto the slow object-array fallback. Fix: keep
ShortType in the shared table, and instead guard the VectorType.subtype
call site with `subtype.serial_size() is not None` (the same check
VectorType.serialize()/deserialize() use) before consulting the table,
so only the vector-specific dispatch excludes non-fixed-width subtypes.
This also protects any subtype added to the table in the future.
Add tests for the restored scalar smallint fast path (both make_array()
directly and an end-to-end parse_rows() round-trip).

Also fix tests/unit/test_numpy_parser.py's HAVE_NUMPY gating: a bare
`except ImportError` around the numpy_parser/bytesio/etc. imports
could not distinguish "numpy genuinely not installed" from "the
numpy_parser Cython extension is broken despite numpy being present",
silently skipping the whole test module in both cases. Gate instead on
cassandra.cython_deps.HAVE_CYTHON and HAVE_NUMPY (plus the
VERIFY_CYTHON override), matching the `numpytest` convention in
tests/unit/cython/utils.py, and import unconditionally once those are
satisfied so a broken extension surfaces as a real test failure.

Author: mykaul

cassandra/numpy_parser.pyx

@@ -67,20 +67,41 @@ _cqltype_to_numpy = {
     cqltypes.LongType:          np.dtype('>i8'),
     cqltypes.CounterColumnType: np.dtype('>i8'),
     cqltypes.Int32Type:         np.dtype('>i4'),
+    cqltypes.ShortType:         np.dtype('>i2'),
     cqltypes.FloatType:         np.dtype('>f4'),
     cqltypes.DoubleType:        np.dtype('>f8'),
 }
-# Note: ShortType (smallint) is intentionally absent from this table. Unlike
-# LongType/Int32Type/FloatType/DoubleType/CounterColumnType, ShortType does
-# not override serial_size() in cassandra.cqltypes, so it has no fixed
-# per-element width when serialized *inside* a VectorType: the server
-# vint-prefixes each smallint element rather than encoding it as a plain
-# 2-byte big-endian field. Mapping it here would make the vector fast-path
-# assume a fixed 2-byte stride, which does not match the actual wire format
-# and causes a crash (see unpack_row's stride check) instead of falling back
-# to the safe, slower object-array path used for other subtypes without a
-# fixed width (e.g. UTF8Type). See also ByteType (tinyint), which has the
-# same issue and is likewise absent from this table.
+# This table is consulted from two different call sites in make_array()
+# below, and the two call sites do NOT mean the same thing by "fixed width":
+#
+#   - Scalar top-level columns (`_cqltype_to_numpy[coltype]`): here ShortType
+#     (smallint) genuinely is fixed-width on the wire (a plain 2-byte
+#     big-endian field, no length prefix), so it belongs in this table and
+#     gets the fast masked-array path, exactly like Int32Type/LongType/etc.
+#
+#   - VectorType.subtype dispatch (`_cqltype_to_numpy[subtype]` in the
+#     VectorType branch below): Cassandra/Scylla vint-prefixes each element
+#     *inside* a vector unless the subtype overrides serial_size() with a
+#     fixed value (see cqltypes.VectorType.serialize/deserialize, which
+#     branch on exactly this). ShortType (and ByteType/tinyint) do not
+#     override serial_size() -- it returns None from the _CassandraType
+#     base -- so inside a vector they are NOT fixed-width, even though
+#     ShortType *is* fixed-width as a scalar column. Using this table
+#     directly for a vector subtype without checking serial_size() would
+#     make the vector fast-path assume a fixed 2-byte stride that does not
+#     match the actual wire format, causing a crash (see unpack_row's
+#     stride check) instead of falling back to the safe, slower
+#     object-array path used for other subtypes without a fixed width
+#     (e.g. UTF8Type).
+#
+# Rather than removing ShortType from this shared table (which would also
+# silently break the scalar smallint fast path, since both call sites read
+# from the same dict), the VectorType call site below guards with
+# `subtype.serial_size() is not None` before consulting this table at all.
+# That mirrors the check VectorType itself uses to decide its own wire
+# encoding, so it is the general, authoritative test -- and it also
+# protects any subtype added to this table in the future without requiring
+# every caller to separately remember the vector-vs-scalar distinction.
 
 obj_dtype = np.dtype('O')
 
@@ -123,7 +144,7 @@ def make_arrays(ParseDesc desc, array_size):
             (e.g. this can be fed into pandas.DataFrame)
     """
     array_descs = np.empty((desc.rowsize,), arrDescDtype)
-    arrays = [None] * desc.rowsize
+    arrays = []
 
     for i, coltype in enumerate(desc.coltypes):
         arr = make_array(coltype, array_size)
@@ -136,7 +157,7 @@ def make_arrays(ParseDesc desc, array_size):
         except AttributeError:
             array_descs[i]['mask_ptr'] = 0
             array_descs[i]['mask_stride'] = 1
-        arrays[i] = arr
+        arrays.append(arr)
 
     return array_descs, arrays
 
@@ -151,13 +172,26 @@ def make_array(coltype, array_size):
         # VectorType - create 2D array (rows x vector_dimension)
         vector_size = coltype.vector_size
         subtype = coltype.subtype
-        try:
-            dtype = _cqltype_to_numpy[subtype]
-            a = np.ma.empty((array_size, vector_size), dtype=dtype)
-            a.mask = np.zeros((array_size, vector_size), dtype=bool)
-        except KeyError:
-            # Unsupported vector subtype - fall back to object array
-            a = np.empty((array_size,), dtype=obj_dtype)
+        # Only use the fixed-width fast path if the subtype actually has a
+        # fixed per-element wire size *inside a vector*. subtype.serial_size()
+        # is exactly the check VectorType.serialize()/deserialize() use to
+        # decide whether elements are plain fixed-width fields or
+        # vint-length-prefixed (e.g. ShortType/ByteType return None here,
+        # even though ShortType has its own, unrelated entry in
+        # _cqltype_to_numpy for the scalar-column case below). Checking this
+        # first -- rather than only catching KeyError on _cqltype_to_numpy --
+        # also protects any subtype added to that table in the future.
+        if subtype.serial_size() is not None:
+            try:
+                dtype = _cqltype_to_numpy[subtype]
+                a = np.ma.empty((array_size, vector_size), dtype=dtype)
+                a.mask = np.zeros((array_size, vector_size), dtype=bool)
+                return a
+            except KeyError:
+                pass
+        # Unsupported vector subtype, or one without a fixed per-element
+        # wire width - fall back to object array.
+        a = np.empty((array_size,), dtype=obj_dtype)
         return a
 
     # Scalar types
@@ -189,10 +223,11 @@ cdef inline int unpack_row(
             Py_INCREF(val)
             (<PyObject **> arr.buf_ptr)[0] = <PyObject *> val
         elif buf.size >= 0:
-            if buf.size > arr.stride:
+            if buf.size != arr.stride:
                 raise ValueError(
-                    "Column %d: received %d bytes but array stride is %d" %
-                    (i, buf.size, arr.stride))
+                    "Column %d (%r): received %d bytes but array stride is %d "
+                    "(payload must exactly match the expected element size)" %
+                    (i, desc.colnames[i], buf.size, arr.stride))
             memcpy(<char *> arr.buf_ptr, buf.ptr, buf.size)
         else:
             memset(<char *>arr.mask_ptr, 1, arr.mask_stride)

tests/unit/test_numpy_parser.py

@@ -15,21 +15,44 @@
 import struct
 import unittest
 
+from cassandra.cython_deps import HAVE_CYTHON, HAVE_NUMPY
+
 try:
+    from tests import VERIFY_CYTHON
+except ImportError:
+    VERIFY_CYTHON = False
+
+# cassandra.numpy_parser is a Cython extension (gated on HAVE_CYTHON), not a
+# pure-Python module -- HAVE_NUMPY (from cassandra.cython_deps) only tells us
+# whether the *numpy package itself* is importable, not whether numpy_parser
+# built/imports correctly. A single broad `except ImportError` around all of
+# these imports would conflate two very different situations:
+#   - numpy genuinely not installed (expected in a numpy-less environment;
+#     should skip cleanly), vs.
+#   - the Cython extension being broken or missing despite numpy being
+#     present (an unexpected regression that should FAIL the test run
+#     rather than hide behind a silently skipped test -- this is exactly
+#     how a real numpy_parser regression could slip through a green CI run).
+#
+# So gate on HAVE_CYTHON and HAVE_NUMPY together (the same compound
+# condition as the `numpytest` decorator in tests/unit/cython/utils.py,
+# including its VERIFY_CYTHON override for CI configurations that require
+# Cython support to be present); only skip when Cython and/or numpy support
+# is genuinely absent, and otherwise import unconditionally so a broken
+# extension surfaces as a real ImportError/test failure.
+HAVE_NUMPY_PARSER = (HAVE_CYTHON and HAVE_NUMPY) or VERIFY_CYTHON
+
+if HAVE_NUMPY_PARSER:
     import numpy as np
     from cassandra.numpy_parser import NumpyParser
     from cassandra.bytesio import BytesIOReader
     from cassandra.parsing import ParseDesc
     from cassandra.deserializers import obj_array, make_deserializers
 
-    HAVE_NUMPY = True
-except ImportError:
-    HAVE_NUMPY = False
-
 from cassandra import cqltypes
 
 
-@unittest.skipUnless(HAVE_NUMPY, "NumPy not available")
+@unittest.skipUnless(HAVE_NUMPY_PARSER, "NumPy/Cython extension not available")
 class TestNumpyParserVectorType(unittest.TestCase):
     """Tests for VectorType support in NumpyParser"""
 
@@ -476,6 +499,72 @@ def test_null_vector_data_bytes_are_zeroed(self):
         null_row_bytes = arr.data[1].tobytes()
         self.assertEqual(null_row_bytes, b"\x00" * len(null_row_bytes))
 
+    def test_oversized_vector_payload_raises(self):
+        """Test that a vector payload larger than the array stride is rejected.
+
+        Guards against a memcpy buffer overflow: unpack_row() must not trust
+        an oversized buf.size and copy past the end of the preallocated row.
+        """
+        vector_size = 3
+        vector_type = self._create_vector_type(cqltypes.FloatType, vector_size)
+
+        buffer = bytearray()
+        buffer.extend(struct.pack(">i", 1))  # row count
+
+        # Claim 4 floats (16 bytes) worth of payload for a 3-float (12 byte)
+        # column stride.
+        buffer.extend(struct.pack(">i", 16))
+        buffer.extend(struct.pack(">4f", 1.0, 2.0, 3.0, 4.0))
+
+        parser = NumpyParser()
+        reader = BytesIOReader(bytes(buffer))
+
+        desc = ParseDesc(
+            colnames=["vec"],
+            coltypes=[vector_type],
+            column_encryption_policy=None,
+            coldescs=None,
+            deserializers=obj_array([None]),
+            protocol_version=5,
+        )
+
+        with self.assertRaisesRegex(ValueError, r"'vec'"):
+            parser.parse_rows(reader, desc)
+
+    def test_undersized_vector_payload_raises(self):
+        """Test that a vector payload smaller than the array stride is rejected.
+
+        An undersized payload must not be silently accepted: without a
+        strict equality check, memcpy would only fill part of the row,
+        leaving the remaining bytes uninitialized/stale (a correctness and
+        potential info-leak issue), rather than raising.
+        """
+        vector_size = 4
+        vector_type = self._create_vector_type(cqltypes.FloatType, vector_size)
+
+        buffer = bytearray()
+        buffer.extend(struct.pack(">i", 1))  # row count
+
+        # Claim only 3 floats (12 bytes) worth of payload for a 4-float
+        # (16 byte) column stride.
+        buffer.extend(struct.pack(">i", 12))
+        buffer.extend(struct.pack(">3f", 1.0, 2.0, 3.0))
+
+        parser = NumpyParser()
+        reader = BytesIOReader(bytes(buffer))
+
+        desc = ParseDesc(
+            colnames=["vec"],
+            coltypes=[vector_type],
+            column_encryption_policy=None,
+            coldescs=None,
+            deserializers=obj_array([None]),
+            protocol_version=5,
+        )
+
+        with self.assertRaisesRegex(ValueError, r"'vec'"):
+            parser.parse_rows(reader, desc)
+
     def test_unsupported_subtype_falls_back_to_object_array(self):
         """Test that an unsupported vector subtype falls back to an object array"""
         vector_size = 2
@@ -533,5 +622,69 @@ def test_unsupported_subtype_falls_back_end_to_end_via_parse_rows(self):
             self.assertEqual(list(arr[i]), expected_vector)
 
 
+@unittest.skipUnless(HAVE_NUMPY_PARSER, "NumPy/Cython extension not available")
+class TestNumpyParserScalarType(unittest.TestCase):
+    """Regression tests for the plain (non-vector) scalar column fast path.
+
+    `_cqltype_to_numpy` in cassandra/numpy_parser.pyx is shared by two call
+    sites in make_array(): VectorType.subtype dispatch and plain scalar
+    top-level column dispatch. A fix aimed only at excluding a subtype from
+    the vector fast path (e.g. ShortType, which is vint-prefixed *inside* a
+    vector) must not remove that subtype's entry from the shared dict
+    outright, since that would also silently regress the scalar column fast
+    path for the exact same type (ShortType/smallint *is* fixed-width as a
+    scalar column). These tests guard against that regression.
+    """
+
+    def test_scalar_smallint_uses_fast_masked_int16_array(self):
+        """smallint (ShortType) is fixed-width (2 bytes, big-endian, no
+        length prefix) as a scalar column -- unlike inside a VectorType,
+        where it is vint-length-prefixed per element. It must still get the
+        fast masked int16 array here, not fall back to a slow 1D object
+        array.
+        """
+        from cassandra.numpy_parser import make_array
+
+        arr = make_array(cqltypes.ShortType, 5)
+
+        self.assertNotEqual(arr.dtype, np.dtype("O"))
+        self.assertEqual(arr.dtype, np.dtype(">i2"))
+        self.assertTrue(hasattr(arr, "mask"))
+        self.assertEqual(arr.shape, (5,))
+
+    def test_scalar_smallint_round_trips_end_to_end(self):
+        """End-to-end companion: parse actual wire bytes for a smallint
+        column through NumpyParser.parse_rows() and confirm the result is a
+        native int16 (masked) array with the correct values, not an object
+        array.
+        """
+        values = [1, -5, 32767, -32768, 0]
+
+        buffer = bytearray()
+        buffer.extend(struct.pack(">i", len(values)))  # row count
+        for v in values:
+            buffer.extend(struct.pack(">i", 2))  # smallint byte size
+            buffer.extend(struct.pack(">h", v))
+
+        parser = NumpyParser()
+        reader = BytesIOReader(bytes(buffer))
+
+        desc = ParseDesc(
+            colnames=["small"],
+            coltypes=[cqltypes.ShortType],
+            column_encryption_policy=None,
+            coldescs=None,
+            deserializers=obj_array([None]),
+            protocol_version=5,
+        )
+
+        result = parser.parse_rows(reader, desc)
+        arr = result["small"]
+
+        self.assertNotEqual(arr.dtype, np.dtype("O"))
+        self.assertEqual(arr.shape, (len(values),))
+        np.testing.assert_array_equal(arr, np.array(values, dtype=np.int16))
+
+
 if __name__ == "__main__":
     unittest.main()