<flow name="selectParameterizedQuery">
<db:select config-ref="dbConfig">
<db:sql>SELECT * FROM PLANET WHERE name = :name</db:sql>
<db:input-parameters>
#[{'name' : payload}]
</db:input-parameters>
</db:select>
</flow>
Database Connector Examples - Mule 4
Anypoint Connector for Database (Database Connector) examples help you configure database connections, query a database, execute stored procedures, execute DDL statements, execute scripts, execute database transactions, use bulk operations, and define custom data types:
-
Configure a Database Connection
Connect to both popular and generic database. Additionally, configure the connector to connect to a global data source, set a JDBC driver, configure connection pooling, and connection to an Oracle database with TNS. -
Query a Database
Configure the Select operation to query data from a database. -
Insert, Update, and Delete Data
Configure the Insert, Update and Delete operations to manage data. -
Execute Stored Procedures
Configure the Stored Procedure operation to combineinput
,output
, andinput-output
parameters. Additionally, configure dates on stored procedures. -
Call an Oracle Stored Procedure That Uses UDTs
Configure the Stored procedure operation to call an Oracle database stored procedure that uses UDTs for its input and output parameters. Additionally, usecreateStruct
andcreateArray
DataWeave functions to map application data to your custom types. -
Execute DDL Statements
Configure the Execute DDL operation to execute DDL statements that create or modify data within tables or other data structures. -
Execute Bulk Operations
Configure the Bulk insert, Bulk update, and Bulk delete operations for use when each input parameter can take only one value. -
Execute Script
Configure the Execute script operation to execute a script as a single statement. -
Define Custom Data Types
Define custom data types to use when connected to a particular connection provider. -
Execute Database Transactions
Execute database operations in the context of a transaction.
Query a Database
When you configure a database operation using Database Connector, there are several ways to add variable values to the SQL statement you execute in the database.
Use Input Parameters to Protect Database Queries
Use the select
operation to retrieve information from the RDBMS. The primary goal of this operation is to supply a SQL query and use DataWeave to supply the parameters:
In this example, you supply input parameters as key-value pairs, which you create by embedding a DataWeave script. The keys are used with the colon character (:
) to reference a parameter value by name. This is the recommended approach for using parameters in your query.
The advantages of using input parameters to configure the WHERE
clause in a SELECT
statement are that it makes the query immune to SQL injection attacks and enables optimizations that are not possible otherwise, which improves the app’s overall performance.
Since version 1.4.0 (escaping colons):
If you need to use the colon character (
|
Since version 1.8.0 (SQL Casting for PostgreSQL and Snowflake): Database Connector now accepts SQL Casting PostgreSQL and Snowflake syntax’s, through the double colon (::) expression without the need to escape each colon (this feature doesn’t affect the escaping colons behavior). For example:
|
For security reasons, do not directly write <db:sql>SELECT * FROM PLANET WHERE name = #[payload] </db:sql>
.
DataSense is available for the operation’s input and output. The connector analyzes the query and automatically calculates the structure of the query’s output by analyzing the projection section of the SQL statement. At the same time, by comparing the conditions in the WHERE
clause to the table structure, it generates DataSense input to help you build the DataWeave script that generates the input parameters.
Dynamic Queries
Sometimes, you need to parameterize not only the WHERE
clause, but also parts of the query itself. Examples of use cases for this are queries that need to access online versus historic tables that depend on a condition, or complex queries for which the project table columns need to vary.
In the following example, you can see how a full expression is used to produce the query by building a string in which the table depends on a variable $(vars.table)
. An important thing to notice is that although some of the query text is dynamic ("SELECT * FROM $(vars.table)
), the WHERE
clause is still using the best practice of defining the WHERE
condition using input parameters, in this case, WHERE name = :name
:
<set-variable variableName="table" value="PLANET"/>
<db:select config-ref="dbConfig">
<db:sql>#["SELECT * FROM $(vars.table) WHERE name = :name"]</db:sql>
<db:input-parameters>
#[{'name' : payload}]
</db:input-parameters>
</db:select>
Dynamic queries are necessary in the above example and cannot be treated like Input parameters; Input parameters can be applied only to parameters in a WHERE
clause. To modify any other part of the query, you should use DataWeave’s interpolation operator.
In Mule 3, the concept of select was split in parameterized and dynamic queries, and you couldn’t use both at the same time. You had to choose between having a dynamic query or having the advantages of using parameters (SQL Injection protection, PreparedStatement optimization, and so on). Furthermore, the syntax to do one or the other was different, so you had to learn two different ways of doing the same thing. But with Database Connector in Mule 4, you can now use both methods at the same time by using expressions in the query. |
Stream Large Results
Use streaming with queries that return many records, such as in integration use cases. In Mule 4, streaming is transparent and always enabled.
For example, if you submit a query that returns 10K rows, attempting to fetch all those rows at once results in both performance degradation, due to the big pull from the network, and the risk of running out of memory, since all the information must be loaded into RAM.
With streaming, the connector fetches and processes only part of the query at one time, which reduces load on the network and memory. This means that the connector does not fetch the 10K rows at once; instead, it fetches a smaller chunk, and once that chunk is consumed, it fetches the rest.
You can also use the new repeatable streams mechanism, which means you can have DataWeave and other components process the same stream many times, even in parallel. For more information on repeatable streams, see Streaming in Mule 4.0.
Limit Results
Mule allows the connector to handle streaming gracefully. However, that does not mean that it’s a good idea to move large chunks of data from the database to Mule. Even with streaming, a typical SQL query can return many rows, each one containing a lot of information.
The select
operation provides parameters (fetchSize
and maxRows
) to help with this:
<db:select fetchSize="200" maxRows="1000" config-ref="dbConfig">
<db:sql>select * from some_table</db:sql>
</db:select>
This syntax instructs the connector to fetch no more than 1000 rows (the maxRows
value), no more than 200 rows at a time (the fetchSize
value), significantly reducing network and memory load. The fetchSize
value is enforced differently by different JDBC driver providers and often defaults to 10
.
The combination limits the total amount of information that is retrieved (the maxRows
value) and guarantees that the data is returned from the database over the network in smaller chunks (the fetchSize
value).
Query Timeout
Sometimes database queries take a long time to execute. The following factors often cause delays in query execution:
-
An inefficient query, such as one having improper indexing that iterates over many rows
-
A busy RDBMS or network
-
A lock contention
Generally, it’s recommended to set a timeout on the query. To manage timeouts, configure queryTimeout
and queryTimeoutUnit
. The following example shows how to set a timeout for the Select operation, but all operations support setting a timeout:
<db:select queryTimeout="0" queryTimeoutUnit="SECONDS" config-ref="dbConfig">
<db:sql>select * from some_table</db:sql>
</db:select>
Insert, Update, and Delete Data
Database Connector supports insert
, update
, and delete
operations.
Similar to the select operation, these operations support dynamic queries and
parameterization by using embedded DataWeave transformations while also
supporting fetchSize
, maxRows
, and timeout
parameters. Although examples of input
parameterization, dynamic queries, and parameter usage are provided, you can consult
the SELECT
operation documentation for more information.
Insert
Suppose there is a database schema named Products
that has a table named
electronic
. The electronic
table contains columns named id
, name
, description
, price
, and discount
.
The following SQL statement creates the table:
CREATE TABLE electronic(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100),
description VARCHAR(255),
price SMALLINT,
discount TINYINT
)
Given that the table is already created and there is a connection configuration named dbConfig
,
the following operation inserts a new record in the electronic
table:
<db:insert config-ref="dbConfig">
<db:sql>
INSERT INTO electronic(name, description, price, discount)
VALUES ('Coffee Machine', 'Model: XYZ99. Uses small size capsules.', 120, 5)
</db:sql>
</db:insert>
The id
parameter is not passed in the query because it is an autoincrement key that is generated automatically by the database.
To benefit from the advantages of input parameters, structure the equivalent query as follows:
|
The insert operation outputs a Statement Result
object that contains two fields: an affectedRows
integer that indicates how many rows were affected by the query, and a generatedKeys
map
that contains the autogenerated keys. In this example, the output is as follows:
{
"affectedRows": 1,
"generatedKeys": {
}
}
The generatedKeys
map is empty, although a row is successfully added to the table.
The map is empty because this information is disabled by default to avoid overhead.
To activate the information and generate keys, add the autoGenerateKeys
parameter, and then set it to true
:
<db:insert config-ref="dbConfig" autoGenerateKeys="true">
<db:sql>
INSERT INTO electronic(name, description, price, discount)
VALUES (:name, :description, :price, :discount)
</db:sql>
<db:input-parameters>#[{
name: 'Coffee Machine',
description: 'Model:XYZ99. Uses small size capsules.',
price: 120,
discount: 5}]
</db:input-parameters>
</db:insert>
If the statement generates multiple keys per affected row, you can specify which columns
should be returned by setting the parameter autoGeneratedKeysColumnNames
to a list that contains
the desired column names. For example, to ensure that only the id
column is
returned, set the parameter to a DataWeave list with a single string element:
<db:insert config-ref="dbConfig" autoGenerateKeys="true" autoGeneratedKeysColumnNames="#[['id']]">
...
</db:insert>
Update
Given the table that is defined in the insert example, the operation to update the electronic
table
to set the discount to 10 percent for all items that have a price value above 100 is:
<db:update config-ref="dbConfig"> <db:sql><![CDATA[#["UPDATE electronic SET discount = :discount WHERE price > :price"]]]></db:sql> <db:input-parameters>#[{ discount: 10, price: 100 }]</db:input-parameters> </db:update>
The <![CDATA[…]]>
wrapper allows you to use special characters, such as >
or "
, in the query.
Otherwise, you must use XML-escaped versions of those characters, such as >
and "
.
When you use the Anypoint Studio visual user interface for connectors instead of the XML code view, you can type special characters directly into the SQL Query Text
box, and Anypoint Studio automatically
changes the character to its escaped version in the XML view.
Execute Stored Procedures
Invoke stored procedures that combine input
, output`
, and input-output
parameters.
For example:
<!-- Invoke a procedure with input parameters -->
<db:stored-procedure config-ref="dbConfig">
<db:sql>{ call updatePlanetDescription('Venus', :description) }</db:sql>
<db:input-parameters>
#[{'description' : payload}]
</db:input-parameters>
</db:stored-procedure>
<!-- Invoke a procedure with input-output parameters -->
<db:stored-procedure config-ref="dbConfig">
<db:sql>{ call doubleMyInt(:myInt) }</db:sql>
<db:in-out-parameters>
<db:in-out-parameter key="myInt" value="3"/>
</db:in-out-parameters>
</db:stored-procedure>
<!-- Invoke a procedure with both input AND output parameters -->
<db:stored-procedure config-ref="dbConfig">
<db:sql>{ call multiplyInts(:int1, :int2, :result1, :int3, :result2) }</db:sql>
<db:input-parameters>
#[{
'int1' : 3,
'int2' : 4,
'int3' : 5
}]
</db:input-parameters>
<db:output-parameters>
<db:output-parameter key="result1" type="INTEGER"/>
<db:output-parameter key="result2" type="INTEGER"/>
<db:output-parameter key="myInt" type="INTEGER"/>
</db:output-parameters>
</db:stored-procedure>
Many combinations are possible.
After execution, the resulting values of the output
and input-output
parameters are available in the operation’s result. Leverage DataSense to help you get them.
For Database Connector version 1.4.0 or later, make sure that no whitespace is between the name of the stored procedure and the first parentheses.
Database Connector supports the use of callable statements with the format { call procedureName(:param1, :param2, …, :paramN) }
where parameters matches positionally, that is to say, :paramN
matches the Nth
parameter in the stored procedure declaration.
Suppose you are using an Oracle database which you have initialized using with the following table and stored procedure:
CREATE TABLE SYSTEM.employees(
employee_id INTEGER GENERATED BY DEFAULT AS IDENTITY,
employee_name VARCHAR2(100),
employee_age INTEGER,
employee_birthday TIMESTAMP,
PRIMARY KEY(employee_id)
);
CREATE PROCEDURE createEmployee(e_name VARCHAR2, e_age NUMBER, e_birth_date DATE) AS
BEGIN
INSERT INTO SYSTEM.employees(employee_name, employee_age, employee_birthday) VALUES(e_name, e_age, e_birth_date);
END;
The connector does not support named parameters in callable statements, so attempting to match parameters by name ( { call createEmployee(employee_age ⇒ :age, e_birth_date ⇒ :date, e_name ⇒ :name) }
) does not work. Parameters must be provided in the appropriate order: { call createEmployee(e_name ⇒ :name, employee_age ⇒ :age, e_birth_date ⇒ :date) }
.
Use Dates on Stored Procedures
Database Connector does not support the use of engine-specific embedded functions. For example, if you want to change a date to a specific format before calling a procedure in an Oracle database, the following approach does not work: { call createEmployee(e_name ⇒ :name, employee_age ⇒ :age, e_birth_date ⇒ TO_DATE(:date, 'YYYY-MM-DD HH:mm:ss')) }.
Use DataWeave for all data transformations before the invocation to the callable statement. Using the previous example, for instance, perform the following transformation:
<db:stored-procedure doc:name="Create Employee" config-ref="Database_Config">
<db:sql>{ call createEmployee(:name, :age, :date) }</db:sql>
<db:input-parameters>
<![CDATA[#[%dw 2.0
output application/json
fun format(d: DateTime) = d as String { format: "yyyy-MM-dd HH:mm:ss" }
---
{ 'date': format(|2019-10-31T13:00:00.000Z|), 'name': 'rick', 'age': 60 }
]]]>
</db:input-parameters>
</db:stored-procedure>
Execute DDL Statements
Data Definition Language (DDL) statements, are special types of SQL statements that do not directly access or modify data. Instead, they create, modify, or destroy data structures, stored procedures, and so on. You can also use DDL statements to create or modify data within tables or other data structures.
Create Table Example
Suppose you have a database schema named Products
, and you also have a connection named dbConfig
that is configured to access Products
. The following operation, named Execute DDL
, creates a new table named electronic
that has columns named id
, name
, description
, price
, and discount
:
<db:execute-ddl config-ref="dbConfig">
<db:sql>
CREATE TABLE electronic(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100),
description VARCHAR(255),
price SMALLINT,
discount TINYINT
)
</db:sql>
</db:execute-ddl>
The Execute DDL
operation creates a table and returns 0
unless there is an error.
Execute Bulk Operations
The insert
, update
, and delete
operations can be used for the cases in which each input parameter can take only one value. Alternatively, bulk operations allow you to run a single query using a set of parameters values.
You can avoid unnecessary steps by doing a bulk operation so that:
-
The query is parsed only once.
-
Only one database connection is required since a single statement is executed.
-
Network overhead is minimized.
-
RDBMS can execute the bulk operation atomically.
For these use cases, the connector offers three operations: <db:bulk-insert>
, <db:bulk-update>
and <db:bulk-delete>
.
These operations are similar to their single counterparts, except that instead of receiving input parameters as key-value pairs, the operations expect them as a list of key-value pairs.
For example:
<db:bulk-insert config-ref="dbConfig" >
<db:bulk-input-parameters>
#[[{'id': 2, 'name': 'George', 'lastName': 'Costanza'}, {'id': 3, 'name': 'Cosmo', 'lastName': 'Kramer'}]]
</db:bulk-input-parameters>
<db:sql>
insert into customers (id, name, lastName) values (:id, :name, :lastName)
</db:sql>
</db:bulk-insert>
If you don’t use bulk operations, when performing a delete
operation, many rows could match the criteria and get deleted if only one criteria (POSITION = X)
is provided.
The same concept applies for update. If you use UPDATE PRODUCTS set PRICE = PRICE * 0.9 where PRICE > :price
, you may want to apply a 10% discount on many products, but the price
input parameter accepts only one value.
If you want to apply different discount rates on products that have different prices, you can execute many operations.
The following example is a payload that is a list of objects of the following structure { price : number, discountRate: number}
:
<foreach>
<db:update config-ref="dbConfig">
<db:input-parameters>
#[
{
'discountRate' : payload.discountRate,
'price' : payload.price,
}
]
</db:input-parameters>
<db:sql>
UPDATE PRODUCTS set PRICE = PRICE * :discountRate where PRICE > :price
</db:sql>
</db:update>
</foreach>
The previous operation accomplishes the task but is inefficient. For each element in the list, one query must be executed for each element of the operation:
-
The query is parsed.
-
Parameters are resolved.
-
A connection to the database is acquired (either by getting one from the pool or establishing a new one).
-
All the network overhead is paid.
-
The RDBMS processes the query and applies changes.
-
The connection is released.
If an error arises while executing one of the operations (for example, if bulk insert fails to insert 1 out of 100 rows), a single exception is thrown.
It may happen that while some statements in the bulk operation can be successfully executed, some may result in an error. When this occurs, it will be up to the driver to either:
-
Stop execution immediately and ignore all remaining operations, or
-
Continue to execute the remaining statements.
In both cases, whenever an error occurs you can examine your application logs to see which caused the failure. When this occurs a single exception is thrown describing what went wrong.
Database Connector Data Types Reference
The following examples include information about database input parameters and defining custom data types you can use when connected to a particular connection provider.
Force Parameter Types
Under uncommon circumstances, you might need to force the underlying JDBC driver to coerce an input parameter to a specific type. For example, if your JDBC driver cannot determine the type of the input, or a parameter is of a custom type, you might need to apply force by specifying the type of each input parameter:
<db:bulk-insert queryTimeout="0" queryTimeoutUnit="SECONDS">
<db:sql>INSERT INTO PLANET(POSITION, NAME) VALUES (:position, :name)</db:sql>
<db:parameter-types>
<db:parameter-type key="name" type="VARCHAR" />
<db:parameter-type key="position" type="INTEGER" />
</db:parameter-types>
</db:bulk-insert>
Define Custom Data Types
There are exclusive and common parameters for each connection provider, such as Derby and Oracle. A child element of the connection provider element defines custom data types you can use when connected to a particular provider. For example:
<db:config name="dbConfig">
<db:derby-connection url="jdbc:derby:muleEmbeddedDB;create=true">
<db:column-types>
<!-- Derby uses JAVA_OBJECT for UDT-->
<db:column-type typeName="CONTACT_DETAILS" id="2000"/>
</db:column-types>
</db:derby-connection>
</db:config>
Database Transactions
You can execute database operations in the context of a transaction. Each operation has a transactionalAction
value that specifies the type of joining action that operations can take regarding the active transaction if there is one. For example, the select
operation has the following possible actions:
-
ALWAYS_JOIN
Expects a transaction to be in progress when a message is received. If there is no transaction, an error is raised. -
JOIN_IF_POSSIBLE
Joins the current transaction if one is available. Otherwise, no transaction occurs. -
NOT_SUPPORTED
Executes outside any existing transaction.
Group Operations
Sometimes you need to execute several queries atomically in the context of the same transaction. For example, during a bank account transfer, you need to subtract money from one account and add it in another, but if any of the two operations fail, roll back both:
<db:update config-ref=”db”>
<db:sql>UPDATE ACCOUNT set BALANCE = BALANCE - :money where ID = :source</db:sql>
<db:input-parameters>#[{‘money’ : payload.money, ‘source’: payload.source}]</db:input-parameters>
</db:update>
<db:update config-ref=”db”>
<db:sql>UPDATE ACCOUNT set BALANCE = BALANCE + :money where ID = :target</db:sql>
<db:input-parameters>#[{‘money’ : payload.money, ‘target’’: payload.target}]</db:input-parameters>
</db:update>
If these queries are executed in the context of an already existing transaction, the queries belong to the same transaction. If there’s no active transaction, you can start one by using the <try>
scope:
<try transactionalAction="ALWAYS_BEGIN">
<db:update config-ref="db">
<db:sql>UPDATE ACCOUNT set BALANCE = BALANCE - :money where ID = :source</db:sql>
<db:input-parameters>#[{'money' : payload.money, 'source': payload.source}]</db:input-parameters>
</db:update>
<db:update config-ref="db">
<db:sql>UPDATE ACCOUNT set BALANCE = BALANCE + :money where ID = :target</db:sql>
<db:input-parameters>#[{'money' : payload.money, 'target'': payload.target}]</db:input-parameters>
</db:update>
</try>
Execute Script Reference
This operation executes a script of random length as a single statement. Execute script differs from other operations in the following ways:
-
The script can contain multiple statements.
-
Statements can be of different types.
-
No input or output parameters are accepted.
The execute-script
operation runs any script that does not involve a SQL projection. You can use <db:execute-script>
in the following ways:
-
Embed
execute-script
in an operation. -
Reference
execute-script
from a file.
You cannot use both ways of executing a script simultaneously.
Executing a script returns an array of integer numbers, one element per each executed statement. Each number represents the number of objects affected by the statement.
Embed in an Operation
<db:execute-script config-ref="dbConfig">
<db:sql>
update PLANET set NAME='Mercury' where POSITION=0;
update PLANET set NAME='Mercury' where POSITION=4
</db:sql>
</db:execute-script>
Reference From a File
<flow name="executeScriptFromFile">
<db:execute-script config-ref="dbConfig" file="integration/executescript/bulk-script.sql" />
</flow>
The execute script operation frequently creates schemas and tables, inserts data, and performs data rotation. Data rotation is performed by nightly jobs that move and archive data into historic tables and purge on-line tables.
Execute script and bulk operations are intended for different uses. For example, although you can build a script that inserts many rows into the database, you could not conveniently provide dynamic parameters to the script. Also, you need to ensure that the script protects against a SQL injection attack.
Executing a select statement in a script returns no data and poses a data input problem.