From 232f58c9afe8868446eab93d5d534585e5d6a99a Mon Sep 17 00:00:00 2001 From: Martin Gallwey Date: Thu, 10 Sep 2026 16:37:15 +0100 Subject: [PATCH] Test --- .circleci/config.yml | 6 +- .gitignore | 3 + MANIFEST.in | 3 +- Makefile | 4 + pynuodb/_codes.pxi | 47 ++ pynuodb/_cutil.h | 135 +++++ pynuodb/_fetch.pyx | 412 ++++++++++++++ pynuodb/_resultset.pxi | 40 ++ pynuodb/encodedsession.py | 33 ++ pynuodb/result_set.py | 9 + pyproject.toml | 3 + setup.py | 68 ++- tests/nuodb_cython_test.py | 722 +++++++++++++++++++++++++ tests/perf/test_insert_select_bench.py | 32 ++ 14 files changed, 1513 insertions(+), 4 deletions(-) create mode 100644 pynuodb/_codes.pxi create mode 100644 pynuodb/_cutil.h create mode 100644 pynuodb/_fetch.pyx create mode 100644 pynuodb/_resultset.pxi create mode 100644 pyproject.toml create mode 100644 tests/nuodb_cython_test.py diff --git a/.circleci/config.yml b/.circleci/config.yml index f15d224..24fdeeb 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -41,8 +41,10 @@ jobs: steps: - checkout - run: - name: Install make - command: dnf install make -y + name: Install build tools + command: | + PYVER=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")') + dnf install -y make gcc "python${PYVER}-devel" - run: name: Install pip command: | diff --git a/.gitignore b/.gitignore index c8618e3..1a7a030 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,9 @@ *.so +# Cython-generated C source (rebuilt from .pyx by setup.py) +pynuodb/_fetch.c + /.virttemp /.testtemp diff --git a/MANIFEST.in b/MANIFEST.in index 05ac65b..f3ecc96 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1 +1,2 @@ -include README.rst LICENSE \ No newline at end of file +include README.rst LICENSE +recursive-include pynuodb *.pyx *.pxd *.pxi *.c *.h \ No newline at end of file diff --git a/Makefile b/Makefile index 5c478d5..7e453a2 100644 --- a/Makefile +++ b/Makefile @@ -32,6 +32,8 @@ SUDO ?= sudo -n NUODB_HOME ?= /opt/nuodb _INSTALL_CMD = $(PIP) install '.[crypto]' +_BUILD_EXT_CMD = $(PIP) install 'setuptools>=40.8.0' 'Cython>=3.0' \ + && $(PYTHON) setup.py build_ext --inplace _VERIFY_CMD = $(NUODB_HOME)/bin/nuocmd show domain _PYTEST_CMD = $(MKDIR) $(ARTIFACTDIR) $(RESULTSDIR) \ && TMPDIR='$(TMPDIR)' PATH="$(NUODB_HOME)/bin:$$PATH" \ @@ -43,11 +45,13 @@ all: install: $(_INSTALL_CMD) + $(_BUILD_EXT_CMD) check: mypy pylint fulltest fulltest: $(_INSTALL_CMD) + $(_BUILD_EXT_CMD) $(PIP) install -r test_requirements.txt $(_VERIFY_CMD) $(_PYTEST_CMD) diff --git a/pynuodb/_codes.pxi b/pynuodb/_codes.pxi new file mode 100644 index 0000000..4e0b170 --- /dev/null +++ b/pynuodb/_codes.pxi @@ -0,0 +1,47 @@ +# Wire-protocol type codes, mirrored from protocol.py as C ints so +# decode_next_batch's per-cell range checks in _fetch.pyx don't pay for a +# Python attribute lookup on every column. +from . import protocol as _protocol + +cdef int NULL_V = _protocol.NULL +cdef int TRUE_V = _protocol.TRUE +cdef int FALSE_V = _protocol.FALSE +cdef int INTMINUS10 = _protocol.INTMINUS10 +cdef int INT0 = _protocol.INT0 # inline value 0: end-of-batch marker +cdef int INT31 = _protocol.INT31 +cdef int INTLEN0 = _protocol.INTLEN0 # == INT31; codes 52-59 carry 1-8 byte integers +cdef int INTLEN8 = _protocol.INTLEN8 +cdef int SCALEDLEN0 = _protocol.SCALEDLEN0 # base for 1-8 byte scaled decimals (61-68) +cdef int SCALEDLEN8 = _protocol.SCALEDLEN8 +cdef int UTF8COUNT0 = _protocol.UTF8COUNT0 # base for length-prefixed strings (69-72) +cdef int UTF8COUNT1 = _protocol.UTF8COUNT1 +cdef int UTF8COUNT4 = _protocol.UTF8COUNT4 +cdef int OPAQUECOUNT0 = _protocol.OPAQUECOUNT0 # base for length-prefixed binary (73-76) +cdef int OPAQUECOUNT1 = _protocol.OPAQUECOUNT1 +cdef int OPAQUECOUNT4 = _protocol.OPAQUECOUNT4 +cdef int DOUBLELEN0 = _protocol.DOUBLELEN0 # 77 == double 0.0; 78-85 carry 1-8 byte doubles +cdef int DOUBLELEN8 = _protocol.DOUBLELEN8 +cdef int MILLISECLEN0 = _protocol.MILLISECLEN0 # base for 1-8 byte millisecond timestamps +cdef int MILLISECLEN8 = _protocol.MILLISECLEN8 +cdef int NANOSECLEN0 = _protocol.NANOSECLEN0 # base for 1-8 byte nanosecond timestamps +cdef int NANOSECLEN8 = _protocol.NANOSECLEN8 +cdef int TIMELEN0 = _protocol.TIMELEN0 # base for 1-4 byte ms-since-midnight +cdef int TIMELEN4 = _protocol.TIMELEN4 +cdef int UTF8LEN0 = _protocol.UTF8LEN0 # base for 0-39 byte inline-length strings +cdef int UTF8LEN39 = _protocol.UTF8LEN39 +cdef int OPAQUELEN0 = _protocol.OPAQUELEN0 # base for 0-39 byte inline-length binary +cdef int OPAQUELEN39 = _protocol.OPAQUELEN39 +cdef int BLOBLEN0 = _protocol.BLOBLEN0 # base for 0-4 byte length-prefixed BLOB +cdef int BLOBLEN4 = _protocol.BLOBLEN4 +cdef int CLOBLEN0 = _protocol.CLOBLEN0 # base for 0-4 byte length-prefixed CLOB +cdef int CLOBLEN4 = _protocol.CLOBLEN4 +cdef int UUID_C = _protocol.UUID +cdef int SCALEDDATELEN0 = _protocol.SCALEDDATELEN0 # 201-208 carry 1-8 byte scaled dates +cdef int SCALEDDATELEN8 = _protocol.SCALEDDATELEN8 +cdef int SCALEDTIMELEN0 = _protocol.SCALEDTIMELEN0 # 209-216 carry 1-8 byte scaled times +cdef int SCALEDTIMELEN8 = _protocol.SCALEDTIMELEN8 +cdef int SCALEDTIMESTAMPLEN0 = _protocol.SCALEDTIMESTAMPLEN0 # 217-224 carry 1-8 byte scaled timestamps +cdef int SCALEDTIMESTAMPLEN8 = _protocol.SCALEDTIMESTAMPLEN8 +# No LEN1..LEN8 range: 234-240 belong to ARRAYLEN/SCALEDCOUNT3/DEBUGBARRIER. +cdef int SCALEDTIMESTAMPNOTZLEN0 = _protocol.SCALEDTIMESTAMPNOTZLEN0 +cdef int SCALEDTIMESTAMPNOTZ = _protocol.SCALEDTIMESTAMPNOTZ diff --git a/pynuodb/_cutil.h b/pynuodb/_cutil.h new file mode 100644 index 0000000..cca787d --- /dev/null +++ b/pynuodb/_cutil.h @@ -0,0 +1,135 @@ +#ifndef PYNUODB_CUTIL_H +#define PYNUODB_CUTIL_H +/* C support for _fetch.pyx's decode loop: raw-PyObject* value builders for + the row tuple, and big-endian integer/double readers for the wire + protocol. */ + +#include +#include +#include + +#if defined(_MSC_VER) +#include +#define NUODB_INLINE static __inline +#define NUODB_LIKELY(x) (x) +#define NUODB_UNLIKELY(x) (x) +#define NUODB_BSWAP16(x) _byteswap_ushort(x) +#define NUODB_BSWAP32(x) _byteswap_ulong(x) +#define NUODB_BSWAP64(x) _byteswap_uint64(x) +#else +#define NUODB_INLINE static inline +#define NUODB_LIKELY(x) __builtin_expect(!!(x), 1) +#define NUODB_UNLIKELY(x) __builtin_expect(!!(x), 0) +#define NUODB_BSWAP16(x) __builtin_bswap16(x) +#define NUODB_BSWAP32(x) __builtin_bswap32(x) +#define NUODB_BSWAP64(x) __builtin_bswap64(x) +#endif + +/* Build a value as a raw PyObject* for decode_next_batch to steal into a + row tuple via _pynuodb_tuple_steal. Don't replace these with Cython's + cpython.* pxd bindings: those take/return `object`, not `PyObject*`, + which would route the value through Cython's own refcounting first. */ +NUODB_INLINE void _pynuodb_tuple_steal(PyObject *t, Py_ssize_t i, PyObject *o) +{ + PyTuple_SET_ITEM(t, i, o); +} + +NUODB_INLINE PyObject *_pynuodb_long_from_long(long v) +{ + return PyLong_FromLong(v); +} + +NUODB_INLINE PyObject *_pynuodb_long_from_longlong(long long v) +{ + return PyLong_FromLongLong(v); +} + +NUODB_INLINE PyObject *_pynuodb_decode_utf8(const char *s, Py_ssize_t n) +{ + return PyUnicode_DecodeUTF8(s, n, NULL); +} + +NUODB_INLINE PyObject *_pynuodb_incref(PyObject *o) +{ + Py_INCREF(o); + return o; +} + +NUODB_INLINE PyObject *_pynuodb_none(void) { return Py_None; } +NUODB_INLINE PyObject *_pynuodb_true(void) { return Py_True; } +NUODB_INLINE PyObject *_pynuodb_false(void) { return Py_False; } + +NUODB_INLINE unsigned long long _pynuodb_be_u64(const unsigned char *p, int n) +{ + switch (n) + { + case 0: return 0; + case 1: return p[0]; + case 2: { + uint16_t x; + memcpy(&x, p, 2); + return NUODB_BSWAP16(x); + } + case 4: { + uint32_t x; + memcpy(&x, p, 4); + return NUODB_BSWAP32(x); + } + case 8: { + uint64_t x; + memcpy(&x, p, 8); + return NUODB_BSWAP64(x); + } + default: { + unsigned long long v = 0; + for (int i = 0; i < n; i++) { + v = (v << 8) | p[i]; + } + return v; + } + } +} + +/* Sign-extends the n-byte big-endian unsigned value from _pynuodb_be_u64*/ +NUODB_INLINE long long _pynuodb_be_i64(const unsigned char *p, int n) +{ + if (n <= 0) { + return 0; + } + unsigned long long v = _pynuodb_be_u64(p, n); + unsigned long long m = (n < 8) ? (1ULL << ((n << 3) - 1)) : 0x8000000000000000ULL; + return (long long)((v ^ m) - m); +} + +NUODB_INLINE double _pynuodb_be_double(const unsigned char *p, int n) +{ + unsigned char buf[8] = {0}; + if (n < 0) { + n = 0; + } else if (n > 8) { + n = 8; + } + memcpy(buf, p, n); + uint64_t v = _pynuodb_be_u64(buf, 8); + double d; + memcpy(&d, &v, 8); + return d; +} + +/* Try to make boundary checking fast through inline and marking the happy path as likely*/ +NUODB_INLINE int _pynuodb_avail_ok(Py_ssize_t pos, Py_ssize_t need, Py_ssize_t n) +{ + return NUODB_LIKELY(need >= 0 && pos <= n - need); +} + +NUODB_INLINE PyObject *_pynuodb_pylong_be_signed(const unsigned char *p, Py_ssize_t n) +{ + if (NUODB_UNLIKELY(n > 8)) { + PyErr_SetString(PyExc_OverflowError, + "_pynuodb_pylong_be_signed: width > 8 bytes not supported"); + return NULL; + } + return PyLong_FromLongLong(_pynuodb_be_i64(p, (int)n)); +} + +#endif /* PYNUODB_CUTIL_H */ diff --git a/pynuodb/_fetch.pyx b/pynuodb/_fetch.pyx new file mode 100644 index 0000000..6122b5d --- /dev/null +++ b/pynuodb/_fetch.pyx @@ -0,0 +1,412 @@ +# cython: language_level=3 +# cython: boundscheck=False +# cython: wraparound=False +# cython: cdivision=True +"""Cython-accelerated hot paths for the NuoDB Python driver. + +Replaces result_set.ResultSet and the decode loop in +EncodedSession.fetch_result_set_next(). decode_next_batch() handles common +wire types inline; anything else goes through exotic_fn back into +EncodedSession.getValue(). +""" + +from cpython.bytes cimport PyBytes_FromStringAndSize +from cpython.bytearray cimport PyByteArray_FromStringAndSize +from cpython.tuple cimport PyTuple_New +from cpython.ref cimport PyObject + +import decimal as _decimal +import uuid as _uuid +from . import datatype as _datatype +from .exception import DataError, EndOfStream + +_Decimal = _decimal.Decimal +_Binary = _datatype.Binary +_DateFromTicks = _datatype.DateFromTicks +_TimeFromTicks = _datatype.TimeFromTicks +_TimestampFromTicks = _datatype.TimestampFromTicks +_UUID = _uuid.UUID + +cdef tuple _POW10 = tuple(10 ** i for i in range(256)) + +cdef extern from "_cutil.h": + void _pynuodb_tuple_steal(PyObject *t, Py_ssize_t i, PyObject *o) + PyObject *_pynuodb_long_from_long(long v) except NULL + PyObject *_pynuodb_long_from_longlong(long long v) except NULL + PyObject *_pynuodb_decode_utf8(const char *s, Py_ssize_t n) except NULL + PyObject *_pynuodb_incref(PyObject *o) + PyObject *_pynuodb_none() + PyObject *_pynuodb_true() + PyObject *_pynuodb_false() + double _pynuodb_be_double(const unsigned char *p, int n) nogil + long long _pynuodb_be_i64(const unsigned char *p, int n) nogil + unsigned long long _pynuodb_be_u64(const unsigned char *p, int n) nogil + object _pynuodb_pylong_be_signed(const unsigned char *p, Py_ssize_t n) + int _pynuodb_avail_ok(Py_ssize_t pos, Py_ssize_t need, Py_ssize_t n) nogil + + +include "_codes.pxi" +include "_resultset.pxi" + + +cdef int _raise_bounds_error(Py_ssize_t pos, Py_ssize_t need, Py_ssize_t n, + const char* what) except -1: + if need < 0: + raise EndOfStream( + '%s: length prefix exceeds representable size at offset %d' + % (what.decode('ascii'), pos)) + raise EndOfStream( + '%s: end of stream reached (need %d bytes at offset %d, have %d)' + % (what.decode('ascii'), need, pos, n)) + + +cdef inline int _check_avail(Py_ssize_t pos, Py_ssize_t need, Py_ssize_t n, + const char* what) except -1: + """Raise EndOfStream if data[pos:pos+need] would run past the buffer.""" + if not _pynuodb_avail_ok(pos, need, n): + _raise_bounds_error(pos, need, n, what) + return 0 + + +cdef inline object _read_scaled_operand(const unsigned char* base, Py_ssize_t* pos, + int len0, int code, Py_ssize_t n, + int* scale_out, const char* what): + """Read a scale byte then an n-byte big-endian signed pylong operand.""" + cdef int nbytes = code - len0 + _check_avail(pos[0] + 1, 1 + nbytes, n, what) + scale_out[0] = base[pos[0] + 1] + cdef object value_obj = _pynuodb_pylong_be_signed(base + pos[0] + 2, nbytes) + pos[0] += 2 + nbytes + return value_obj + + +# helper methods for this file + +cdef inline object _make_decimal(value, int scale): + if scale == 0: + return _Decimal(value) + return _Decimal(f"{value}E{-scale}") + + +# SCALEDTIME/SCALEDTIMESTAMP wire values carry ticks at the column's own +# scale; _unpack_time_scale splits that into (seconds, microseconds) for +# TimeFromTicks/TimestampFromTicks, which both take microsecond precision. +cdef int _MICROS_SCALE = 6 # microseconds = 10 ** -_MICROS_SCALE seconds +cdef int _MICROS_PER_SEC = 1000000 # 10 ** _MICROS_SCALE + + +cdef inline object _unpack_time_scale(int scale, time_val): + cdef object shiftr = _POW10[scale] + cdef object ticks = time_val // shiftr + cdef object fraction = time_val % shiftr + cdef object micros + if scale > _MICROS_SCALE: + micros = fraction // _POW10[scale - _MICROS_SCALE] + else: + micros = fraction * _POW10[_MICROS_SCALE - scale] + if micros < 0: + micros = micros % _MICROS_PER_SEC + ticks = ticks + 1 + return ticks, micros + + +cdef inline object _make_scaled_date(date_val, int scale): + return _DateFromTicks(date_val // ((10) ** scale)) + + +cdef inline object _make_scaled_time(int scale, time_val, tz): + seconds, micros = _unpack_time_scale(scale, time_val) + return _TimeFromTicks(seconds, micros, tz) + + +cdef inline object _make_scaled_ts(int scale, stamp_val, tz): + seconds, micros = _unpack_time_scale(scale, stamp_val) + return _TimestampFromTicks(seconds, micros, tz) + + +cdef inline object _make_scaled_ts_notz(int scale, stamp_val): + seconds, micros = _unpack_time_scale(scale, stamp_val) + return _TimestampFromTicks(seconds, micros, None) + + +def decode_next_batch(bytearray data, Py_ssize_t pos, int col_count, + list results, object exotic_fn, object tz_info=None): + """Decode one server batch from the wire buffer. + + :param data: bytearray holding the raw server message (self.__input). + :param pos: read cursor (self.__inpos) at entry. + :param col_count: columns per row. + :param results: list to which decoded row-tuples are appended in place. + :param exotic_fn: callable(pos) -> (value, new_pos) for non-fast-path + types, should be EncodedSession._cython_exotic_decode. `data` must + not be resized (a bytearray with a live memoryview export refuses + this at runtime -- self.__input.extend(...), say, would raise + BufferError) or reassigned: `self.__input = ...` elsewhere is not + blocked by anything and would leave `base` pointing at the old + buffer with no error at all. + :param tz_info: tzinfo for SCALEDTIME / SCALEDTIMESTAMP construction. + tz_info=None with a tz-bearing code produces a naive value, not the + local zone. + :returns: (new_pos, complete) + """ + cdef: + Py_ssize_t n = len(data) + unsigned char[:] mv = data + const unsigned char* base + int code, nbytes, col, scale + Py_ssize_t length, marker_pos + long long ival + bint complete = False + object marker_obj, val, value_obj + object row_tup + object empty_str = u'' + PyObject* row_ptr + PyObject* empty_str_ptr = empty_str + + if n == 0: + return pos, False + + base = &mv[0] + + while pos < n: + code = base[pos] + if INTMINUS10 <= code <= INT31: + pos += 1 + if code == INT0: # marker 0 -> end of batch + complete = True + break + else: + marker_pos = pos + marker_obj, pos = exotic_fn(pos) + if pos <= marker_pos: + # exotic_fn must consume at least the marker byte, or a + # col_count == 0 batch would loop here forever. + raise EndOfStream( + 'exotic_fn did not advance past offset %d' % marker_pos) + # The reference decoder reads a row marker with getInt(), which + # only accepts integer-shaped codes (10-59) and raises for + # anything else; exotic_fn here goes through the general + # getValue() dispatch instead, so it must be re-checked the + # same way -- otherwise a marker byte for, say, a UUID gets + # silently decoded and treated as "a row follows". + if type(marker_obj) is not int: + raise DataError( + 'Not an integer: row marker at offset %d' % marker_pos) + if marker_obj == 0: + complete = True + break + + row_tup = PyTuple_New(col_count) + row_ptr = row_tup + # Wire codes handled inline below (see protocol.py for the named + # constants): 1-3 NULL/TRUE/FALSE, 10-59 integers, 61-68 SCALED + # decimal, 69-72 UTF-8 counted, 73-76 OPAQUE counted, 77-85 DOUBLE, + # 86-103 MILLISEC/NANOSEC, 104-108 TIME, 109-148 UTF-8 inline, + # 149-188 OPAQUE inline, 189-193 BLOB, 194-198 CLOB, 200 UUID, + # 201-208 SCALEDDATE, 209-216 SCALEDTIME, 217-224 SCALEDTIMESTAMP, + # 241 SCALEDTIMESTAMPNOTZ. Everything else (VECTOR, SCALEDCOUNT2/3, + # LOBSTREAM, ARRAY, DEBUGBARRIER, a future protocol code) falls + # through to exotic_fn. + for col in range(col_count): + _check_avail(pos, 1, n, b'type code') + code = base[pos] + + # Code 0 has no defined meaning and must fall through to + # exotic_fn, not match here. + if NULL_V <= code <= FALSE_V: + if code == NULL_V: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(_pynuodb_none())) + elif code == TRUE_V: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(_pynuodb_true())) + else: # FALSE_V + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(_pynuodb_false())) + pos += 1 + + elif INTMINUS10 <= code <= INTLEN8: + if code <= INT31: # inline -10 .. 31 + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_long(code - INT0)) + pos += 1 + else: # INTLEN1..INTLEN8: 52..59 + nbytes = code - INTLEN0 + pos += 1 + _check_avail(pos, nbytes, n, b'INTLEN integer') + ival = _pynuodb_be_i64(base + pos, nbytes) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_longlong(ival)) + pos += nbytes + + elif UTF8LEN0 <= code <= UTF8LEN39: + length = code - UTF8LEN0 + pos += 1 + if length: + _check_avail(pos, length, n, b'UTF8 (inline length)') + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_decode_utf8((base + pos), length)) + else: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(empty_str_ptr)) + pos += length + + elif UTF8COUNT1 <= code <= UTF8COUNT4: + nbytes = code - UTF8COUNT0 + pos += 1 + _check_avail(pos, nbytes, n, b'UTF8COUNT length prefix') + length = _pynuodb_be_u64(base + pos, nbytes) + pos += nbytes + if length: + _check_avail(pos, length, n, b'UTF8 (counted)') + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_decode_utf8((base + pos), length)) + else: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(empty_str_ptr)) + pos += length + + elif OPAQUELEN0 <= code <= OPAQUELEN39: + length = code - OPAQUELEN0 + pos += 1 + _check_avail(pos, length, n, b'OPAQUE (inline length)') + val = _Binary(PyByteArray_FromStringAndSize( + (base + pos), length)) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += length + + elif OPAQUECOUNT1 <= code <= OPAQUECOUNT4: + nbytes = code - OPAQUECOUNT0 + pos += 1 + _check_avail(pos, nbytes, n, b'OPAQUECOUNT length prefix') + length = _pynuodb_be_u64(base + pos, nbytes) + pos += nbytes + _check_avail(pos, length, n, b'OPAQUE (counted)') + val = _Binary(PyByteArray_FromStringAndSize( + (base + pos), length)) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += length + + elif DOUBLELEN0 <= code <= DOUBLELEN8: + nbytes = code - DOUBLELEN0 + pos += 1 + _check_avail(pos, nbytes, n, b'DOUBLE') + val = float(_pynuodb_be_double(base + pos, nbytes)) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += nbytes + + elif MILLISECLEN0 <= code <= NANOSECLEN8: + if code <= MILLISECLEN8: + nbytes = code - MILLISECLEN0 + else: + nbytes = code - NANOSECLEN0 + pos += 1 + if nbytes == 0: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_long(0)) + else: + _check_avail(pos, nbytes, n, b'MILLISEC/NANOSEC timestamp') + ival = _pynuodb_be_i64(base + pos, nbytes) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_longlong(ival)) + pos += nbytes + + elif TIMELEN0 <= code <= TIMELEN4: + nbytes = code - TIMELEN0 + pos += 1 + if nbytes == 0: + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_long(0)) + else: + _check_avail(pos, nbytes, n, b'TIME') + ival = _pynuodb_be_u64(base + pos, nbytes) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_long_from_longlong(ival)) + pos += nbytes + + # SCALEDLEN8 == UTF8COUNT0 == 68; this range is inside (60,68], + # distinct from UTF8COUNT (69-72) above. + elif SCALEDLEN0 < code <= SCALEDLEN8: + value_obj = _read_scaled_operand(base, &pos, SCALEDLEN0, code, n, &scale, b'SCALED decimal') + val = _make_decimal(value_obj, scale) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + elif BLOBLEN0 <= code <= BLOBLEN4: + nbytes = code - BLOBLEN0 + pos += 1 + if nbytes == 0: + length = 0 + else: + _check_avail(pos, nbytes, n, b'BLOBLEN length prefix') + length = _pynuodb_be_u64(base + pos, nbytes) + pos += nbytes + _check_avail(pos, length, n, b'BLOB') + val = _Binary(PyByteArray_FromStringAndSize( + (base + pos), length)) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += length + + elif CLOBLEN0 <= code <= CLOBLEN4: + nbytes = code - CLOBLEN0 + pos += 1 + if nbytes == 0: + length = 0 + else: + _check_avail(pos, nbytes, n, b'CLOBLEN length prefix') + length = _pynuodb_be_u64(base + pos, nbytes) + pos += nbytes + _check_avail(pos, length, n, b'CLOB') + val = PyByteArray_FromStringAndSize((base + pos), length) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += length + + # UUID (200) and SCALEDDATELEN0 (200) are the same wire code; + # the strict `<` in the SCALEDDATE check below is what keeps + # them apart, not the order these branches appear in. + elif code == UUID_C: + pos += 1 + _check_avail(pos, 16, n, b'UUID') + val = _UUID(bytes=PyBytes_FromStringAndSize((base + pos), 16)) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + pos += 16 + + elif SCALEDDATELEN0 < code <= SCALEDDATELEN8: + value_obj = _read_scaled_operand(base, &pos, SCALEDDATELEN0, code, n, &scale, b'SCALEDDATE') + val = _make_scaled_date(value_obj, scale) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + elif SCALEDTIMELEN0 < code <= SCALEDTIMELEN8: + value_obj = _read_scaled_operand(base, &pos, SCALEDTIMELEN0, code, n, &scale, b'SCALEDTIME') + val = _make_scaled_time(scale, value_obj, tz_info) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + elif SCALEDTIMESTAMPLEN0 < code <= SCALEDTIMESTAMPLEN8: + value_obj = _read_scaled_operand(base, &pos, SCALEDTIMESTAMPLEN0, code, n, &scale, b'SCALEDTIMESTAMP') + val = _make_scaled_ts(scale, value_obj, tz_info) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + # Code 241, always 8 bytes (LEN0=233). + elif code == SCALEDTIMESTAMPNOTZ: + value_obj = _read_scaled_operand(base, &pos, SCALEDTIMESTAMPNOTZLEN0, code, n, &scale, b'SCALEDTIMESTAMPNOTZ') + val = _make_scaled_ts_notz(scale, value_obj) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + else: + val, pos = exotic_fn(pos) + _pynuodb_tuple_steal(row_ptr, col, + _pynuodb_incref(val)) + + results.append(row_tup) + + return pos, complete diff --git a/pynuodb/_resultset.pxi b/pynuodb/_resultset.pxi new file mode 100644 index 0000000..d1c97fd --- /dev/null +++ b/pynuodb/_resultset.pxi @@ -0,0 +1,40 @@ +cdef class ResultSet: + """Drop-in replacement for result_set.ResultSet with C-typed attributes. + + fetchone() and is_complete() become direct C calls when invoked from + other Cython code (cpdef dispatch). Python callers see the same + interface. + """ + + cdef public int handle + cdef public int col_count + cdef public list results + cdef public int results_idx + cdef public bint complete + + def __init__(self, int handle, int col_count, list initial_results, + bint complete): + self.handle = handle + self.col_count = col_count + self.results = initial_results + self.results_idx = 0 + self.complete = complete + + def clear_results(self): + del self.results[:] + self.results_idx = 0 + + def add_row(self, row): + self.results.append(row) + + cpdef bint is_complete(self): + # Looks backwards: True if the server signalled end-of-results, OR + # there are still buffered rows unread. Matches result_set.ResultSet. + return self.complete or self.results_idx != len(self.results) + + cpdef object fetchone(self): + cdef int idx = self.results_idx + if idx == len(self.results): + return None + self.results_idx = idx + 1 + return self.results[idx] diff --git a/pynuodb/encodedsession.py b/pynuodb/encodedsession.py index 067f519..d89e887 100644 --- a/pynuodb/encodedsession.py +++ b/pynuodb/encodedsession.py @@ -37,6 +37,15 @@ from . import result_set from .datatype import LOCALZONE_NAME +# When the cython implementation is available, we will use it. +# tests toggle _HAVE_FETCH_ACCEL to do comparison tests between the +# two implementations +try: + from . import _fetch as _fetch_accel + _HAVE_FETCH_ACCEL = True +except ImportError: + _HAVE_FETCH_ACCEL = False + REMOVE_FORMAT = 0 @@ -504,6 +513,14 @@ def fetch_result_set(self, stmt): complete = False init_results = [] # type: List[result_set.Row] + if _HAVE_FETCH_ACCEL: + pos, complete = _fetch_accel.decode_next_batch( + self.__input, self.__inpos, colcount, + init_results, self._cython_exotic_decode, + self.timezone_info) + self.__inpos = pos + return result_set.ResultSet(handle, colcount, init_results, complete) + # If we hit the end of the stream without next==0, there are more # results to fetch. while self._hasBytes(1): @@ -520,6 +537,13 @@ def fetch_result_set(self, stmt): return result_set.ResultSet(handle, colcount, init_results, complete) + def _cython_exotic_decode(self, pos): + # type: (int) -> tuple + """Bridge: _fetch_accel hands wire types it doesn't fast-path back here.""" + self.__inpos = pos + val = self.getValue() + return val, self.__inpos + def fetch_result_set_next(self, resultset): # type: (result_set.ResultSet) -> None """Get more rows from this result set.""" @@ -528,6 +552,15 @@ def fetch_result_set_next(self, resultset): resultset.clear_results() + if _HAVE_FETCH_ACCEL: + pos, complete = _fetch_accel.decode_next_batch( + self.__input, self.__inpos, resultset.col_count, + resultset.results, self._cython_exotic_decode, + self.timezone_info) + self.__inpos = pos + resultset.complete = complete + return + while self._hasBytes(1): if self.getInt() == 0: resultset.complete = True diff --git a/pynuodb/result_set.py b/pynuodb/result_set.py index 2cb148a..d0eb23e 100644 --- a/pynuodb/result_set.py +++ b/pynuodb/result_set.py @@ -60,3 +60,12 @@ def fetchone(self): res = self.results[self.results_idx] self.results_idx += 1 return res + + +# Replace the Python implementation above with the Cython cdef class when the +# extension has been built. The interface is identical; fetchone() and +# is_complete() become near-C-speed cpdef calls. +try: + from ._fetch import ResultSet # noqa: F811 pylint: disable=unused-import +except ImportError: + pass diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e877754 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["setuptools>=40.8.0", "wheel", "Cython>=3.0"] +build-backend = "setuptools.build_meta" diff --git a/setup.py b/setup.py index 4e38cb3..28d86c9 100644 --- a/setup.py +++ b/setup.py @@ -19,10 +19,74 @@ is not intalled. """ +import glob import os import re -from setuptools import setup +from setuptools import Extension, setup +from setuptools.command.build_ext import build_ext + +# When building a wheel from a checkout we compile .pyx via Cython. End-user +# installs from an sdist do not need Cython: MANIFEST.in ships the generated +# .c files so the fallback branch compiles those directly. +try: + from Cython.Build import cythonize + HAS_CYTHON = True +except ImportError: + HAS_CYTHON = False + + +class _OptionalBuildExt(build_ext): + """build_ext, but the --inplace copy-back step does not abort the + install for an extension whose compile was skipped (optional=True on + a missing/broken C toolchain -- see _find_extensions()). The base + class's copy_extensions_to_source() copies every extension in + self.extensions unconditionally and errors if the built artifact + isn't there; this filters down to extensions that actually built + before delegating to it. + """ + + def copy_extensions_to_source(self): + all_exts = self.extensions + built = [] + skipped = [] + for ext in all_exts: + filename = self.get_ext_filename(self.get_ext_fullname(ext.name)) + if os.path.exists(os.path.join(self.build_lib, filename)): + built.append(ext) + else: + skipped.append(ext.name) + for name in skipped: + self.warn('not copying "%s": build was skipped' % name) + self.extensions = built + try: + build_ext.copy_extensions_to_source(self) + finally: + self.extensions = all_exts + + +def _find_extensions(): + suffix = '.pyx' if HAS_CYTHON else '.c' + sources = sorted(glob.glob(os.path.join('pynuodb', '*' + suffix))) + exts = [ + Extension(src[:-len(suffix)].replace(os.sep, '.'), [src]) + for src in sources + ] + if HAS_CYTHON and exts: + exts = cythonize(exts, compiler_directives={"language_level": "3"}) + for ext in exts: + # optional=True: a missing/broken C toolchain (no gcc/clang/MSVC) + # must not abort the whole install. encodedsession.py already + # falls back to a pure Python decode loop when `import _fetch` + # fails, so skipping the extension here just means that fallback + # is what runs. Set after cythonize(), not on the Extension above: + # cythonize() rebuilds its own Extension objects and does not + # carry this flag over from the ones it was given. + ext.optional = True + return exts + + +_ext_modules = _find_extensions() with open(os.path.join(os.path.dirname(__file__), 'pynuodb', '__init__.py')) as v: m = re.search(r"^ *__version__ *= *'(.*?)'", v.read(), re.M) @@ -40,6 +104,8 @@ description='NuoDB Python driver', keywords='nuodb scalable cloud database', packages=['pynuodb'], + ext_modules=_ext_modules, + cmdclass={'build_ext': _OptionalBuildExt}, url='https://github.com/nuodb/nuodb-python', license='BSD License', long_description=open(readme).read(), diff --git a/tests/nuodb_cython_test.py b/tests/nuodb_cython_test.py new file mode 100644 index 0000000..70daba0 --- /dev/null +++ b/tests/nuodb_cython_test.py @@ -0,0 +1,722 @@ +# -*- coding: utf-8 -*- +"""Verify that the Cython acceleration extension is built and wired in. + +(C) Copyright 2025 Dassault Systemes SE. All Rights Reserved. + +This software is licensed under a BSD 3-Clause License. +See the LICENSE file provided with this software. +""" + +import decimal +import random +import struct + +import pytest + +import pynuodb +import pynuodb.result_set as _rs +from pynuodb import protocol as _protocol + +from . import nuodb_base + + +def test_fetch_extension_importable(): + """The compiled extension module must be importable.""" + import pynuodb._fetch # noqa: F401 pylint: disable=unused-import + + +def test_result_set_is_cython(): + """pynuodb.result_set.ResultSet must be the Cython class, not the + pure-Python fallback.""" + assert _rs.ResultSet.__module__ == 'pynuodb._fetch', ( + "ResultSet came from %s; the Cython extension is not active" + % (_rs.ResultSet.__module__,)) + + +def test_decode_next_batch_exported(): + """The batch decoder used by EncodedSession must be exported.""" + import pynuodb._fetch as _fetch + assert callable(getattr(_fetch, 'decode_next_batch', None)) + + +def _no_exotic(pos): + raise AssertionError("exotic_fn should not be called by these buffers") + + +def test_decode_next_batch_rejects_truncated_length_prefix(): + """A length-prefix byte count that runs past the end of the buffer + (OPAQUECOUNT/UTF8COUNT/etc.) must raise EndOfStream, not read past the + buffer. OPAQUECOUNT0+2 = 74 claims a 2-byte length prefix but the buffer + ends right after the type code.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import EndOfStream + + buf = bytearray([51, 74]) + with pytest.raises(EndOfStream): + _fetch.decode_next_batch(buf, 0, 1, [], _no_exotic, None) + + +def test_decode_next_batch_rejects_truncated_payload(): + """A length prefix that's valid on its own but whose claimed payload + runs past the end of the buffer must raise EndOfStream. UTF8LEN0+5 = 114 + claims 5 payload bytes; only 2 are present.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import EndOfStream + + buf = bytearray([51, 114, ord('h'), ord('i')]) + with pytest.raises(EndOfStream): + _fetch.decode_next_batch(buf, 0, 1, [], _no_exotic, None) + + +def test_decode_next_batch_propagates_invalid_utf8(): + """Malformed UTF-8 in a string cell must raise UnicodeDecodeError + instead of silently storing a NULL pointer into the row tuple. + UTF8LEN0+2 = 111 claims 2 payload bytes; 0xFF is not a valid UTF-8 + start byte.""" + import pynuodb._fetch as _fetch + + buf = bytearray([51, 111, 0xFF, 0xFE]) + with pytest.raises(UnicodeDecodeError): + _fetch.decode_next_batch(buf, 0, 1, [], _no_exotic, None) + + +def test_decode_next_batch_rejects_code_zero(): + """Wire code 0 has no defined meaning in protocol.py, and pure-Python + getValue() raises DataError for it. The NULL/TRUE/FALSE fast path + (`code <= FALSE_V`) must not also match 0 and silently decode it as + False -- it must fall through to exotic_fn like every other + unrecognized code.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import DataError + + def raise_data_error(pos): + raise DataError("getValue: Invalid type code: 0") + + buf = bytearray([51, 0]) # marker (nonzero inline), then type code 0 + with pytest.raises(DataError): + _fetch.decode_next_batch(buf, 0, 1, [], raise_data_error, None) + + +def test_decode_next_batch_row_marker_exotic_fn_must_advance(): + """A non-inline row marker (outside INTMINUS10..INT31) is decoded via + exotic_fn. If exotic_fn ever returned without advancing pos, the outer + `while pos < n` loop would never terminate. Must raise instead of + hanging.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import EndOfStream + + def stuck_exotic_fn(pos): + return 1, pos # does not advance + + buf = bytearray([5]) # 5 is outside INTMINUS10(10)..INT31(51): non-inline marker + with pytest.raises(EndOfStream): + _fetch.decode_next_batch(buf, 0, 1, [], stuck_exotic_fn, None) + + +def test_decode_next_batch_zero_marker_via_exotic_fn_ends_batch(): + """The reference decoder treats *any* integer-valued row marker of 0 as + end-of-batch (`if self.getInt() == 0: complete = True`), including one + encoded as INTLEN1 rather than the inline zero code. The Cython path + routes non-inline markers through exotic_fn but never checked the + decoded value against 0, so this case fell through to decoding a row + out of whatever bytes happened to follow.""" + import pynuodb._fetch as _fetch + + def zero_via_intlen1(pos): + return 0, pos + 2 # simulates getInt() consuming INTLEN1(52) + value byte 0x00 + + buf = bytearray([52, 0]) # INTLEN1(52), value byte 0x00 + results = [] + pos, complete = _fetch.decode_next_batch(buf, 0, 1, results, zero_via_intlen1, None) + assert complete is True + assert results == [] + assert pos == 2 + + +def test_decode_next_batch_non_integer_marker_raises(): + """A row marker that isn't an integer-shaped code (10-59) must raise, + matching the reference decoder's getInt(), which raises DataError for + any other code rather than accepting whatever getValue() would have + decoded it as.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import DataError + + def non_integer_marker(pos): + return "not an int", pos + 1 + + buf = bytearray([200]) # any non-inline code; the stub ignores it + with pytest.raises(DataError): + _fetch.decode_next_batch(buf, 0, 1, [], non_integer_marker, None) + + +def test_decode_next_batch_scaled_date_rejects_one_byte_short_buffer(): + """_read_scaled_operand is called with `pos` still pointing at the + (unconsumed) type-code byte: the scale byte is at pos+1 and the data + bytes run through pos+nbytes+1. A buffer that is exactly one byte too + short to hold the last data byte must raise EndOfStream. Before the + fix, _check_avail(pos, 1+nbytes, n) was satisfied by a buffer one byte + short of what the helper actually reads, an out-of-bounds read on a + truncated buffer.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import EndOfStream + from pynuodb import protocol as _protocol + + nbytes = 4 + code = _protocol.SCALEDDATELEN0 + nbytes + full = bytearray([51, code, 2, 0x00, 0x01, 0x02, 0x03]) # marker, code, scale, 4 data bytes + truncated = full[:-1] # missing the last data byte + with pytest.raises(EndOfStream): + _fetch.decode_next_batch(truncated, 0, 1, [], _no_exotic, None) + + +# --- Synthetic wire-buffer encoder, used only by the truncation fuzz test +# below. Independent of EncodedSession's own encoder (putValue() etc.) -- +# it builds raw bytes directly from protocol.py's constants, matching the +# formats decode_next_batch documents for each branch. + +def _min_signed_bytes(value): + nbytes = 1 + while True: + try: + value.to_bytes(nbytes, 'big', signed=True) + return nbytes + except OverflowError: + nbytes += 1 + + +def _encode_null(): + return bytes([_protocol.NULL]) + + +def _encode_bool(value): + return bytes([_protocol.TRUE if value else _protocol.FALSE]) + + +def _encode_int(value): + if -10 <= value <= 31: + return bytes([_protocol.INT0 + value]) + nbytes = _min_signed_bytes(value) + return bytes([_protocol.INTLEN0 + nbytes]) + value.to_bytes(nbytes, 'big', signed=True) + + +def _encode_str(value): + payload = value.encode('utf-8') + if len(payload) <= 39: + return bytes([_protocol.UTF8LEN0 + len(payload)]) + payload + nbytes = max(1, -(-len(payload).bit_length() // 8)) + header = bytes([_protocol.UTF8COUNT0 + nbytes]) + length_bytes = len(payload).to_bytes(nbytes, 'big') + return header + length_bytes + payload + + +def _encode_bytes(value): + if len(value) <= 39: + return bytes([_protocol.OPAQUELEN0 + len(value)]) + value + nbytes = max(1, -(-len(value).bit_length() // 8)) + header = bytes([_protocol.OPAQUECOUNT0 + nbytes]) + length_bytes = len(value).to_bytes(nbytes, 'big') + return header + length_bytes + value + + +def _encode_double(value): + return bytes([_protocol.DOUBLELEN0 + 8]) + struct.pack('>d', value) + + +def _encode_uuid(raw16): + return bytes([_protocol.UUID]) + raw16 + + +def _encode_scaled(value, scale): + nbytes = _min_signed_bytes(value) + header = bytes([_protocol.SCALEDLEN0 + nbytes, scale & 0xFF]) + return header + value.to_bytes(nbytes, 'big', signed=True) + + +def _encode_row(col_bytes_list): + marker = bytes([_protocol.INT0 + 1]) # any non-zero inline marker: a row follows + return marker + b''.join(col_bytes_list) + + +def _encode_batch(rows): + body = b''.join(_encode_row(r) for r in rows) + end_marker = bytes([_protocol.INT0]) + return bytearray(body + end_marker) + + +def test_decode_next_batch_truncation_never_misbehaves(): + """Truncating a valid multi-row, multi-type buffer at every possible + length must never do anything but raise a clean exception or return + normally -- never read past the buffer, never crash. This sweeps the + bounds check ahead of every variable-length read across every column + type, truncation point, and column ordering, not just the handful of + hand-picked cases in the tests above.""" + import pynuodb._fetch as _fetch + from pynuodb.exception import EndOfStream + + columns = [ + _encode_null(), + _encode_bool(True), + _encode_bool(False), + _encode_int(5), + _encode_int(-12345), + _encode_int(9876543210), + _encode_str(''), + _encode_str('short'), + _encode_str('x' * 80), + _encode_bytes(b''), + _encode_bytes(b'\x00\x01\x02short'), + _encode_bytes(b'\xff' * 80), + _encode_double(3.5), + _encode_uuid(bytes(range(16))), + _encode_scaled(123456789, 4), + ] + + rng = random.Random(20260910) + for _ in range(15): + cols = columns[:] + rng.shuffle(cols) + full = _encode_batch([cols, cols]) + for trunc_len in range(len(full)): + buf = bytearray(full[:trunc_len]) + try: + pos, _complete = _fetch.decode_next_batch( + buf, 0, len(cols), [], _no_exotic, None) + assert 0 <= pos <= len(buf) + except (EndOfStream, UnicodeDecodeError): + pass + + +def _encode_int_width(value, nbytes): + """Like _encode_int, but forces a specific INTLEN width (1-8) instead + of picking the minimal one, so a small value can still exercise every + width _pynuodb_be_i64 supports.""" + return bytes([_protocol.INTLEN0 + nbytes]) + value.to_bytes(nbytes, 'big', signed=True) + + +def test_decode_next_batch_signed_int_matches_reference(): + """_pynuodb_be_i64 now derives every width from _pynuodb_be_u64 plus + one branchless sign-extension formula ((v ^ m) - m) instead of a + hand-unrolled switch per width, with cases 3/5/6/7 no longer even + special-cased in _pynuodb_be_u64. Sweep INTLEN1..INTLEN8 (every + nbytes 1-8), including each width's min/max boundary values where the + sign bit sits right at the edge of the mask, and compare against + Python's own signed big-endian decode.""" + import pynuodb._fetch as _fetch + + rng = random.Random(20260910) + for nbytes in range(1, 9): + lo, hi = -(1 << (nbytes * 8 - 1)), (1 << (nbytes * 8 - 1)) - 1 + candidates = {lo, hi, lo + 1, hi - 1, 0} + candidates |= {rng.randint(lo, hi) for _ in range(50)} + for value in candidates: + col = _encode_int_width(value, nbytes) + buf = _encode_batch([[col]]) + results = [] + _fetch.decode_next_batch(buf, 0, 1, results, _no_exotic, None) + assert results[0][0] == value, (nbytes, value, results[0][0]) + + +def test_decode_next_batch_scaled_decimal_matches_reference(): + """_make_decimal now builds the Decimal from a formatted string + ("%dE%d" % (value, -scale)) instead of a manually-built digit tuple. + Sweep values across every real nbytes width (1-8 signed bytes) and a + range of scale-byte values (the wire scale byte is unsigned, 0-255), + and compare the decoded Decimal -- both by == and by as_tuple(), which + would catch a same-value-different-representation divergence that == + alone would miss -- against decimal.Decimal built the same way the + reference decoder's getScaledInt() does it (encodedsession.py), + independently of _fetch.pyx.""" + import pynuodb._fetch as _fetch + + def reference_decimal(value, scale): + sign = 1 if value < 0 else 0 + digits = tuple(int(c) for c in str(abs(value))) + return decimal.Decimal((sign, digits, -scale)) + + rng = random.Random(20260910) + values = [] + for nbytes in range(1, 9): + lo, hi = -(1 << (nbytes * 8 - 1)), (1 << (nbytes * 8 - 1)) - 1 + values += [lo, hi, 0] + values += [rng.randint(lo, hi) for _ in range(20)] + scales = [0, 1, 6, 127, 255] + [rng.randint(0, 255) for _ in range(10)] + + for value in values: + for scale in scales: + col = _encode_scaled(value, scale) + buf = _encode_batch([[col]]) + results = [] + _fetch.decode_next_batch(buf, 0, 1, results, _no_exotic, None) + got = results[0][0] + expected = reference_decimal(value, scale) + assert got.as_tuple() == expected.as_tuple(), (value, scale, got, expected) + assert got == expected + + +_MIXED_TYPES_QUERY = """ + select cast(42 as int), + cast('hello' as varchar(16)), + cast(3.5 as double), + cast(99.95 as decimal(10,2)), + cast('2024-01-15' as date), + cast('12:34:56' as time), + cast('2024-01-15 12:34:56' as timestamp), + true, + null + from system.dual + union all + select cast(-1 as int), + cast('naive cafe' as varchar(16)), + cast(0.0 as double), + cast(0.00 as decimal(10,2)), + cast('1970-01-01' as date), + cast('00:00:00' as time), + cast('1970-01-01 00:00:00' as timestamp), + false, + null + from system.dual +""" + + +class TestNuoDBCython(nuodb_base.NuoBase): + def test_cython_matches_pure_python(self): + """fetchall() results must be byte-identical between the Cython + decode path and the pure-Python fallback, across one value of + every wire type the fast path covers.""" + from pynuodb import encodedsession + + if not getattr(encodedsession, '_HAVE_FETCH_ACCEL', False): + pytest.skip("Cython extension not loaded; nothing to compare") + + def run_query(): + con = self._connect() + try: + cursor = con.cursor() + cursor.execute(_MIXED_TYPES_QUERY) + return cursor.fetchall() + finally: + con.close() + + cython_rows = run_query() + + encodedsession._HAVE_FETCH_ACCEL = False + try: + python_rows = run_query() + finally: + encodedsession._HAVE_FETCH_ACCEL = True + + assert cython_rows == python_rows + + def test_cython_matches_pure_python_blob_clob(self): + """BLOB/CLOB columns must decode identically under Cython and pure + Python, across an empty value, a short value (inline OPAQUELEN/ + UTF8LEN encoding), and a long value (counted OPAQUECOUNT/UTF8COUNT + encoding, >39 bytes/chars).""" + from pynuodb import encodedsession + + if not getattr(encodedsession, '_HAVE_FETCH_ACCEL', False): + pytest.skip("Cython extension not loaded; nothing to compare") + + con = self._connect() + try: + cursor = con.cursor() + cursor.execute("DROP TABLE IF EXISTS cython_blob_clob") + cursor.execute("CREATE TABLE cython_blob_clob (b BLOB, c CLOB)") + rows = [ + (pynuodb.Binary(b''), ''), + (pynuodb.Binary(b'short blob'), 'short clob'), + (pynuodb.Binary(b'x' * 500), 'y' * 500), + ] + cursor.executemany( + "INSERT INTO cython_blob_clob (b, c) VALUES (?, ?)", rows) + con.commit() + finally: + con.close() + + def run_query(): + con2 = self._connect() + try: + cursor = con2.cursor() + cursor.execute( + "SELECT b, c FROM cython_blob_clob ORDER BY LENGTH(c)") + return cursor.fetchall() + finally: + con2.close() + + try: + cython_rows = run_query() + + encodedsession._HAVE_FETCH_ACCEL = False + try: + python_rows = run_query() + finally: + encodedsession._HAVE_FETCH_ACCEL = True + + assert len(cython_rows) == 3 + assert cython_rows == python_rows + assert [type(r[0]) for r in cython_rows] == [type(r[0]) for r in python_rows] + finally: + con = self._connect() + try: + con.cursor().execute("DROP TABLE IF EXISTS cython_blob_clob") + con.commit() + finally: + con.close() + + def test_cython_matches_pure_python_multi_batch(self): + """A result set big enough to span several server batches must + decode identically under Cython and pure Python -- this exercises + fetch_result_set_next() called repeatedly, not just the first + batch.""" + from pynuodb import encodedsession + + if not getattr(encodedsession, '_HAVE_FETCH_ACCEL', False): + pytest.skip("Cython extension not loaded; nothing to compare") + + con = self._connect() + try: + cursor = con.cursor() + cursor.execute("DROP TABLE IF EXISTS cython_ten") + cursor.execute("CREATE TABLE cython_ten (f1 INTEGER)") + cursor.execute( + "INSERT INTO cython_ten" + " VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10)") + con.commit() + finally: + con.close() + + # 10^4 rows -- well above any plausible single-batch size. + query = ("SELECT a.f1, b.f1, c.f1, d.f1" + " FROM cython_ten AS a, cython_ten AS b," + " cython_ten AS c, cython_ten AS d" + " ORDER BY a.f1, b.f1, c.f1, d.f1") + + def run_query(): + con2 = self._connect() + try: + cursor = con2.cursor() + cursor.execute(query) + return cursor.fetchall() + finally: + con2.close() + + try: + cython_rows = run_query() + + encodedsession._HAVE_FETCH_ACCEL = False + try: + python_rows = run_query() + finally: + encodedsession._HAVE_FETCH_ACCEL = True + + assert len(cython_rows) == 10000 + assert cython_rows == python_rows + finally: + con = self._connect() + try: + con.cursor().execute("DROP TABLE IF EXISTS cython_ten") + con.commit() + finally: + con.close() + + def test_empty_result_set(self): + """fetchall() on a query returning zero rows must work through + the Cython decode path (first batch arrives with complete=True + and no rows).""" + con = self._connect() + try: + cursor = con.cursor() + cursor.execute( + "select 1 from system.dual where 1 = 0") + assert cursor.fetchall() == [] + finally: + con.close() + + def test_bool_and_null_singletons(self): + """Regression: an earlier revision of _fetch.pyx evaluated + True at compile time, casting the Python bool literal + to int and yielding a junk pointer (segfault on any SELECT with + a boolean column). The current code gets None/True/False from + _pynuodb_none()/_pynuodb_true()/_pynuodb_false() in _cutil.h, + which sidesteps the issue entirely by never doing that cast in + Cython at all.""" + con = self._connect() + try: + cursor = con.cursor() + cursor.execute("select true, false, null from system.dual") + assert cursor.fetchall() == [(True, False, None)] + finally: + con.close() + + def test_exotic_type_bridge(self): + """Types the Cython fast path doesn't inline (e.g. VECTOR) must + round-trip via the _cython_exotic_decode bridge back into Python's + getValue(). VECTOR is used here because, unlike most of the other + exotic codes, it's directly reachable through the public DB-API + (cast(... as vector(...)).""" + from pynuodb.datatype import Vector + payload = Vector(Vector.DOUBLE, [0.0, 4.0, 5.0]) + con = self._connect() + try: + cursor = con.cursor() + cursor.execute( + "select cast(? as vector(3, double)) from system.dual", + [payload]) + row = cursor.fetchone() + assert list(row[0]) == [0.0, 4.0, 5.0] + finally: + con.close() + + def test_fetchone_through_cython(self): + """fetchall() goes through cursor's batch-drain path; fetchone() + is what actually invokes the Cython ResultSet.fetchone cpdef. + Make sure that path works too.""" + con = self._connect() + try: + cursor = con.cursor() + cursor.execute( + "select 1 from system.dual" + " union all select 2 from system.dual" + " union all select 3 from system.dual") + seen = [] + while True: + row = cursor.fetchone() + if row is None: + break + seen.append(row) + assert seen == [(1,), (2,), (3,)] + finally: + con.close() + + def test_integer_wire_encodings(self): + """Each NuoDB integer wire encoding (INT0..INTLEN8) gets exercised + by a different magnitude. Make sure the Cython int decoder + returns the same value as the Python one for boundary values.""" + from pynuodb import encodedsession + + values = [0, 1, -1, 127, -128, 128, -129, + 32767, -32768, 65535, + 2**31 - 1, -(2**31), 2**31, + 2**62, -(2**62)] + select_parts = ["select cast(%d as bigint) from system.dual" % v + for v in values] + query = " union all ".join(select_parts) + + def run_query(): + con = self._connect() + try: + cursor = con.cursor() + cursor.execute(query) + return cursor.fetchall() + finally: + con.close() + + cython_rows = run_query() + assert [r[0] for r in cython_rows] == values + + if getattr(encodedsession, '_HAVE_FETCH_ACCEL', False): + encodedsession._HAVE_FETCH_ACCEL = False + try: + python_rows = run_query() + finally: + encodedsession._HAVE_FETCH_ACCEL = True + assert cython_rows == python_rows + + def test_decode_next_batch_random_value_fuzz(self): + """Random values across every column type in one batch, decoded + through both the Cython and pure-Python paths, must match exactly. + Many random rows per run, covering both the inline/counted + boundary (39/40 bytes or chars) and the NULL-substitution path + for every column, rather than one fixed row of hand-picked + values.""" + from pynuodb import encodedsession + + if not getattr(encodedsession, '_HAVE_FETCH_ACCEL', False): + pytest.skip("Cython extension not loaded; nothing to compare") + + rng = random.Random(20260910) + kinds = ['int', 'str', 'double', 'decimal', 'bool', 'blob', 'clob'] + num_rows = 40 + + def random_value(kind): + if kind == 'int': + magnitude = rng.choice([10, 100, 1000, 10 ** 6, 10 ** 9, + 10 ** 15, 2 ** 62]) + return rng.randint(-magnitude, magnitude) + if kind == 'str': + length = rng.choice([0, 1, 5, 39, 40, 41, 100, 300]) + alphabet = ('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ' + '0123456789 ') + return ''.join(rng.choice(alphabet) for _ in range(length)) + if kind == 'double': + return rng.choice([0.0, -0.0, 1.5, -2.25, 3.14159265358979, + rng.uniform(-1e10, 1e10)]) + if kind == 'decimal': + sign = '-' if rng.random() < 0.5 else '' + whole = rng.randint(0, 10 ** 12) + frac = rng.randint(0, 9999) + return decimal.Decimal('%s%d.%04d' % (sign, whole, frac)) + if kind == 'bool': + return rng.choice([True, False]) + if kind == 'blob': + length = rng.choice([0, 1, 39, 40, 41, 100, 300]) + return pynuodb.Binary(bytes(rng.randrange(256) + for _ in range(length))) + if kind == 'clob': + length = rng.choice([0, 1, 39, 40, 41, 100, 300]) + alphabet = 'abcdefghijklmnopqrstuvwxyz ' + return ''.join(rng.choice(alphabet) for _ in range(length)) + raise AssertionError(kind) + + rows = [] + for i in range(num_rows): + row = [None if rng.random() < 0.1 else random_value(kind) + for kind in kinds] + rows.append(tuple(row) + (i,)) + + con = self._connect() + try: + cursor = con.cursor() + cursor.execute("DROP TABLE IF EXISTS cython_fuzz") + cursor.execute( + "CREATE TABLE cython_fuzz (" + "int_col BIGINT, str_col VARCHAR(300), dbl_col DOUBLE, " + "dec_col DECIMAL(18,4), bool_col BOOLEAN, " + "blob_col BLOB, clob_col CLOB, seq_col INTEGER)") + cursor.executemany( + "INSERT INTO cython_fuzz (int_col, str_col, dbl_col, dec_col," + " bool_col, blob_col, clob_col, seq_col)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?)", rows) + con.commit() + finally: + con.close() + + def run_query(): + con2 = self._connect() + try: + cursor = con2.cursor() + cursor.execute( + "SELECT int_col, str_col, dbl_col, dec_col, bool_col," + " blob_col, clob_col FROM cython_fuzz ORDER BY seq_col") + return cursor.fetchall() + finally: + con2.close() + + try: + cython_rows = run_query() + + encodedsession._HAVE_FETCH_ACCEL = False + try: + python_rows = run_query() + finally: + encodedsession._HAVE_FETCH_ACCEL = True + + assert len(cython_rows) == num_rows + assert cython_rows == python_rows + finally: + con = self._connect() + try: + con.cursor().execute("DROP TABLE IF EXISTS cython_fuzz") + con.commit() + finally: + con.close() diff --git a/tests/perf/test_insert_select_bench.py b/tests/perf/test_insert_select_bench.py index ab3d8b9..ae5c78f 100644 --- a/tests/perf/test_insert_select_bench.py +++ b/tests/perf/test_insert_select_bench.py @@ -25,6 +25,7 @@ import pytest +import pynuodb from tests import nuodb_base @@ -256,3 +257,34 @@ def target(): assert len(result) == _LARGE finally: con.close() + + def test_fetchall_binary_types(self, benchmark): + """SELECT over BLOB / CLOB / BINARY VARYING columns. + + Exercises the OPAQUE/BLOB/CLOB decode branches specifically (the + ones that build a bytearray/bytes object per cell), which the other + benchmarks above don't touch at all. + """ + con = self._connect() + try: + cur = con.cursor() + cur.execute("DROP TABLE IF EXISTS perf_binary") + cur.execute( + "CREATE TABLE perf_binary (b BLOB, c CLOB, v BINARY VARYING(200))") + con.commit() + rows = [(pynuodb.Binary(('blob-%d-' % i).encode() * 20), + ('clob-%d-' % i) * 20, + pynuodb.Binary(('var-%d-' % i).encode() * 5)) + for i in range(_SMALL)] + cur.executemany( + "INSERT INTO perf_binary (b, c, v) VALUES (?, ?, ?)", rows) + con.commit() + + def target(): + cur.execute("SELECT b, c, v FROM perf_binary") + return cur.fetchall() + + result = benchmark.pedantic(target, warmup_rounds=2, rounds=100, iterations=1) + assert len(result) == _SMALL + finally: + con.close()