Execute an Oracle Stored Procedure With Nested Table as a Parameter

The objective of this tutorial is to demonstrate the steps required to execute an Oracle stored procedure with a nested table as one of the parameters from a Mule flow.

To demonstrate the application, I will be using a simple use case of inserting employee records by calling an Oracle stored procedure with a nested table as one of the parameters. Each employee record has two columns: the employee’s department and the nested table of a data structure with employee name and employee number as attributes.

Prerequisites

• JDK 1.8.x

• Anypoint Studio 6.2+

• Mule Runtime 3.8.3 (embedded)

• Oracle 11.2 XE

Oracle Backend Tasks

To implement and demonstrate the use case with the Mule application, you need to execute the following Oracle scripts (and in this order):

Create an Oracle object type EMPLOYEE with the following PLSQL statement.

CREATE TYPE EMPLOYEE AS OBJECT
(EMPNAME VARCHAR2 (50),
  EMPNO INTEGER
);

Create an Oracle nested table EMPLOYEE_LIST of type EMPLOYEE with the following PLSQL statement.

CREATE TYPE EMPLOYEE_LIST AS TABLE OF EMPLOYEE;

Create an Oracle database table EMPLOYEE_TABLE with the second column type as a nested table.

CREATE TABLE EMPLOYEE_TABLE
(DEPTNAME VARCHAR2 (20),
 EMPLOYEES EMPLOYEE_LIST
) NESTED TABLE EMPLOYEES STORE AS NTABLE;

Create an Oracle stored procedure INSERT_EMPLOYEE with the second parameter type as a nested table.

CREATE OR REPLACE PROCEDURE INSERT_EMPLOYEE (DEPT  VARCHAR2, EMP_LIST   EMPLOYEE_LIST) AS
BEGIN
     INSERT INTO EMPLOYEE_TABLE VALUES (DEPT, EMP_LIST);
END;
/

Implementation

This application is created using Mule flows developed in Anypoint Studio that exposes endpoints to execute Oraclestored procedures by passing a nested table as one of the parameters and simultaneously queries the nested table.

There are two main flows in the application.

First Flow

The first flow executes the Oracle stored procedure with a nested table of employee details as one of the parameters using the database connector.

This flow takes the input as a CSV file through the File inbound endpoint:

<file:inbound-endpoint path="${file.path}" 
moveToPattern="#[header:INBOUND:originalFilename].backup"
moveToDirectory="${move.to.dir}" 
responseTimeout="10000" doc:name="File: Reads employee data from CSV file">
<file:filename-regex-filter pattern="^.*\.([cC][sS][vV]??)$" 
caseSensitive="false" />
</file:inbound-endpoint>

Next, the input is transformed into a string:

<object-to-string-transformer
doc:name="Object to String: Transforms InputStream into String"/>

Next, the splitter component splits the transformed string based on LF/CR characters. Here, each string represents an employee record containing employee number, employee name, and department separated with a comma:

<splitter expression =   
    "#[org.apache.commons.lang3.StringUtils.split(payload,'\n\r')]"
doc:name = "Splitter: Splits the payload based on line feed &amp; 
                   carriage return characters" />

Next, the expression component splits each string based on commas and returns an array containing the employee record fields defined in the above step:

<expression-transformerExpression =   
         "#[org.apache.commons.lang3.StringUtils.split(payload,',')]"
doc:name="Expression: Splits the string payload based on comma character" />

Next, the variable component assigns the employee’s department from the array defined in the above step into a variable:

<set-variable variableName="deptName" value="#[payload[2]]"
doc:name="Variable: Stores the department name" />

Next, the transform-message component transforms employee name and employee number array elements into a POJO:

<dw:transform-message metadata:id="b3a30f33-9f45-4d34-b4bd-76be3d4a19f"
doc:name="Transform Message: Transforms the splitted fields into 
                     a POJO">
