Merge branch 'main' into ma/examples-only

Author: marregui

TODO.rst

@@ -5,9 +5,9 @@ TODO
 
 Build Tooling
 =============
-* **[HIGH]** Integrate tooling to build binaries for a matrix of operating
-  systems, architectures and Python versions. This will also be our CI.
-  This should help: https://github.com/pypa/cibuildwheel
+* **[MEDIUM]** Transition to Azure, move Linux arm to ARM pipeline without QEMU.
+
+* **[MEDIUM]** Automate Apple Silicon.
 
 
 Docs
@@ -18,19 +18,14 @@ Docs
   See: https://www.sphinx-doc.org/en/master/usage/restructuredtext/directives.html#directive-literalinclude
   The examples should be in the ``examples/`` directory in the repo.
 
-* **[MEDIUM]** Document on a per-version basis.
-
 * **[MEDIUM]** These examples should be tested as part of the unit tests (as they
   are in the C client). This is to ensure they don't "bit rot" as the code
   changes.
 
+* **[LOW]** Document on a per-version basis.
 
 Development
 ===========
-* **[HIGH]** Review API naming!
-
-* **[HIGH]** Implement the auto-commit logic based on a watermark.
-
 * **[MEDIUM]** Once we're done with them, merge in changes in the ``py_client_tweaks`` branch
   of the C client.
 

docs/conf.py

@@ -49,3 +49,18 @@
 napoleon_use_ivar = True
 napoleon_use_rtype = False
 napoleon_use_param = False
+
+autodoc_default_options = {
+    'special-members': '__init__ , __str__ , __enter__ , __exit__',
+    'undoc-members': True
+}
+
+
+# def do_not_skip_dunder_members(_app, _what, name, _obj, would_skip, _options):
+#     if name in ('__init__', '__call__', '__str__', '__enter__', '__exit__'):
+#         return False
+#     return would_skip
+
+
+# def setup(app):
+#     app.connect('autodoc-skip-member', do_not_skip_dunder_members)

src/questdb/ingress.pyx

@@ -91,7 +91,7 @@ class IngressErrorCode(Enum):
     AuthError = line_sender_error_auth_error
     TlsError = line_sender_error_tls_error
 
-    def __str__(self):
+    def __str__(self) -> str:
         return self.name
 
 
@@ -266,14 +266,13 @@ cdef class TimestampNanos:
 
 
 cdef class Sender
+cdef class Buffer
 
 
-cdef int may_flush_on_row_complete(
-    line_sender_buffer* buf,
-    Sender sender) except -1:
+cdef int may_flush_on_row_complete(Buffer buffer, Sender sender) except -1:
     if sender._auto_flush_enabled:
-        if line_sender_buffer_size(buf) >= sender._auto_flush_watermark:
-            print('w00t')
+        if len(buffer) >= sender._auto_flush_watermark:
+            sender.flush(buffer)
 
 
 cdef class Buffer:
@@ -307,6 +306,8 @@ cdef class Buffer:
       * To see the contents, call ``str(buffer)``.
     """
     cdef line_sender_buffer* _impl
+    cdef size_t _init_capacity
+    cdef size_t _max_name_len
     cdef object _row_complete_sender
 
     def __cinit__(self, init_capacity: int=65536, max_name_len: int=127):
@@ -320,12 +321,33 @@ cdef class Buffer:
     cdef inline _cinit_impl(self, size_t init_capacity, size_t max_name_len):
         self._impl = line_sender_buffer_with_max_name_len(max_name_len)
         line_sender_buffer_reserve(self._impl, init_capacity)
+        self._init_capacity = init_capacity
+        self._max_name_len = max_name_len
         self._row_complete_sender = None
 
     def __dealloc__(self):
         self._row_complete_sender = None
         line_sender_buffer_free(self._impl)
 
+    @property
+    def init_capacity(self) -> int:
+        """
+        The initial capacity of the buffer when first created.
+
+        This may grow over time, see ``capacity()``.
+        """
+        return self._init_capacity
+
+    @property
+    def max_name_len(self) -> int:
+        """Maximum length of a table or column name."""
+        return self._max_name_len
+
+    @property
+    def max_name_len(self) -> int:
+        """Maximum length of a table or column name."""
+        return self._max_name_len
+
     def reserve(self, additional: int):
         """
         Ensure the buffer has at least `additional` bytes of future capacity.
@@ -476,9 +498,7 @@ cdef class Buffer:
         if self._row_complete_sender != None:
             sender = PyWeakref_GetObject(self._row_complete_sender)
             if sender != NULL:
-                may_flush_on_row_complete(
-                    self._impl,
-                    <Sender><object>sender)
+                may_flush_on_row_complete(self, <Sender><object>sender)
 
     cdef inline int _at_ts(self, TimestampNanos ts) except -1:
         cdef line_sender_error* err = NULL
@@ -772,9 +792,9 @@ cdef class Sender:
     cdef line_sender* _impl
     cdef Buffer _buffer
     cdef bint _auto_flush_enabled
