JDBC Source

Introduction

An IJdbcSource is a database-related implementation of the ISource interface. It is the generic framework used in Atoti for fetching from external data sources and contributing in an Atoti IDatastore.

The purpose of the IJdbcSource is to load data from a database, through a JDBC driver.

Parameters

An IJdbcSource is built with an IJdbcSourceBuilder created by IJdbcSource.builder().

First you choose one of these types of IJdbcSource that mostly differ in the way the java.sql.ResultSet obtained by executing the query is parsed:

• IJdbcSourceBuilder.arrayRows() parses the ResultSet as Object[]
• IJdbcSourceBuilder.mapRows() parses the ResultSet as Map<String,Object>
• IJdbcSourceBuilder.nativeRows() parses the ResultSet as ResultSetRow
• IJdbcSourceBuilder.customRows() parses the ResultSet as specified by the provided IJdbcRowCreator

Then you can choose several parameters:

• Connection information: an IJdbcSource connects to a given database by using the provided connection information
  • Url, username, password and class name of the JDBC Driver. OR
  • Implementation of the IConnectionSupplier functional interface, which wraps the connection information. IConnectionSupplier#createConnection() provides a java.sql.Connection and loads JDBC drivers according to the implementation.
• Source Properties: these properties are either a field of the JDBC source or a property of one of its fields
  • name: specifies the JDBC source name (defaults to JDBC_SOURCE).
  • poolSize: input argument given to the Source Constructor that drives the size of the thread pool in which the task that executes and fills the channel is executed (defaults to 2). It can be an impacting factor when trying to fetch data of the underlying topic from numerous channels at the same time.
  • appendBatchSize: the max number of elements to poll from the append queue by the append threads (defaults to 1000).

An IJdbcSource does not need a topic to be instantiated, but topics must be added to the source before any query execution. This is done with IJdbcSource#addTopic(IJDBCTopic). Here's the parameters a topic can take:

• topicName: the name of the topic.
• query: the string representation of the SQL query associated with the topic.
• nbAppendThreads: the number of threads which process the raw rows and put them in the database.
• appendQueueSize: size of the append queue that receives the parsed records of a given task. This property should be set at a larger value than appendBatchSize and fetchSize, since those properties correspond to push/pop operations on a collection.
• chunkSize: size of the data chunks which store the processed rows before sending them to the database.
• fetchSize: size used to receive SQL request answer. While fetching data from a java.sql.ResultSet, we only fetch fetchSize-sized pieces of the ResultSet at once, which can be useful for controlling network usage congestion. This property can be the performance bottleneck of the source.

While fetching data with an IJdbcSource, Atoti monitors the size of the append queue. If the queue reaches full capacity, it becomes the limiting factor for the data fetching operation. In that case, one can increase the size of the thread-pool through JDBCTopic#nbAppendThreads, or change the appendBatchSize value of the JDBC Source.

Once the connection has been tested, data can be fetched from the database by calling the IJdbcSource#fetch([...]) methods:

Map<String, IJdbcFetchingInfo> fetch(
    Map<IMessageChannel<String, T>, List<Object>> channelsAndParams);

The fetch([...]) method retrieves all available data as defined by the topics exposed by the channels given in arguments. They can either be SQL queries or Java.sql.PreparedStatement. Then it fills the corresponding channels with the parsed results.

Usage Example

Create an IJdbcSource source with array rows using an IConnectionSupplier implied by the url, username, password and class name of the JDBC Driver:

final IJDBCSource<Object[]> source =
    IJDBCSource.builder()
        .arrayRows()
        .withConnectionInfo(url, username, password, driverClass)
        .withName(sourceName)
        .build();

Additional source properties can be modified for a more fine-tuned configuration.

Create the topic, here a preparedStatement, and add it to the source:

final JDBCTopic topic =
    new JDBCTopic(topicName, "SELECT DESK, BOOK, PNL FROM RECORDS WHERE DESK=?");
