Fixed bug when calling run_pipeline() with a DML RETURNING statement

that results in an error.

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -58,6 +58,8 @@ Thin Mode Changes
 #)  Fixed bug when calling :meth:`Cursor.parse()` with autocommit enabled.
 #)  Fixed bug when parsing a :ref:`tnsnames.ora files <optnetfiles>` file with
     a connect descriptor containing an embedded comment.
+#)  Fixed bug when calling :meth:`AsyncConnection.run_pipeline()` with a DML
+    RETURNING statement that results in an error.
 #)  Fixed error message when a SQL statement is parsed containing a q-string
     without a closing quote.
 #)  Fixed bug affecting Python interpreter shut down using connection pooling

src/oracledb/impl/thin/packet.pyx

@@ -206,6 +206,7 @@ cdef class ReadBuffer(Buffer):
         list _saved_packets
         Capabilities _caps
         bint _check_request_boundary
+        bint _in_pipeline
         object _waiter
         object _loop
 
@@ -346,16 +347,20 @@ cdef class ReadBuffer(Buffer):
                              bint check_connected) except -1:
         """
         Processes a packet. If the packet is a control packet it is processed
-        immediately and discarded; otherwise, it is added to the list of saved
-        packets for this response. If the protocol supports sending the end of
-        request notification we wait for that message type to be returned at
-        the end of the packet.
+        immediately and discarded; if the packet is a marker packet and a
+        pipeline is being processed, the packet is discarded; otherwise, it is
+        added to the list of saved packets for this response. If the protocol
+        supports sending the end of request notification we wait for that
+        message type to be returned at the end of the packet.
         """
         if packet.packet_type == TNS_PACKET_TYPE_CONTROL:
             self._process_control_packet(packet)
             notify_waiter[0] = False
             if check_connected:
                 self._check_connected()
+        elif self._in_pipeline \
+                and packet.packet_type == TNS_PACKET_TYPE_MARKER:
+            notify_waiter[0] = False
         else:
             self._saved_packets.append(packet)
             notify_waiter[0] = \
@@ -396,6 +401,21 @@ cdef class ReadBuffer(Buffer):
             if data_flags == TNS_DATA_FLAGS_EOF:
                 self._pending_error_num = TNS_ERR_SESSION_SHUTDOWN
 
