Added support for warnings and column metadata on a pipeline operation

result object.

Author: anthony-tuininga

doc/src/api_manual/pipeline.rst

@@ -229,6 +229,14 @@ PipelineOpResult objects. These objects contain the results of the executed
 PipelineOpResult Attributes
 ---------------------------
 
+.. attribute:: PipelineOpResult.columns
+
+    This read-only attribute is a list of :ref:`FetchInfo<fetchinfoobj>`
+    objects. This attribute will be None for operations that do not return
+    rows.
+
+    .. versionadded:: 2.5.0
+
 .. attribute:: PipelineOpResult.error
 
     This read-only attribute returns the error that occurred when running this
@@ -248,3 +256,11 @@ PipelineOpResult Attributes
 
     This read-only attribute returns the rows that were fetched by the
     operation, if a query was executed.
+
+.. attribute:: PipelineOpResult.warning
+
+    This read-only attribute returns any warning that was encountered when
+    running this operation. If no warning was encountered, then the value None
+    is returned. See :ref:`PL/SQL Compilation Warnings <pipelinewarning>`.
+
+    .. versionadded:: 2.5.0

doc/src/release_notes.rst

@@ -41,6 +41,11 @@ Thin Mode Changes
     method is mostly useful for applications with multiple threads concurrently
     creating connections to databases when the application starts
     (`issue 408 <https://github.com/oracle/python-oracledb/issues/408>`__).
+#)  Added attribute :data:`PipelineOpResult.warning` to provide information
+    about any warning that was encountered during the execution of a pipeline
+    operation.
+#)  Added attribute :data:`PipelineOpResult.columns` to provide information
+    about any query column metadata returned from a pipeline operation.
 #)  Fixed bug causing some pooled connections to be permanently marked as busy
     and unavailable for reuse
     (`issue 392 <https://github.com/oracle/python-oracledb/issues/392>`__).

doc/src/user_guide/asyncio.rst

@@ -305,11 +305,15 @@ Pipelining Database Operations
     In this release, pipelining support is experimental and subject to change.
 
 Pipelining allows an application to send multiple, independent statements to
-Oracle Database with one call. While the database processes the pipeline of
-statements, the application can continue with non-database work. Pipelined
-operations are executed sequentially by the database. They do not execute
-concurrently. When all the pipelined operations have executed, their results
-are returned to the application.
+Oracle Database with one call. The database can be kept busy without waiting
+for the application to receive a result set and send the next statement.  While
+the database processes the pipeline of statements, the application can continue
+with non-database work. When the database has executed all the pipelined
+operations, their results are returned to the application.
+
+Pipelined operations are executed sequentially by the database. They do not
+execute concurrently. It is local tasks that can be executed at the same time
+the database is working.
 
 Effective use of Oracle Database Pipelining can increase the responsiveness of
 an application and improve overall system throughput. Pipelining is useful when
@@ -382,7 +386,10 @@ To execute the pipeline, call :meth:`AsyncConnection.run_pipeline()`.
 
 The operations are all sent to the database and executed.  The method returns a
 list of :ref:`PipelineOpResult objects <pipelineopresultobjs>`, one entry per
-operation.
+operation. The objects contain information about the execution of the relevant
+operation, such as any error number, PL/SQL function return value, or any query
+rows and column metadata.
+
 
 The :attr:`Connection.call_timeout` value has no effect on pipeline operations.
 To limit the time for a pipeline, use an `asyncio timeout
@@ -424,20 +431,20 @@ An example of pipelining is:
 
     asyncio.run(main())
 
-See `pipelining1.py
-<https://github.com/oracle/python-oracledb/tree/main/samples/pipelining1.py>`__
+See `pipelining_basic.py
+<https://github.com/oracle/python-oracledb/tree/main/samples/pipelining_basic.py>`__
 for a runnable example.
 