<dw:set-payload><![CDATA[%dw 1.0
%output application/java
---
{
empName: payload[1],
empNo: payload[0]

} as :object {class: "com.oracle.nested.table.example.EmployeeObj"}]]></dw:set-payload>
</dw:transform-message>

Next, the collection-aggregator returns a collection of POJOs:

<collection-aggregator failOnTimeout="true"
doc:name="Collection Aggregator: Aggregates the POJOs into a List" />

Next, the Groovy script component creates an object of type oracle.sql.ARRAY from the collection of POJOs. Here, the Oracle PLSQL nested table is mapped with oracle.sql.ARRAY (Oracle JDBC Mapping):

<scripting:component doc:name="Script: Creates SQL ARRAY from POJO List">
<scripting:script engine = "Groovy">  
        <![CDATA[oracle.jdbc.pool.OracleDataSource ds=(oracle.jdbc.pool.OracleDataSource)muleContext.getRegistry().get("employeeDataSource");
            oracle.jdbc.OracleConnection conn = (oracle.jdbc.OracleConnection) ds.getConnection();
            Object [] emp = payload.toArray();
            oracle.sql.ARRAY array=conn.createARRAY("SCHEMA_NAME.EMPLOYEE_LIST",emp);
            payload = array;]]>
</scripting:script>
</scripting:component>

Finally, the Database connector executes the stored procedure with parameters department and an object of type oracle.sql.ARRAY:

<db:stored-procedure config-ref="Oracle_Configuration" doc:name="Database">
<db:parameterized-query><![CDATA[CALL 
             INSERT_EMPLOYEE(:dept,:emp_list)]]></db:parameterized-query>
<db:in-param name="dept" type="VARCHAR" value="#[flowVars.deptName]" />
<db:in-param name="emp_list" type="ARRAY" value="#[payload]" />
</db:stored-procedure>

Second Flow

The second flow queries the Oracle nested table using the database connector.

The flow takes the employee’s department name as a query parameter through the HTTP_Listener source. 

<http:listener config-ref="HTTP_Listener_Configuration"
path="/nestedTableRetrieve" doc:name="HTTP" />

Next, the department name is assigned to a variable using the variable component:

<set-variable variableName="deptName"
value="#[message.inboundProperties.'http.query.params'.dept]"
doc:name="Department Name" />

Next, a variable is created to store the result of the query using the variable component:

<set-variable variableName="result" value="#['']" doc:name="Result" />

Next, the SQL query is executed using the Database connector to retrieve the Oracle nested table:

