> ## Documentation Index > Fetch the complete documentation index at: https://docs.activeviam.com/llms.txt > Use this file to discover all available pages before exploring further. # 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` * `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: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} Map fetch( Map, List> 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: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} final IJDBCSource 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: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} final IJDBCTopic 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: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} final IStoreMessageChannelFactory channelFactory = new ArrayJDBCMessageChannelFactory(source, datastore); final IMessageChannel channel = channelFactory.createChannel(topicName, storeName); ``` Specify the requested values in the topic, then execute the query and fill the channel: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} 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: ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} 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`](#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`](#jdbctask) and [`JDBCAppendRunnable`](#jdbcappendrunnable) are documented more thoroughly below. Diagram of fetch ### JDBCTask Each `JDBCTask` performs the following tasks: * Executes the query on the database. * Launches and keeps track of concurrent [`JDBCAppendRunnable`](#jdbcappendrunnable)s 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 [`JDBCAppendRunnable`](#jdbcappendrunnable)s 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`](#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`. * 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](../monitoring/health_dispatcher), under the tags *jdbc* and *source*. The following snippet adds the basic implementation of the listener to the handler stack. ```java theme={"languages":{"custom":["/engine/python-sdk/6.2/languages/pycon.tmLanguage.json"]}} 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.