source.addTopic(topic);

Default Topic parameters are implied here, but can also be modified to tune the performances of the source.

An IStoreMessageChannelFactory implementation creates the channels between the topics of the source and the stores of a datastore:

final JDBCMessageChannelFactory channelFactory =
    new JDBCMessageChannelFactory(source, datastore);
final IMessageChannel<String, QfsResultSetRow> channel =
    channelFactory.createChannel(topicName, store.toString());

Specify the requested values in the topic, then execute the query and fill the channel:

source.fetch(channel, Arrays.asList("Desk A"));

Handling vectors

JDBC ARRAY type will be automatically converted to an IVector which underlying type depends on the type of the ARRAY in the database. To pick a specific vector type, core column calculators can be used: DoubleArrayJdbcColumnCalculator, FloatArrayJdbcColumnCalculator, IntegerArrayJdbcColumnCalculator or LongArrayJdbcColumnCalculator.

For instance for double vectors:

channelFactory.setCalculatedColumns(
    topicName, storeName, List.of(new DoubleArrayJdbcColumnCalculator("PNL_VECTOR")));

Some JDBC connector serialize the ARRAY type into a JSON String. There are also core calculators to handle these: JsonDoubleArrayJdbcColumnCalculator, JsonFloatArrayJdbcColumnCalculator, JsonIntegerArrayJdbcColumnCalculator or JsonLongArrayJdbcColumnCalculator. As the type is inconsistent between the JDBC ResultSet (String) and the table (IVector) when using these calculators, one must also call JDBCMessageChannelFactory#setOverridingType(String, String, int) to set the type of the column as vector.

Fetching data

IJdbcSource#fetch([...]) performs most of the task of an IJdbcSource: it executes queries on a database and puts the parsed data in channels that feed into an IDatastore. Also, if the PARSING_REPORT_ENABLED property is set to true, it returns an IJdbcFetchingInfo object for each topic, which contains various statistics about the fetched data, such as which columns it contains, how many lines were published to the datastore and how long this process took.

This is performed in parallel:

For each entry in the input map (for instance, a MessageChannel-Parameters pair), fetch([...]) creates a JDBCTask. Those are executed concurrently on a thread pool dimensioned by an input property of the source.

Here is a sequence diagram displaying the execution of the fetch([...]) method on multiple channels. JDBCTask and JDBCAppendRunnable are documented more thoroughly below.

JDBCTask

Each JDBCTask performs the following tasks:

• Executes the query on the database.

• Launches and keeps track of concurrent JDBCAppendRunnables on a thread pool dimensioned by a property of the topic related to the JDBCTask.

• Creates an append queue from which the runnables poll.

• Fetches the corresponding java.sql.ResultSet.

  To avoid network congestion, data from the ResultSet is recovered by fetchSize-sized pieces.

• Iterates over the fetchSize-sized piece of the ResultSet until the data is entirely recovered:

  • Parses the data as a collection of records by using the implementation of the current source.
  • Feeds it to the append queue.

When the ResultSet has been entirely processed, the JDBCAppendRunnables are notified, and we wait for the termination of all the linked threads to proceed and terminate the JDBCTask.

JDBCAppendRunnable

Each JDBCAppendRunnable performs the following tasks until the underlying JDBCTask sends notification that its append queue is empty and won't be filled anymore:

• Attempts to drain the queue by batchSize-sized pieces into a List<Record>.
• Pushes the records from the List into the channel's IMessage by chunkSize-sized pieces.

Monitoring JDBC Source

The JDBC Source contributes to the Health Event Monitoring, under the tags jdbc and source.

The following snippet adds the basic implementation of the listener to the handler stack.

HealthEventDispatcher.INSTANCE.addEventHandler(new LoggingJdbcHealthEventHandler());

This uses the default logger to report all JDBC operations. By default, there is no filtering on the received events.