diff --git a/cassandra/bytesio.pxd b/cassandra/bytesio.pxd index d52d3fa8fe..2bffef067c 100644 --- a/cassandra/bytesio.pxd +++ b/cassandra/bytesio.pxd @@ -18,3 +18,4 @@ cdef class BytesIOReader: cdef Py_ssize_t pos cdef Py_ssize_t size cdef char *read(self, Py_ssize_t n = ?) except NULL + cdef void seek(self, Py_ssize_t new_pos) diff --git a/cassandra/bytesio.pyx b/cassandra/bytesio.pyx index 1a57911fcf..1c26e04595 100644 --- a/cassandra/bytesio.pyx +++ b/cassandra/bytesio.pyx @@ -42,3 +42,19 @@ cdef class BytesIOReader: cdef char *res = self.buf_ptr + self.pos self.pos = newpos return res + + cdef void seek(self, Py_ssize_t new_pos): + """ + Reposition the reader to an absolute offset within the buffer, so + the next read() starts at `new_pos`. + + `buf_ptr` is the base address of `buf` (the underlying bytes + object); it is set once in __init__ and is never reassigned + anywhere else -- read() always computes `buf_ptr + pos`, so `pos` + is the only mutable cursor state and the only field a caller ever + needs to reset. This method exists so callers reposition the + reader through a single explicit entry point instead of reaching + into `pos` (and, in older code, the always-redundant `buf_ptr`) + directly at the call site. + """ + self.pos = new_pos diff --git a/cassandra/ioutils.pyx b/cassandra/ioutils.pyx index b0ab4f16cb..2700901747 100644 --- a/cassandra/ioutils.pyx +++ b/cassandra/ioutils.pyx @@ -15,7 +15,7 @@ include 'cython_marshal.pyx' from cassandra.buffer cimport Buffer, from_ptr_and_size -from libc.stdint cimport int32_t +from libc.stdint cimport int32_t, uint16_t from cassandra.bytesio cimport BytesIOReader @@ -45,3 +45,9 @@ cdef inline int32_t read_int(BytesIOReader reader) except ?0xDEAD: buf.ptr = reader.read(4) buf.size = 4 return unpack_num[int32_t](&buf) + +cdef inline uint16_t read_short(BytesIOReader reader) except ?0xFFFE: + cdef Buffer buf + buf.ptr = reader.read(2) + buf.size = 2 + return unpack_num[uint16_t](&buf) diff --git a/cassandra/metadata_parser.pyx b/cassandra/metadata_parser.pyx new file mode 100644 index 0000000000..6241be0c31 --- /dev/null +++ b/cassandra/metadata_parser.pyx @@ -0,0 +1,218 @@ +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Cython-optimized metadata parsing for CQL protocol ResultMessage. + +Uses BytesIOReader for zero-copy reads, eliminating per-read bytes allocation +that dominates recv_results_metadata cost. +""" + +include "ioutils.pyx" + + +# ---------- low-level readers on BytesIOReader ---------- +# read_int(BytesIOReader) and read_short(BytesIOReader) are provided by ioutils.pyx + +cdef inline str read_string_br(BytesIOReader reader): + """Read a [string]: a [short] n, followed by n bytes of UTF-8.""" + cdef uint16_t size = read_short(reader) + cdef char *ptr = reader.read(size) + return ptr[:size].decode('utf8') + +cdef inline bytes read_binary_string_br(BytesIOReader reader): + """Read a [short bytes]: a [short] n, followed by n raw bytes.""" + cdef uint16_t size = read_short(reader) + cdef char *ptr = reader.read(size) + return ptr[:size] + +cdef inline bytes read_binary_longstring_br(BytesIOReader reader): + """Read a [bytes]: an [int] n, followed by n raw bytes. + + Unlike row values (where a negative length is a valid NULL/"not set" + sentinel handled by get_buf() in ioutils.pyx), the [bytes] values read + here (paging_state, result_metadata_id) are only read when the protocol + says they are present, so a negative length here means a malformed or + corrupted frame. Reject it explicitly instead of passing it through: + BytesIOReader.read() treats any negative n as "read to the end of the + buffer" rather than raising, and slicing a char* with a negative stop + bound is not a validated length -- on CPython that currently surfaces + as an opaque SystemError from PyBytes_FromStringAndSize instead of a + clean, catchable protocol error. + """ + cdef int32_t size = read_int(reader) + if size < 0: + raise ValueError( + "Invalid negative length %d for [bytes] value" % size) + cdef char *ptr = reader.read(size) + return ptr[:size] + + +# ---------- flag constants (mirrored from ResultMessage) ---------- +# These MUST stay in sync with the class attributes in ResultMessage (protocol.py). +# They are duplicated here as compile-time DEF constants for Cython performance. + +DEF _FLAGS_GLOBAL_TABLES_SPEC = 0x0001 +DEF _HAS_MORE_PAGES_FLAG = 0x0002 +DEF _NO_METADATA_FLAG = 0x0004 +DEF _METADATA_ID_FLAG = 0x0008 +DEF _CONTINUOUS_PAGING_FLAG = 0x40000000 +DEF _CONTINUOUS_PAGING_LAST_FLAG = 0x80000000 + + +# ---------- read_type using BytesIOReader ---------- + +cdef object _read_type_br(BytesIOReader reader, dict type_codes_map, object user_type_map, + object ListType, object SetType, object MapType, + object TupleType, object UserType, object CUSTOM_TYPE, + object lookup_casstype, object NotSupportedError): + """ + Cython version of ResultMessage.read_type() operating on BytesIOReader. + + Parameters are passed in to avoid module-level imports from protocol.py + (which would create circular dependencies). They are captured once in the + closure created by make_recv_results_metadata(). + """ + cdef uint16_t optid = read_short(reader) + + typeclass = type_codes_map.get(optid) + if typeclass is None: + raise NotSupportedError( + "Unknown data type code 0x%04x. Have to skip entire result set." % (optid,)) + + if typeclass is ListType or typeclass is SetType: + subtype = _read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError) + typeclass = typeclass.apply_parameters((subtype,)) + elif typeclass is MapType: + keysubtype = _read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError) + valsubtype = _read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError) + typeclass = typeclass.apply_parameters((keysubtype, valsubtype)) + elif typeclass is TupleType: + num_items = read_short(reader) + types = tuple(_read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError) + for _ in range(num_items)) + typeclass = typeclass.apply_parameters(types) + elif typeclass is UserType: + ks = read_string_br(reader) + udt_name = read_string_br(reader) + num_fields = read_short(reader) + names_and_types = tuple( + (read_string_br(reader), + _read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError)) + for _ in range(num_fields)) + # zip(*()) with num_fields == 0 raises ValueError (not enough values + # to unpack), matching ResultMessage.read_type()'s pure-Python + # behavior for the same malformed/degenerate case: a real UDT + # always has at least one field, so a UserType with num_fields == 0 + # on the wire indicates a corrupted or malformed frame and should + # raise rather than silently producing an empty-field UDT. + names, types = zip(*names_and_types) + specialized_type = typeclass.make_udt_class(ks, udt_name, names, types) + specialized_type.mapped_class = user_type_map.get(ks, {}).get(udt_name) + typeclass = specialized_type + elif typeclass is CUSTOM_TYPE: + classname = read_string_br(reader) + typeclass = lookup_casstype(classname) + + return typeclass + + +# ---------- public factory: creates closures that capture type objects ---------- + +def make_recv_results_metadata(dict type_codes_map, object CUSTOM_TYPE, + object ListType, object SetType, object MapType, + object TupleType, object UserType, + object lookup_casstype, object NotSupportedError): + """ + Factory that returns a recv_results_metadata function suitable for use + as an unbound method replacement on FastResultMessage. + + The closure captures the type-code map and type objects once, so they + don't have to be looked up on every call. + + All type objects -- including the NotSupportedError exception class + raised for an unrecognized type code -- are passed in by the caller + (cython_protocol_handler() in protocol.py, which already has them as + module-level names) rather than imported here. cassandra.metadata_parser + is itself imported from inside cassandra.protocol's own module body + (cython_protocol_handler() runs while "import cassandra.protocol" is + still executing), so a module-level + "from cassandra.protocol import ResultMessage, CUSTOM_TYPE" here would be a + self-referential import back into the partially-initialized protocol + module -- the same class of import-order fragility that used to make + cassandra.cython_deps.HAVE_CYTHON detection depend on which cassandra + module a process happened to import first. It happens to work today only + because ResultMessage/CUSTOM_TYPE are defined earlier in protocol.py than + cython_protocol_handler() is called; passing them in avoids relying on + that ordering at all. + """ + def read_type_br_closure(BytesIOReader reader, user_type_map): + return _read_type_br(reader, type_codes_map, user_type_map, + ListType, SetType, MapType, TupleType, UserType, + CUSTOM_TYPE, lookup_casstype, NotSupportedError) + + def recv_results_metadata(self, BytesIOReader reader, user_type_map): + """ + Cython-optimized recv_results_metadata operating on BytesIOReader. + Replaces ResultMessage.recv_results_metadata. + """ + cdef int32_t flags = read_int(reader) + cdef int32_t colcount = read_int(reader) + + if flags & _HAS_MORE_PAGES_FLAG: + self.paging_state = read_binary_longstring_br(reader) + + if flags & _NO_METADATA_FLAG: + return + + if flags & _CONTINUOUS_PAGING_FLAG: + self.continuous_paging_seq = read_int(reader) + self.continuous_paging_last = flags & _CONTINUOUS_PAGING_LAST_FLAG + + if flags & _METADATA_ID_FLAG: + self.result_metadata_id = read_binary_string_br(reader) + + cdef str ksname, cfname, colname + cdef object coltype + cdef int i + cdef list column_metadata = [None] * colcount + + if flags & _FLAGS_GLOBAL_TABLES_SPEC: + ksname = read_string_br(reader) + cfname = read_string_br(reader) + for i in range(colcount): + colname = read_string_br(reader) + coltype = read_type_br_closure(reader, user_type_map) + column_metadata[i] = (ksname, cfname, colname, coltype) + else: + for i in range(colcount): + ksname = read_string_br(reader) + cfname = read_string_br(reader) + colname = read_string_br(reader) + coltype = read_type_br_closure(reader, user_type_map) + column_metadata[i] = (ksname, cfname, colname, coltype) + + self.column_metadata = column_metadata + + return recv_results_metadata diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 9dfdbf3022..fb9ad1c44b 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -1264,15 +1264,30 @@ def cython_protocol_handler(colparser): The default is to use obj_parser.ListParser """ from cassandra.row_parser import make_recv_results_rows + from cassandra.metadata_parser import make_recv_results_metadata + + # Pass already-available module-level names in explicitly instead of + # letting metadata_parser import them itself: cassandra.metadata_parser is + # imported here, from inside cassandra.protocol's own module body, so a + # "from cassandra.protocol import ResultMessage, CUSTOM_TYPE" inside + # metadata_parser would be a self-referential import back into this + # (still-initializing) module. NotSupportedError is injected the same + # way, so metadata_parser never needs to import it from protocol.py at + # runtime either. + recv_results_metadata_br = make_recv_results_metadata( + ResultMessage.type_codes, CUSTOM_TYPE, + ListType, SetType, MapType, TupleType, UserType, lookup_casstype, + NotSupportedError, + ) class FastResultMessage(ResultMessage): """ Cython version of Result Message that has a faster implementation of - recv_results_row. + recv_results_rows using BytesIOReader for zero-copy metadata + row parsing. """ # type_codes = ResultMessage.type_codes.copy() code_to_type = dict((v, k) for k, v in ResultMessage.type_codes.items()) - recv_results_rows = make_recv_results_rows(colparser) + recv_results_rows = make_recv_results_rows(colparser, recv_results_metadata_br) class CythonProtocolHandler(_ProtocolHandler): """ diff --git a/cassandra/row_parser.pyx b/cassandra/row_parser.pyx index 88277a4593..356441e13f 100644 --- a/cassandra/row_parser.pyx +++ b/cassandra/row_parser.pyx @@ -19,13 +19,27 @@ from cassandra.deserializers import make_deserializers include "ioutils.pyx" -def make_recv_results_rows(ColumnParser colparser): +def make_recv_results_rows(ColumnParser colparser, recv_results_metadata_br): + """ + Create a recv_results_rows closure that uses: + - recv_results_metadata_br: Cython metadata parser operating on BytesIOReader + - colparser: Cython column parser for row data + + A single BytesIOReader is created from the full remaining buffer and used + for both metadata parsing and row parsing, eliminating the per-read bytes + allocation overhead of Python BytesIO. + """ def recv_results_rows(self, f, int protocol_version, user_type_map, result_metadata, column_encryption_policy): """ Parse protocol data given as a BytesIO f into a set of columns (e.g. list of tuples) This is used as the recv_results_rows method of (Fast)ResultMessage """ - self.recv_results_metadata(f, user_type_map) + # Create ONE BytesIOReader for the entire remaining buffer. + # This is used for both metadata parsing and row parsing. + reader = BytesIOReader(f.read()) + + # Use Cython-optimized metadata parsing on BytesIOReader + recv_results_metadata_br(self, reader, user_type_map) column_metadata = self.column_metadata or result_metadata @@ -35,14 +49,16 @@ def make_recv_results_rows(ColumnParser colparser): desc = ParseDesc(self.column_names, self.column_types, column_encryption_policy, [ColDesc(md[0], md[1], md[2]) for md in column_metadata], make_deserializers(self.column_types), protocol_version) - reader = BytesIOReader(f.read()) + # The reader's position is now right after the metadata; + # row data follows immediately — no need to create a second reader. + # Save position so we can rewind to the start of row data on error. + cdef Py_ssize_t rows_start_pos = reader.pos try: self.parsed_rows = colparser.parse_rows(reader, desc) except Exception as e: # Use explicitly the TupleRowParser to display better error messages for column decoding failures rowparser = TupleRowParser() - reader.buf_ptr = reader.buf - reader.pos = 0 + reader.seek(rows_start_pos) rowcount = read_int(reader) for i in range(rowcount): rowparser.unpack_row(reader, desc) diff --git a/tests/unit/cython/bytesio_testhelper.pyx b/tests/unit/cython/bytesio_testhelper.pyx index 595cd29cc8..30ead88c2b 100644 --- a/tests/unit/cython/bytesio_testhelper.pyx +++ b/tests/unit/cython/bytesio_testhelper.pyx @@ -37,3 +37,21 @@ def test_read_eof(): with pytest.raises(EOFError): reader.read(2) reader.read(1) # see that we can still read this + +def test_seek(): + # seek() must reposition the reader so that a subsequent read() lands + # at the correct byte offset and returns the correct bytes -- this is + # the same repositioning row_parser.pyx relies on when it rewinds the + # reader on the recv_results_rows exception-recovery path. + cdef BytesIOReader reader = BytesIOReader(b'abcdef') + assert reader.read(4)[:4] == b'abcd' # pos -> 4 + + reader.seek(1) + assert reader.read(3)[:3] == b'bcd' # pos -> 4 again, via a different route + + reader.seek(0) + assert reader.read(6)[:6] == b'abcdef' # full rewind still reads correctly + + reader.seek(6) + with pytest.raises(EOFError): + reader.read(1) diff --git a/tests/unit/cython/test_bytesio.py b/tests/unit/cython/test_bytesio.py index 0f27663391..d0626aac8c 100644 --- a/tests/unit/cython/test_bytesio.py +++ b/tests/unit/cython/test_bytesio.py @@ -30,3 +30,7 @@ def test_reading(self): @cythontest def test_reading_error(self): bytesio_testhelper.test_read_eof() + + @cythontest + def test_seek(self): + bytesio_testhelper.test_seek() diff --git a/tests/unit/cython/test_metadata_parser.py b/tests/unit/cython/test_metadata_parser.py new file mode 100644 index 0000000000..611e3b5ba5 --- /dev/null +++ b/tests/unit/cython/test_metadata_parser.py @@ -0,0 +1,176 @@ +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Tests for cassandra.metadata_parser, the Cython BytesIOReader-based +implementation of ResultMessage.recv_results_metadata. +""" + +import io +import unittest + +from tests.unit.cython.utils import cythontest + +try: + from cassandra.metadata_parser import make_recv_results_metadata + from cassandra.bytesio import BytesIOReader + from cassandra.protocol import ResultMessage, CUSTOM_TYPE, RESULT_KIND_ROWS, NotSupportedError + from cassandra.cqltypes import (ListType, SetType, MapType, TupleType, + UserType, lookup_casstype) + from cassandra.marshal import int32_pack, uint16_pack +except ImportError: + make_recv_results_metadata = None + + +# Type code for UserType (UDT), per the CQL binary protocol spec (0x0030). +_USER_TYPE_CODE = 0x0030 + + +def _string(s): + b = s.encode('utf8') + return uint16_pack(len(b)) + b + + +class MetadataParserTest(unittest.TestCase): + """Test the Cython metadata parser (cassandra.metadata_parser).""" + + def _make_recv(self, not_supported_error=None): + return make_recv_results_metadata( + ResultMessage.type_codes, CUSTOM_TYPE, + ListType, SetType, MapType, TupleType, UserType, lookup_casstype, + not_supported_error if not_supported_error is not None else NotSupportedError) + + @cythontest + def test_negative_paging_state_length_rejected(self): + """ + A malformed/corrupted frame carrying a negative length for the + paging_state [bytes] value must raise a clean, catchable ValueError + instead of an opaque SystemError from PyBytes_FromStringAndSize (or + worse -- undefined behavior from slicing a char* with a negative + bound). See read_binary_longstring_br() in metadata_parser.pyx. + """ + recv = self._make_recv() + flags = 0x0002 | 0x0004 # _HAS_MORE_PAGES_FLAG | _NO_METADATA_FLAG + for bad_len in (-1, -2, -1000, -(2 ** 31)): + buf = int32_pack(flags) + int32_pack(0) + int32_pack(bad_len) + reader = BytesIOReader(buf) + msg = ResultMessage(RESULT_KIND_ROWS) + with self.assertRaises(ValueError): + recv(msg, reader, {}) + + @cythontest + def test_valid_paging_state_roundtrip(self): + recv = self._make_recv() + flags = 0x0002 | 0x0004 # _HAS_MORE_PAGES_FLAG | _NO_METADATA_FLAG + buf = int32_pack(flags) + int32_pack(0) + int32_pack(3) + b'abc' + reader = BytesIOReader(buf) + msg = ResultMessage(RESULT_KIND_ROWS) + recv(msg, reader, {}) + self.assertEqual(msg.paging_state, b'abc') + + @cythontest + def test_no_metadata_flag_skips_column_parsing(self): + recv = self._make_recv() + flags = 0x0004 # _NO_METADATA_FLAG + buf = int32_pack(flags) + int32_pack(0) + reader = BytesIOReader(buf) + msg = ResultMessage(RESULT_KIND_ROWS) + recv(msg, reader, {}) + self.assertIsNone(msg.column_metadata) + + @cythontest + def test_column_metadata_matches_pure_python(self): + """ + The Cython recv_results_metadata (BytesIOReader-based) must produce + the same column_metadata as the pure-Python + ResultMessage.recv_results_metadata for identical wire bytes, for + both the global-tables-spec and the per-column keyspace/table-name + wire layouts. + """ + for global_tables_spec in (True, False): + flags = 0x0001 if global_tables_spec else 0x0000 + body = int32_pack(flags) + int32_pack(2) + if global_tables_spec: + body += _string('ks') + _string('table') + body += _string('col1') + uint16_pack(0x0009) + body += _string('col2') + uint16_pack(0x0009) + else: + body += _string('ks') + _string('table') + _string('col1') + uint16_pack(0x0009) + body += _string('ks') + _string('table') + _string('col2') + uint16_pack(0x0009) + + recv = self._make_recv() + reader = BytesIOReader(body) + cy_msg = ResultMessage(RESULT_KIND_ROWS) + recv(cy_msg, reader, {}) + + py_msg = ResultMessage(RESULT_KIND_ROWS) + py_msg.recv_results_metadata(io.BytesIO(body), {}) + + self.assertEqual(cy_msg.column_metadata, py_msg.column_metadata) + + @cythontest + def test_unknown_type_code_raises_injected_exception(self): + """ + _read_type_br() must raise the NotSupportedError class *injected* + into make_recv_results_metadata() by the caller, not import it from + cassandra.protocol at call time. Passing in a stand-in exception + class (instead of the real cassandra.protocol.NotSupportedError) + and asserting *that* class is what gets raised proves the raise + goes through the injected closure variable rather than a hidden + "from cassandra.protocol import NotSupportedError" inside the hot + parsing path -- which would defeat the whole point of injecting + CUSTOM_TYPE/ListType/etc. instead of importing them, as documented + in make_recv_results_metadata()'s own docstring. + """ + class StandInNotSupportedError(Exception): + pass + + recv = self._make_recv(not_supported_error=StandInNotSupportedError) + flags = 0x0001 # _FLAGS_GLOBAL_TABLES_SPEC + unknown_optid = 0x00FF # not a key in ResultMessage.type_codes + body = (int32_pack(flags) + int32_pack(1) + + _string('ks') + _string('table') + + _string('col1') + uint16_pack(unknown_optid)) + reader = BytesIOReader(body) + msg = ResultMessage(RESULT_KIND_ROWS) + with self.assertRaises(StandInNotSupportedError): + recv(msg, reader, {}) + + @cythontest + def test_zero_field_user_type_matches_pure_python(self): + """ + A UserType (UDT) with num_fields == 0 is not something a real + CREATE TYPE can ever produce (a UDT always has at least one field), + so it only ever shows up on the wire as malformed/corrupted data. + ResultMessage.read_type() (the pure-Python path) already rejects it + -- zip(*()) unpacked into two variables raises ValueError -- so the + Cython path must raise the same way instead of silently returning + an empty-fields UDT, to avoid a behavioral divergence between the + two implementations for the same malformed input. + """ + user_type_bytes = (uint16_pack(_USER_TYPE_CODE) + + _string('ks') + _string('mytype') + uint16_pack(0)) + + with self.assertRaises(ValueError): + ResultMessage.read_type(io.BytesIO(user_type_bytes), {}) + + flags = 0x0001 # _FLAGS_GLOBAL_TABLES_SPEC + body = (int32_pack(flags) + int32_pack(1) + + _string('ks') + _string('table') + + _string('col1') + user_type_bytes) + recv = self._make_recv() + reader = BytesIOReader(body) + msg = ResultMessage(RESULT_KIND_ROWS) + with self.assertRaises(ValueError): + recv(msg, reader, {}) diff --git a/tests/unit/cython/test_row_parser.py b/tests/unit/cython/test_row_parser.py new file mode 100644 index 0000000000..8af25efb69 --- /dev/null +++ b/tests/unit/cython/test_row_parser.py @@ -0,0 +1,106 @@ +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Tests for cassandra.row_parser, in particular the exception-recovery path +in recv_results_rows() that re-parses row data with TupleRowParser after +the primary ColumnParser fails partway through. +""" + +import io +import types +import unittest + +from tests.unit.cython.utils import cythontest + +try: + from cassandra.row_parser import make_recv_results_rows + from cassandra.obj_parser import ListParser + from cassandra.cqltypes import Int32Type + from cassandra.marshal import int32_pack + from cassandra import DriverException +except ImportError: + make_recv_results_rows = None + + +def _no_op_recv_results_metadata_br(self, reader, user_type_map): + """ + Stand-in for the real (Cython) recv_results_metadata: sets + column_metadata directly without consuming anything from `reader`, so + row data starts at offset 0 and the test can control the exact bytes + the exception-recovery path re-reads. + """ + self.column_metadata = [('ks', 'table', 'col', Int32Type)] + + +class RowParserExceptionRecoveryTest(unittest.TestCase): + """ + recv_results_rows() saves the reader's position before the primary + (fast) column-parsing attempt and, if that attempt raises, rewinds the + reader to that saved position before re-parsing row-by-row with + TupleRowParser for a clearer error message (see row_parser.pyx). + + That rewind must leave the reader able to correctly re-read the exact + same bytes it started with. If the rewind were incomplete -- e.g. if + BytesIOReader tracked a second, separate cursor that the rewind forgot + to reset -- the re-parse would read from the wrong offset and either + misdecode silently or blow up with an unrelated/garbled error instead + of cleanly re-reporting the original failure. + """ + + @cythontest + def test_exception_recovery_rereads_correct_bytes(self): + recv_results_rows = make_recv_results_rows(ListParser(), _no_op_recv_results_metadata_br) + + # Row 1 is a valid 4-byte int32 value. Row 2 declares a [bytes] + # length of 2, which is too short for Int32Type's 4-byte decode -- + # this is what makes the *first* (fast) parse attempt raise. + row1 = int32_pack(4) + int32_pack(123) + row2 = int32_pack(2) + b'xy' + body = int32_pack(2) + row1 + row2 # rowcount = 2 + + msg = types.SimpleNamespace(column_metadata=None, parsed_rows=None) + with self.assertRaises(DriverException) as ctx: + recv_results_rows(msg, io.BytesIO(body), 4, {}, None, None) + + # The recovery path re-reads rowcount and both rows from the + # rewound position. If it landed on the correct bytes, the + # re-raised error still names the same column/type and the same + # underlying cause as the original failure -- not some other + # column, a bogus rowcount-driven EOFError, or silent garbage. + message = str(ctx.exception) + self.assertIn('"col"', message) + self.assertIn('Requested more than length of buffer', message) + + @cythontest + def test_exception_recovery_after_valid_prefix_rows(self): + """ + Same as above, but with several valid rows before the bad one, so + the saved/rewound position is a genuinely non-zero, mid-buffer + offset rather than 0 -- exercising an actual seek-backwards rather + than a no-op rewind to the start. + """ + recv_results_rows = make_recv_results_rows(ListParser(), _no_op_recv_results_metadata_br) + + good_rows = b''.join(int32_pack(4) + int32_pack(v) for v in (1, 2, 3)) + bad_row = int32_pack(2) + b'xy' + body = int32_pack(4) + good_rows + bad_row # rowcount = 4 + + msg = types.SimpleNamespace(column_metadata=None, parsed_rows=None) + with self.assertRaises(DriverException) as ctx: + recv_results_rows(msg, io.BytesIO(body), 4, {}, None, None) + + message = str(ctx.exception) + self.assertIn('"col"', message) + self.assertIn('Requested more than length of buffer', message)