+    async def discard_pipeline_responses(self, ssize_t num_responses):
+        """
+        Discards the specified number of responses after the pipeline has
+        encountered an exception.
+        """
+        while num_responses > 0:
+            if not self.has_response():
+                await self.wait_for_response_async()
+            while True:
+                self._start_packet()
+                if self._current_packet.has_end_of_response():
+                    break
+            num_responses -= 1
+        self.reset_packets()
+
     cdef int notify_packet_received(self) except -1:
         """
         Notify the registered waiter that a packet has been received. This is
@@ -637,6 +657,21 @@ cdef class ReadBuffer(Buffer):
         if notify_waiter:
             self._start_packet()
 
+    cdef bint has_response(self):
+        """
+        Returns a boolean indicating if the list of saved packets contains all
+        of the packets for a response from the database. This method can only
+        be called if support for the end of response bit is present.
+        """
+        cdef:
+            Packet packet
+            ssize_t i, max_pos
+        for i in range(self._next_packet_pos, len(self._saved_packets)):
+            packet = <Packet> self._saved_packets[i]
+            if packet.has_end_of_response():
+                return True
+        return False
+
     cdef int reset_packets(self) except -1:
         """
         Resets the list of saved packets and the saved position (called when a
@@ -718,6 +753,22 @@ cdef class ReadBuffer(Buffer):
                     break
         self._start_packet()
 
+    async def wait_for_response_async(self):
+        """
+        Wait for packets to arrive in response to the request that was sent to
+        the database (using asyncio). This method will not return until the
+        complete response has been received. This requires the "end of
+        response" capability available in Oracle Database 23ai and higher. This
+        method also assumes that the current list of saved packets does not
+        contain a full response.
+        """
+        try:
+            self._check_request_boundary = True
+            self._waiter = self._loop.create_future()
+            await self._waiter
+        finally:
+            self._check_request_boundary = False
+
 
 @cython.final
 cdef class WriteBuffer(Buffer):

src/oracledb/impl/thin/protocol.pyx

@@ -446,13 +446,17 @@ cdef class Protocol(BaseProtocol):
     cdef int _receive_packet(self, Message message,
                              bint check_request_boundary=False) except -1:
         cdef:
+            bint orig_check_request_boundary
             ReadBuffer buf = self._read_buf
             uint16_t refuse_message_len
             const char_type* ptr
+        orig_check_request_boundary = buf._check_request_boundary
         buf._check_request_boundary = \
                 check_request_boundary and self._caps.supports_end_of_response
-        buf.wait_for_packets_sync()
-        buf._check_request_boundary = False
+        try:
+            buf.wait_for_packets_sync()
+        finally:
+            buf._check_request_boundary = orig_check_request_boundary
         if buf._current_packet.packet_type == TNS_PACKET_TYPE_MARKER:
             self._reset(message)
         elif buf._current_packet.packet_type == TNS_PACKET_TYPE_REFUSE:
@@ -832,7 +836,12 @@ cdef class BaseAsyncProtocol(BaseProtocol):
         await buf.wait_for_packets_async()
         buf._check_request_boundary = False
         if buf._current_packet.packet_type == TNS_PACKET_TYPE_MARKER:
-            await self._reset(in_pipeline)
+            if in_pipeline:
+                # skip to next packet as the marker packet doesn't contain any
+                # useful information
+                buf.wait_for_packets_sync()
+            else:
+                await self._reset()
         elif buf._current_packet.packet_type == TNS_PACKET_TYPE_REFUSE:
             self._write_buf._packet_sent = False
             buf.skip_raw_bytes(2)
@@ -843,26 +852,20 @@ cdef class BaseAsyncProtocol(BaseProtocol):
                 ptr = buf.read_raw_bytes(refuse_message_len)
                 message.error_info.message = ptr[:refuse_message_len].decode()
 
-    async def _reset(self, bint in_pipeline=False):
+    async def _reset(self):
         cdef:
             uint8_t marker_type, packet_type
-            uint8_t expected_marker_type
 
-        # send reset marker, if not in a pipeline
-        # NOTE: the expected marker type is different inside a pipeline!
-        if in_pipeline:
-            expected_marker_type = TNS_MARKER_TYPE_BREAK
-        else:
-            expected_marker_type = TNS_MARKER_TYPE_RESET
-            self._send_marker(self._write_buf, TNS_MARKER_TYPE_RESET)
+        # send reset marker
+        self._send_marker(self._write_buf, TNS_MARKER_TYPE_RESET)
 
         # read and discard all packets until a reset marker is received
         while True:
             packet_type = self._read_buf._current_packet.packet_type
             if packet_type == TNS_PACKET_TYPE_MARKER:
                 self._read_buf.skip_raw_bytes(2)
                 self._read_buf.read_ub1(&marker_type)
-                if marker_type == expected_marker_type:
+                if marker_type == TNS_MARKER_TYPE_RESET:
                     break
             await self._read_buf.wait_for_packets_async()
 
@@ -907,17 +910,24 @@ cdef class BaseAsyncProtocol(BaseProtocol):
         database. An end pipeline message is sent to the database and then
         the responses to all of the messages are processed.
         """
-        cdef Message message, end_message
+        cdef:
+            ssize_t num_responses_to_discard
+            ReadBuffer buf = self._read_buf
+            Message message, end_message
         end_message = conn_impl._create_message(EndPipelineMessage)
         end_message.send(self._write_buf)
+        buf._check_request_boundary = True
+        buf._in_pipeline = True
         try:
+            num_responses_to_discard = len(messages) + 1
             for message in messages:
                 try:
-                    await self._receive_packet(message,
-                                               check_request_boundary=True,
-                                               in_pipeline=True)
+                    if not buf.has_response():
+                        await buf.wait_for_response_async()
+                    buf._start_packet()
                     message.preprocess()
-                    message.process(self._read_buf)
+                    message.process(buf)
+                    num_responses_to_discard -= 1
                     self._process_call_status(conn_impl, message.call_status)
                     message._check_and_raise_exception()
                 except Exception as e:
@@ -926,11 +936,15 @@ cdef class BaseAsyncProtocol(BaseProtocol):
                     message.pipeline_result_impl._capture_err(e)
             await self._receive_packet(end_message,
                                        check_request_boundary=True)
-            end_message.process(self._read_buf)
+            end_message.process(buf)
+            num_responses_to_discard = 0
             end_message._check_and_raise_exception()
         except:
-            await self._reset()
+            await buf.discard_pipeline_responses(num_responses_to_discard)
             raise
+        finally:
+            buf._check_request_boundary = False
+            buf._in_pipeline = False
 
 
 class AsyncProtocol(BaseAsyncProtocol, asyncio.Protocol):

tests/test_7600_pipelining_async.py

@@ -872,6 +872,52 @@ async def test_7642(self):
         names = [i.name for i in results[0].columns]
         self.assertEqual(names, ["USER"])
 
+    async def test_7643(self):
+        "7643 - test DML returning with error - pipeline error"
+        out_value = self.cursor.var(oracledb.DB_TYPE_RAW)
+        pipeline = oracledb.create_pipeline()
+        pipeline.add_execute("truncate table TestTempTable")
+        pipeline.add_execute(
+            """
+            insert into TestTempTable (IntCol, StringCol1)
+            values (1, 'Value for first row')
+            returning StringCol1 into :1
+            """,
+            [out_value],
+        )
+        pipeline.add_commit()
+        pipeline.add_fetchone("select user from dual")
+        with self.assertRaisesFullCode("ORA-01465"):
+            await self.conn.run_pipeline(pipeline)
+        await self.cursor.execute("select user from dual")
+        (fetched_value,) = await self.cursor.fetchone()
+        self.assertEqual(fetched_value, test_env.get_main_user().upper())
+
+    async def test_7644(self):
+        "7644 - test DML returning with error - pipeline continue"
+        out_value = self.cursor.var(oracledb.DB_TYPE_RAW)
+        pipeline = oracledb.create_pipeline()
+        pipeline.add_execute("truncate table TestTempTable")
+        pipeline.add_execute(
+            """
+            insert into TestTempTable (IntCol, StringCol1)
+            values (1, 'Value for first row')
+            returning StringCol1 into :1
+            """,
+            [out_value],
+        )
+        pipeline.add_commit()
+        pipeline.add_fetchone("select user from dual")
+        results = await self.conn.run_pipeline(
+            pipeline, continue_on_error=True
+        )
+        self.assertEqual(results[1].error.full_code, "ORA-01465")
+        user = test_env.get_main_user().upper()
+        self.assertEqual(results[3].rows, [(user,)])
+        await self.cursor.execute("select user from dual")
+        (fetched_value,) = await self.cursor.fetchone()
+        self.assertEqual(fetched_value, test_env.get_main_user().upper())
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()