-    cdef size_t _auto_flush_watermark
-    cdef object _init_capacity
-    cdef object _max_name_len
+    cdef ssize_t _auto_flush_watermark
+    cdef size_t _init_capacity
+    cdef size_t _max_name_len
 
     def __cinit__(
             self,
@@ -785,9 +805,9 @@ cdef class Sender:
             tuple auth=None,
             object tls=False,
             object read_timeout=None,
-            int init_capacity=65536,
+            int init_capacity=65536,  # 64KiB
             int max_name_len=127,
-            object auto_flush=32768):
+            object auto_flush=64512):  # 63KiB
         cdef line_sender_error* err = NULL
 
         cdef line_sender_utf8 host_utf8
@@ -887,6 +907,10 @@ cdef class Sender:
         self._auto_flush_enabled = not not auto_flush
         self._auto_flush_watermark = int(auto_flush) \
             if self._auto_flush_enabled else 0
+        if self._auto_flush_watermark < 0:
+            raise ValueError(
+                'auto_flush_watermark must be >= 0, '
+                f'not {self._auto_flush_watermark}')
 
     def new_buffer(self):
         """
@@ -895,10 +919,20 @@ cdef class Sender:
         The buffer is set up with the configured `init_capacity` and
         `max_name_len`.
         """
-        self._buffer = Buffer(
+        return Buffer(
             init_capacity=self._init_capacity,
             max_name_len=self._max_name_len)
 
+    @property
+    def init_capacity(self) -> int:
+        """The initial capacity of the sender's internal buffer."""
+        return self._init_capacity
+
+    @property
+    def max_name_len(self) -> int:
+        """Maximum length of a table or column name."""
+        return self._max_name_len
+
     def connect(self):
         cdef line_sender_error* err = NULL
         if self._opts == NULL:
@@ -929,6 +963,9 @@ cdef class Sender:
         self._buffer.row(*args, **kwargs)
 
     cpdef flush(self, Buffer buffer=None, bint clear=True):
+        if buffer is None and not clear:
+            raise ValueError('The internal buffer must always be cleared.')
+
         cdef line_sender_error* err = NULL
         cdef line_sender_buffer* c_buf = NULL
         if self._impl == NULL:

test/test.py

@@ -176,6 +176,16 @@ def test_flush_3(self):
                         sender.row('tbl1', symbols={'a': 'b'})
                         sender.flush()
 
+    def test_flush_4(self):
+        # Clearing of the internal buffer is not allowed.
+        with Server() as server:
+            with self.assertRaises(ValueError):
+                with qi.Sender('localhost', server.port) as sender:
+                    server.accept()
+                    sender.row('tbl1', symbols={'a': 'b'})
+                    sender.flush(buffer=None, clear=False)
+
+
     def test_independent_buffer(self):
         buf = qi.Buffer()
         buf.row('tbl1', symbols={'sym1': 'val1'})
@@ -211,8 +221,50 @@ def test_auto_flush(self):
             with qi.Sender('localhost', server.port, auto_flush=4) as sender:
                 server.accept()
                 sender.row('tbl1', symbols={'sym1': 'val1'})
-                # msgs = server.recv()
-                # self.assertEqual(msgs, [b'tbl1,sym1=val1', b'tbl1,sym2=val2'])
+                self.assertEqual(len(sender), 0)  # auto-flushed buffer.
+                msgs = server.recv()
+                self.assertEqual(msgs, [b'tbl1,sym1=val1'])
+
+    def test_immediate_auto_flush(self):
+        with Server() as server:
+            with qi.Sender('localhost', server.port, auto_flush=True) as sender:
+                server.accept()
+                sender.row('tbl1', symbols={'sym1': 'val1'})
+                self.assertEqual(len(sender), 0)  # auto-flushed buffer.
+                msgs = server.recv()
+                self.assertEqual(msgs, [b'tbl1,sym1=val1'])
+
+    def test_dont_auto_flush(self):
+        with Server() as server:
+            with qi.Sender('localhost', server.port, auto_flush=0) as sender:
+                server.accept()
+                while len(sender) < 131072:  # 128KiB
+                    sender.row('tbl1', symbols={'sym1': 'val1'})
+                msgs = server.recv()
+                self.assertEqual(msgs, [])
+
+    def test_dont_flush_on_exception(self):
+        with Server() as server:
+            with self.assertRaises(RuntimeError):
+                with qi.Sender('localhost', server.port) as sender:
+                    server.accept()
+                    sender.row('tbl1', symbols={'sym1': 'val1'})
+                    self.assertEqual(str(sender), 'tbl1,sym1=val1\n')
+                    raise RuntimeError('Test exception')
+            msgs = server.recv()
+            self.assertEqual(msgs, [])
+
+    def test_new_buffer(self):
+        sender = qi.Sender(
+            host='localhost',
+            port=9009,
+            init_capacity=1024,
+            max_name_len=10)
+        buffer = sender.new_buffer()
+        self.assertEqual(buffer.init_capacity, 1024)
+        self.assertEqual(buffer.max_name_len, 10)
+        self.assertEqual(buffer.init_capacity, sender.init_capacity)
+        self.assertEqual(buffer.max_name_len, sender.max_name_len)
 
 
 if __name__ == '__main__':