Add Ref Cursor support for the procedure call

ballerina/Dependencies.toml

@@ -23,7 +23,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "cache"
-version = "3.7.0"
+version = "3.7.1"
 scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "constraint"},
@@ -69,7 +69,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "http"
-version = "2.10.5"
+version = "2.10.6"
 scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "auth"},
@@ -282,7 +282,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "observe"
-version = "1.2.0"
+version = "1.2.2"
 scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"}
@@ -345,6 +345,7 @@ modules = [
 org = "ballerina"
 name = "time"
 version = "2.4.0"
+scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"}
 ]

ballerina/types.bal

@@ -1321,6 +1321,19 @@ public distinct class XMLOutParameter {
     } external;
 }
 
+# Represents the Cursor Out Parameters in `sql:ParameterizedCallQuery`.
+public class CursorOutParameter {
+
+    # Parses returned SQL result set values to a ballerina stream value.
+    #
+    # + typeDesc - The `typedesc` of the record to which the result needs to be returned
+    # + return - Stream of records in the `rowType` type
+    public isolated function get(typedesc<record {}> rowType = <>) returns stream <rowType, Error?> = @java:Method {
+        'class: "io.ballerina.stdlib.sql.nativeimpl.OutParameterProcessor",
+        name: "getOutCursorValue"
+    } external;
+};
+
 # Represents SQL InOutParameter in `sql:ParameterizedCallQuery`.
 public class InOutParameter {
     Value 'in;
@@ -1340,7 +1353,7 @@ public class InOutParameter {
 }
 
 # Generic type that can be passed to `sql:ParameterizedCallQuery` to indicate procedure/function parameters.
-public type Parameter Value|InOutParameter|OutParameter;
+public type Parameter Value|InOutParameter|OutParameter|CursorOutParameter;
 
 # The object constructed through backtick surrounded strings. Dynamic parameters of `sql:Parameter` type can be indicated using `${<variable name>}`
 # such as `` `The sql:ParameterizedQuery is ${variable_name}` ``.

changelog.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Support for Cursor based result set retrieval in procedure calls
 
 ### Changed
 - [Revert Accept escaped backtick as insertions in parameterised query](https://github.com/ballerina-platform/ballerina-standard-library/issues/2056)

docs/spec/spec.md

@@ -240,6 +240,24 @@ These types can be used to retrieve values from SQL stored procedures using the
    ```
    Type of the returned value is inferred from LHS of the expression.
 
+In addition to the above parameters, it has `CursorOutParameter` to retrieve the result set from the SQL stored procedure.
+
+   ```ballerina
+   # Parses returned SQL result set values to a ballerina stream value.
+   #
+   # + typeDesc - The `typedesc` of the record to which the result needs to be returned
+   # + return - Stream of records in the `rowType` type
+   public isolated function get(typedesc<record {}> typeDesc = <>) returns stream<record {}, Error?>;
+   ```
+
+   ```ballerina
+   CursorOutParameter cursor = new;
+
+   // Execute the DB call method
+
+   stream<record{}, sql:Error?> resultStream = cursor.get();
+   ```
+
 ## 3.3. Query concatenation
 
 `sql:ParameterizedQuery` can be concatenated using util methods such as `sql:queryConcat()` and 

native/src/main/java/io/ballerina/stdlib/sql/Constants.java

@@ -56,6 +56,7 @@ private Constants() {
     public static final String STATEMENT_NATIVE_DATA_FIELD = "Statement";
     public static final String COLUMN_DEFINITIONS_DATA_FIELD = "ColumnDefinition";
     public static final String RECORD_TYPE_DATA_FIELD = "recordType";
+    public static final String REF_CURSOR_VALUE_NATIVE_DATA = "RefCursorValue";
 
     public static final String PROCEDURE_CALL_RESULT = "ProcedureCallResult";
     public static final String TYPE_DESCRIPTIONS_NATIVE_DATA_FIELD = "TypeDescription";
@@ -297,6 +298,7 @@ private OutParameterTypes() {
         public static final String BOOLEAN = "BooleanOutParameter";
         public static final String BOOLEAN_ARRAY = "BooleanArrayOutParameter";
         public static final String REF = "RefOutParameter";
+        public static final String REF_CURSOR = "CursorOutParameter";
         public static final String STRUCT = "StructOutParameter";
         public static final String XML = "XMLOutParameter";
     }

native/src/main/java/io/ballerina/stdlib/sql/nativeimpl/CallProcessor.java

@@ -159,7 +159,7 @@ private static Object nativeCallExecutable(BObject client, BObject paramSQLStrin
                 }
 
                 populateOutParameters(statement, paramSQLString, outputParamTypes,
-                        resultParameterProcessor);
+                        resultParameterProcessor, procedureCallResult);
 
                 procedureCallResult.addNativeData(STATEMENT_NATIVE_DATA_FIELD, statement);
                 procedureCallResult.addNativeData(CONNECTION_NATIVE_DATA_FIELD, connection);
@@ -227,7 +227,8 @@ private static void setCallParameters(Connection connection, CallableStatement s
 
     private static void populateOutParameters(CallableStatement statement, BObject paramSQLString,
                                               HashMap<Integer, Integer> outputParamTypes,
-                                              AbstractResultParameterProcessor resultParameterProcessor)
+                                              AbstractResultParameterProcessor resultParameterProcessor,
+                                              BObject procedureCallResult)
             throws SQLException, ApplicationError {
         if (outputParamTypes.size() == 0) {
             return;
@@ -334,7 +335,10 @@ private static void populateOutParameters(CallableStatement statement, BObject p
                     result = resultParameterProcessor.processBoolean(statement, paramIndex);
                     break;
                 case Types.REF:
+                case Types.REF_CURSOR:
                     result = resultParameterProcessor.processRef(statement, paramIndex);
+                    // This is to clean up the result set attached to the ref cursor out parameter when procedure call result is closed.
+                    procedureCallResult.addNativeData(Constants.REF_CURSOR_VALUE_NATIVE_DATA, result);
                     break;
                 case Types.STRUCT:
                     result = resultParameterProcessor.processStruct(statement, paramIndex);
@@ -462,6 +466,9 @@ private static int getOutParameterType(BObject typedValue,
             case Constants.OutParameterTypes.REF:
                 sqlTypeValue = Types.REF;
                 break;
+            case Constants.OutParameterTypes.REF_CURSOR:
+                sqlTypeValue = Types.REF_CURSOR;
+                break;
             case Constants.OutParameterTypes.STRUCT:
                 sqlTypeValue = Types.STRUCT;
                 break;

native/src/main/java/io/ballerina/stdlib/sql/nativeimpl/OutParameterProcessor.java

@@ -19,9 +19,11 @@
 package io.ballerina.stdlib.sql.nativeimpl;
 
 import io.ballerina.runtime.api.TypeTags;
+import io.ballerina.runtime.api.types.RecordType;
 import io.ballerina.runtime.api.types.Type;
 import io.ballerina.runtime.api.utils.TypeUtils;
 import io.ballerina.runtime.api.values.BObject;
+import io.ballerina.runtime.api.values.BStream;
 import io.ballerina.runtime.api.values.BTypedesc;
 import io.ballerina.stdlib.sql.Constants;
 import io.ballerina.stdlib.sql.exception.ApplicationError;
@@ -35,6 +37,7 @@
 import java.sql.Clob;
 import java.sql.Date;
 import java.sql.NClob;
+import java.sql.ResultSet;
 import java.sql.RowId;
 import java.sql.SQLException;
 import java.sql.SQLXML;
@@ -61,10 +64,22 @@ public static Object getOutParameterValue(BObject result, BTypedesc typeDesc) {
         return get(result, typeDesc, DefaultResultParameterProcessor.getInstance(), "OutParameter");
     }
 
+    public static BStream getOutCursorValue(BObject result, BTypedesc typeDesc) {
+        return get(result, typeDesc, DefaultResultParameterProcessor.getInstance());
+    }
+
     public static Object getInOutParameterValue(BObject result, BTypedesc typeDesc) {
         return get(result, typeDesc, DefaultResultParameterProcessor.getInstance(), "InOutParameter");
     }
 
+    public static BStream get(BObject result, Object recordType,
+                              AbstractResultParameterProcessor resultParameterProcessor) {
+        Object value = result.getNativeData(Constants.ParameterObject.VALUE_NATIVE_DATA);
+        RecordType streamConstraint = (RecordType) TypeUtils.getReferredType(
+                ((BTypedesc) recordType).getDescribingType());
+        return resultParameterProcessor.convertCursorValue((ResultSet) value, streamConstraint);
+    }
+
     public static Object get(BObject result, BTypedesc typeDesc,
                               AbstractResultParameterProcessor resultParameterProcessor, String parameterType) {
         int sqlType = (int) result.getNativeData(Constants.ParameterObject.SQL_TYPE_NATIVE_DATA);

native/src/main/java/io/ballerina/stdlib/sql/nativeimpl/QueryProcessor.java

@@ -44,7 +44,6 @@
 import io.ballerina.stdlib.sql.parameterprocessor.AbstractStatementParameterProcessor;
 import io.ballerina.stdlib.sql.utils.ColumnDefinition;
 import io.ballerina.stdlib.sql.utils.ErrorGenerator;
-import io.ballerina.stdlib.sql.utils.ModuleUtils;
 import io.ballerina.stdlib.sql.utils.PrimitiveTypeColumnDefinition;
 import io.ballerina.stdlib.sql.utils.Utils;
 
@@ -55,6 +54,7 @@
 import java.util.List;
 
 import static io.ballerina.stdlib.sql.datasource.SQLWorkerThreadPool.SQL_EXECUTOR_SERVICE;
+import static io.ballerina.stdlib.sql.utils.Utils.getErrorStream;
 
 /**
  * This class provides the query processing implementation which executes sql queries.
@@ -266,17 +266,6 @@ private static Object getRecordOrPrimitiveTypeBValue(
         return createValue(resultSet, 1, definition, resultParameterProcessor);
     }
 
-    private static BStream getErrorStream(Object recordType, BError errorValue) {
-        return ValueCreator.createStreamValue(
-                TypeCreator.createStreamType(((BTypedesc) recordType).getDescribingType(),
-                        PredefinedTypes.TYPE_NULL), createRecordIterator(errorValue));
-    }
-
-    private static BObject createRecordIterator(BError errorValue) {
-        return ValueCreator.createObjectValue(ModuleUtils.getModule(), Constants.RESULT_ITERATOR_OBJECT,
-                errorValue, null);
-    }
-
     public static BMap<BString, Object> createRecord(ResultSet resultSet, List<ColumnDefinition> columnDefinitions,
                                                      RecordType recordConstraint,
                                                      AbstractResultParameterProcessor resultParameterProcessor)

native/src/main/java/io/ballerina/stdlib/sql/parameterprocessor/AbstractResultParameterProcessor.java

@@ -19,12 +19,14 @@
 
 import io.ballerina.runtime.api.creators.ValueCreator;
 import io.ballerina.runtime.api.types.Field;
+import io.ballerina.runtime.api.types.RecordType;
 import io.ballerina.runtime.api.types.StructureType;
 import io.ballerina.runtime.api.types.Type;
 import io.ballerina.runtime.api.utils.JsonUtils;
 import io.ballerina.runtime.api.values.BArray;
 import io.ballerina.runtime.api.values.BError;
 import io.ballerina.runtime.api.values.BObject;
+import io.ballerina.runtime.api.values.BStream;
 import io.ballerina.runtime.api.values.BString;
 import io.ballerina.stdlib.sql.Constants;
 import io.ballerina.stdlib.sql.exception.ConversionError;
@@ -116,6 +118,8 @@ public abstract Object convertBoolean(boolean value, int sqlType, Type type, boo
 
     public abstract Object convertStruct(Struct value, int sqlType, Type type) throws DataError, SQLException;
 
+    public abstract BStream convertCursorValue(ResultSet value, RecordType recordType);
+
     public abstract Object convertXml(SQLXML value, int sqlType, Type type) throws DataError, SQLException;
 
     public abstract Object convertCustomOutParameter(Object value, String outParamObjectName, int sqlType,

native/src/main/java/io/ballerina/stdlib/sql/parameterprocessor/DefaultResultParameterProcessor.java

@@ -30,15 +30,21 @@
 import io.ballerina.runtime.api.utils.TypeUtils;
 import io.ballerina.runtime.api.utils.XmlUtils;
 import io.ballerina.runtime.api.values.BArray;
+import io.ballerina.runtime.api.values.BError;
 import io.ballerina.runtime.api.values.BMap;
 import io.ballerina.runtime.api.values.BObject;
+import io.ballerina.runtime.api.values.BStream;
 import io.ballerina.runtime.api.values.BString;
 import io.ballerina.runtime.api.values.BXml;
 import io.ballerina.stdlib.sql.Constants;
+import io.ballerina.stdlib.sql.exception.ApplicationError;
 import io.ballerina.stdlib.sql.exception.DataError;
 import io.ballerina.stdlib.sql.exception.FieldMismatchError;
 import io.ballerina.stdlib.sql.exception.TypeMismatchError;
 import io.ballerina.stdlib.sql.exception.UnsupportedTypeError;
+import io.ballerina.stdlib.sql.utils.ColumnDefinition;
+import io.ballerina.stdlib.sql.utils.ErrorGenerator;
+import io.ballerina.stdlib.sql.utils.ModuleUtils;
 import io.ballerina.stdlib.sql.utils.PrimitiveTypeColumnDefinition;
 import io.ballerina.stdlib.sql.utils.Utils;
 
@@ -62,6 +68,7 @@
 import java.util.List;
 
 import static io.ballerina.runtime.api.utils.StringUtils.fromString;
+import static io.ballerina.stdlib.sql.utils.Utils.getErrorStream;
 
 /**
  * This class implements methods required convert SQL types into ballerina types and
@@ -325,6 +332,30 @@ protected BMap<BString, Object> createUserDefinedType(Struct structValue, Struct
         return struct;
     }
 
+    public BStream convertCursorValue(ResultSet resultSet, RecordType streamConstraint) {
+        if (resultSet == null) {
+            return null;
+        }
+        try {
+            List<ColumnDefinition> columnDefinitions = Utils.getColumnDefinitions(resultSet, streamConstraint);
+            BObject resultIterator = ValueCreator.createObjectValue(ModuleUtils.getModule(),
+                    Constants.RESULT_ITERATOR_OBJECT, null, getBalStreamResultIterator());
+            resultIterator.addNativeData(Constants.RESULT_SET_NATIVE_DATA_FIELD, resultSet);
+            resultIterator.addNativeData(Constants.COLUMN_DEFINITIONS_DATA_FIELD, columnDefinitions);
+            resultIterator.addNativeData(Constants.RECORD_TYPE_DATA_FIELD, streamConstraint);
+            return ValueCreator.createStreamValue(TypeCreator.createStreamType(streamConstraint,
+                    PredefinedTypes.TYPE_NULL),
+                    resultIterator);
+        } catch (ApplicationError applicationError) {
+            BError errorValue = ErrorGenerator.getSQLApplicationError(applicationError);
+            return getErrorStream(streamConstraint, errorValue);
+        } catch (SQLException sqlException) {
+            BError errorValue = ErrorGenerator.getSQLDatabaseError(sqlException,
+                    "Error while retrieving column definition from result set.");
+            return getErrorStream(streamConstraint, errorValue);
+        }
+    }
+
     @Override
     protected void createUserDefinedTypeSubtype(Field internalField, StructureType structType)
             throws DataError {

native/src/main/java/io/ballerina/stdlib/sql/utils/ProcedureCallResultUtils.java

@@ -114,6 +114,17 @@ public static Object getNextQueryResult(
     public static Object closeCallResult(BObject procedureCallResult) {
         Statement statement = (Statement) procedureCallResult.getNativeData(Constants.STATEMENT_NATIVE_DATA_FIELD);
         Connection connection = (Connection) procedureCallResult.getNativeData(Constants.CONNECTION_NATIVE_DATA_FIELD);
+        // This is to clean up the result set attached to the ref cursor out parameter.
+        // This is to avoid the result set to be open after the call result is closed.
+        Object resultSet = procedureCallResult.getNativeData(Constants.REF_CURSOR_VALUE_NATIVE_DATA);
+        if (resultSet instanceof ResultSet) {
+            try {
+                ((ResultSet) resultSet).close();
+                procedureCallResult.addNativeData(Constants.REF_CURSOR_VALUE_NATIVE_DATA, null);
+            } catch (SQLException e) {
+                return ErrorGenerator.getSQLDatabaseError(e, "Error when closing the result set.");
+            }
+        }
         return cleanUpConnection(procedureCallResult, null, statement, connection);
     }
 }

native/src/main/java/io/ballerina/stdlib/sql/utils/Utils.java

@@ -37,7 +37,9 @@
 import io.ballerina.runtime.api.values.BMap;
 import io.ballerina.runtime.api.values.BMapInitialValueEntry;
 import io.ballerina.runtime.api.values.BObject;
+import io.ballerina.runtime.api.values.BStream;
 import io.ballerina.runtime.api.values.BString;
+import io.ballerina.runtime.api.values.BTypedesc;
 import io.ballerina.runtime.api.values.BValue;
 import io.ballerina.runtime.transactions.TransactionResourceManager;
 import io.ballerina.stdlib.sql.Constants;
@@ -1374,4 +1376,15 @@ public static void disableHikariLogs() {
             }
         }
     }
+
+    public static BStream getErrorStream(Object recordType, BError errorValue) {
+        return ValueCreator.createStreamValue(
+                TypeCreator.createStreamType(((BTypedesc) recordType).getDescribingType(),
+                        PredefinedTypes.TYPE_NULL), createRecordIterator(errorValue));
+    }
+
+    private static BObject createRecordIterator(BError errorValue) {
+        return ValueCreator.createObjectValue(ModuleUtils.getModule(), Constants.RESULT_ITERATOR_OBJECT,
+                errorValue, null);
+    }
 }