<db:select config-ref="Oracle_Configuration" streaming="true" doc:name="Database">
<db:parameterized-query><![CDATA[SELECT e.EMPLOYEES FROM  
        employee_table e where DEPTNAME = #[flowVars.deptName]]]>
</db:parameterized-query>
</db:select>

Next, the Groovy component iterates the resultset using JDBC API and Collection API and subsequently compiles the result:

<scripting:component doc:name="Script">
<scripting:script engine="Groovy"><![CDATA[
java.util.Map empMap = new java.util.HashMap();
empMap.put("SCHEMA_NAME.EMPLOYEE", 
                   Class.forName("com.oracle.nested.table.example.EmployeeObj"));
resultSetIterator = payload;
while (resultSetIterator.hasNext()) {
java.util.Map map=resultSetIterator.next();
oracle.sql.ARRAY employeeArray=(oracle.sql.ARRAY)map.get('EMPLOYEES');
Object[] employees=(Object[])employeeArray.getArray(empMap);
for(emp in employees) {
com.oracle.nested.table.example.EmployeeObj employee = 
                     (com.oracle.nested.table.example.EmployeeObj)emp;
    flowVars.result = flowVars.result +  '\nEmployee No: ' + 
                         employee.getEmpNo() + '\t' + 'Employee Name: ' + 
                         employee.getEmpName() + '\n';
}}]]>
</scripting:script>
</scripting:component>

Next, we log the result using the logger component:

<logger message="#[flowVars.result]" level="INFO" doc:name="Logger" />

Finally, the result is transformed to the payload using the Set Payload transformer:

<set-payload value="#[flowVars.result]" doc:name="Set Payload" />

In the global configuration of the database connector, I have to provide the Oracle configuration via Spring Beans because they are registered in Mule registry, they can be referred from Java objects or scripts, and in our case, OracleDataSource is used in the Groovy script of the first flow to obtain the Oracle connection object. The global Spring Bean configuration details for the database connector are as follows:

<spring:beans>
<spring:bean id="employeeDataSource" name="employeeDataSource"  
              class="oracle.jdbc.pool.OracleDataSource">
<spring:property name="password" value="${password}" />
<spring:property name="URL"value="${url}" />
<spring:property name="user" value="${user}" />
<spring:property name="dataSourceName" value="ds" />
</spring:bean>
</spring:beans>

<db:oracle-config name="Oracle_Configuration"
doc:name="Oracle Configuration" dataSource-ref="employeeDataSource">
</db:oracle-config>

The property url should be defined as jdbc:oracle:thin:@server-host:1521:database-name.

The global configuration for the HTTP listener is left with the default values.

<http:listener-config name="HTTP_Listener_Configuration"
host="0.0.0.0" port="8081" doc:name="HTTP Listener Configuration" />

In this application, I have used an Oracle nested table of a user-defined type or Oracle object type named EMPLOYEE that has two attributes: EmpName and EmpNo. I will be explicitly creating a custom Java class for SQL-Java mapping for the Oracle object EMPLOYEE. In this case, the custom Java class EmployeeObj will implement the JDBC standard java.sql.SQLData interface. The following Java class is used in the Groovy script encapsulated within the script component of both flows. The Transform Message component in first flow transforms the input source into an object of the following Java class.

package com.oracle.nested.table.example;

import java.sql.SQLData;
import java.sql.SQLException;
import java.sql.SQLInput;
import java.sql.SQLOutput;

public class EmployeeObj implements SQLData {

 private int empNo;
 private String empName;
 private String sql_type;

 public EmployeeObj() throws SQLException {
  // TODO Auto-generated constructor stub
 }

 public String getEmpName() {
  return empName;
 }

 public EmployeeObj(int empNo, String empName, String sQLTypeName) {
  super();
  this.empName = empName;
  this.empNo = empNo;
  sql_type = "SCHEMA_NAME.EMPLOYEE";
 }

 public void setEmpName(String empName) {
  this.empName = empName;
 }

 public int getEmpNo() {
  return empNo;
 }

 public void setEmpNo(int empNo) {
  this.empNo = empNo;
 }

 public void setSQLTypeName(String sQLTypeName) {
  sql_type = sQLTypeName;
 }

 @Override
 public String getSQLTypeName() throws SQLException {
  // TODO Auto-generated method stub
  return "SCHEMA_NAME.EMPLOYEE";
 }

 @Override
 public void readSQL(SQLInput stream, String typeName) throws
 SQLException {
  // TODO Auto-generated method stub
  sql_type = typeName;
  empName = stream.readString();
  empNo = stream.readInt();
 }

 @Override
 public void writeSQL(SQLOutput stream) throws SQLException {
  // TODO Auto-generated method stub
  stream.writeString(empName);
  stream.writeInt(empNo);
 }

}

Now, let's run and test the application.

Now, we will run the first flow by placing a CSV file containing the following employee records in the path polled by the File inbound endpoint and execute the stored procedure.

To test if the values are inserted in the underlying Oracle database table EMPLOYEE_TABLE, we can query using the second flow. Query for the department SALES using the URL http://localhost:8081/nestedTableRetrieve?dept=SALES and you should see the following response:

The query result shown in the above figure ensures that the Oracle nested table column of the underlying database table EMPLOYEE is successfully inserted with values by executing the stored procedure with a nested table as one the parameters.

Conclusion

The CSV file must have a set of employee records pertaining to only one department at a time at the time of flow execution. If you want to put the employee records with multiple departments, then the flow logic for the first flow must be changed accordingly.