-To allow the application to continue with non-database work before processing
+To allow an application to continue with non-database work before processing
 any responses from the database, use code similar to:
 
 .. code-block:: python
 
     async def run_thing_one():
-        return "run_thing_one"
+        return "thing_one"
 
     async def run_thing_two():
-        return "run_thing_two"
+        return "thing_two"
 
     async def main():
         connection = await oracle.connect_async(user="hr", password=userpwd,
@@ -448,29 +455,32 @@ any responses from the database, use code similar to:
         pipeline.add_fetchone("select sysdate from dual")
 
         # Run the pipeline and non-database operations concurrently
-        results = await asyncio.gather(
+        return_values = await asyncio.gather(
             run_thing_one(), run_thing_two(), connection.run_pipeline(pipeline)
         )
-        for r in results:
+
+        for r in return_values:
             if isinstance(r, list):  # the pipeline return list
-                for o in r:
-                    print(o.rows)
+                for result in r:
+                    if result.rows:
+                        for row in result.rows:
+                            print(*row, sep="\t")
             else:
-                print(r)
+                print(r)             # a local operation result
 
         await connection.close()
 
     asyncio.run(main())
 
 Output will be like::
 
-    run_thing_one
-    run_thing_two
-    [('HR',)]
-    [(datetime.datetime(2024, 8, 15, 6, 42, 30),)]
+    thing_one
+    thing_two
+    HR
+    2024-10-29 03:34:43
 
-See `pipelining2.py
-<https://github.com/oracle/python-oracledb/tree/main/samples/pipelining2.py>`__
+See `pipelining_parallel.py
+<https://github.com/oracle/python-oracledb/tree/main/samples/pipelining_parallel.py>`__
 for a runnable example.
 
 Using OUT Binds with Pipelines
@@ -582,6 +592,38 @@ will execute all operations and will display::
     Help: https://docs.oracle.com/error-help/db/ora-00942/
     [(5678,)]
 
+.. _pipelinewarning:
+
+**PL/SQL Compilation Warnings**
+
+:ref:`plsqlwarning` can be identified by checking the :ref:`PipelineOpResult
+Attribute <pipelineopresultobjs>` :attr:`PipelineOpResult.warning`.  For
+example:
+
+.. code-block:: python
+
+    pipeline.add_execute(
+        """create or replace procedure myproc as
+           begin
+              bogus;
+           end;"""
+    )
+    (result,) = await connection.run_pipeline(pipeline)
+
+    print(result.warning.full_code)
+    print(result.warning)
+
+will print::
+
+    DPY-7000
+    DPY-7000: creation succeeded with compilation errors
+
+
+See `pipelining_error.py
+<https://github.com/oracle/python-oracledb/tree/main/samples/pipelining_error.py>`__
+for a runnable example showing warnings and errors.
+
+
 Pipeline Cursor Usage
 ---------------------
 

samples/pipelining1.py samples/pipelining_basic.pysamples/pipelining1.py renamed to samples/pipelining_basic.py

@@ -23,7 +23,7 @@
 # -----------------------------------------------------------------------------
 
 # -----------------------------------------------------------------------------
-# pipelining1.py
+# pipelining_basic.py
 #
 # Demonstrates Oracle Database Pipelining.
 # True pipelining is only available when connected to Oracle Database 23ai
@@ -43,19 +43,48 @@ async def main():
         params=sample_env.get_connect_params(),
     )
 
+    # ------------------------------------------------------------
+
     # Create a pipeline and define the operations
     pipeline = oracledb.create_pipeline()
+
     pipeline.add_fetchone("select user from dual")
+
     pipeline.add_fetchone("select sysdate from dual")
 
+    rows = [
+        (1, "First"),
+        (2, "Second"),
+        (3, "Third"),
+        (4, "Fourth"),
+        (5, "Fifth"),
+        (6, "Sixth"),
+    ]
+    pipeline.add_executemany(
+        "insert into mytab(id, data) values (:1, :2)", rows
+    )
+
+    # pipeline.add_commit()  # uncomment to persist data
+
+    pipeline.add_fetchall("select * from mytab")
+
     # Run the operations in the pipeline.
     # Note although the database receives all the operations at the same time,
     # it will execute each operation sequentially
-    result_1, result_2 = await connection.run_pipeline(pipeline)
+    results = await connection.run_pipeline(pipeline)
+
+    # Print the query results
+    for i, result in enumerate(results):
+        if result.rows:
+            statement = pipeline.operations[i].statement
+            print(f"\nRows from operation {i+1} '{statement}':\n")
+            headings = [col.name for col in result.columns]
+            print(*headings, sep="\t")
+            print("--")
+            for row in result.rows:
+                print(*row, sep="\t")
 
-    # Print the database responses
-    print("rows 1:", result_1.rows)
-    print("rows 2:", result_2.rows)
+    # ------------------------------------------------------------
 
     await connection.close()
 

samples/pipelining_error.py

@@ -0,0 +1,106 @@
+# -----------------------------------------------------------------------------
+# Copyright (c) 2024, Oracle and/or its affiliates.
+#
+# This software is dual-licensed to you under the Universal Permissive License
+# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
+# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+# either license.
+#
+# If you elect to accept the software under the Apache License, Version 2.0,
+# the following applies:
+#
+# 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
+#
+#    https://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.
+# -----------------------------------------------------------------------------
+
+# -----------------------------------------------------------------------------
+# pipelining_error.py
+#
+# Demonstrates warnings and errors with Oracle Database Pipelining.
+# True pipelining is only available when connected to Oracle Database 23ai
+# -----------------------------------------------------------------------------
+
+import asyncio
+
+import oracledb
+import sample_env
+
+
+async def main():
+    connection = await oracledb.connect_async(
+        user=sample_env.get_main_user(),
+        password=sample_env.get_main_password(),
+        dsn=sample_env.get_connect_string(),
+        params=sample_env.get_connect_params(),
+    )
+
+    # ------------------------------------------------------------
+
+    print("\nExecution exception:\n")
+
+    try:
+        pipeline = oracledb.create_pipeline()
+        pipeline.add_fetchall("select 1 from does_not_exist_1")
+        pipeline.add_fetchall("select 2 from does_not_exist_2")
+
+        # By default, the first failure throws an exception
+        await connection.run_pipeline(pipeline)
+
+    except oracledb.Error as e:
+        (error,) = e.args
+        print(error.message)
+
+    # ------------------------------------------------------------
+
+    print("\nContinuing after first error:\n")
+
+    pipeline = oracledb.create_pipeline()
+    pipeline.add_execute(
+        """create or replace procedure myproc as
+           begin
+              bogus;
+           end;"""
+    )
+    pipeline.add_fetchall("select 1 from does_not_exist_3")
+    pipeline.add_fetchall("select 2 from does_not_exist_4")
+    pipeline.add_fetchall("select dummy from dual")
+    results = await connection.run_pipeline(pipeline, continue_on_error=True)
+
+    for i, result in enumerate(results):
+        statement = pipeline.operations[i].statement
+        if result.warning:
+            print(
+                f"Warning {result.warning.full_code} " f"in operation {i+1}:\n"
+            )
+            print(statement)
+            print(f"{result.warning}\n")
+        elif result.error:
+            print(
+                f"Error {result.error.full_code} "
+                f"at position {result.error.offset+1} "
+                f"in operation {i+1}:\n"
+            )
+            print(statement)
+            print(f"{result.error}\n")
+        elif result.rows:
+            print(f"Rows from operation {i+1}:\n")
+            print(statement)
+            for row in result.rows:
+                print(row)
+
+    # ------------------------------------------------------------
+
+    await connection.execute("drop procedure myproc")
+    await connection.close()
+
+
+asyncio.run(main())