Oracle Multitable Consumer

The stage handles unsupported data types in one of the following ways:

Stops the flow

If the stage encounters an unsupported data type, the stage stops the flow after completing the processing of the previous records and displays the following error:

JDBC_37 - Unsupported type 1111 for column.

By default, the stage stops the flow.

Converts to string

If the stage encounters an unsupported data type, the stage converts the data to string when possible, and then continues processing. Not all unsupported data types can successfully be converted to string. When using this option, verify that the data is converted to string as expected.

To configure the stage to attempt to convert unsupported data types to string, on the Advanced tab, set the On Unknown Type property to Convert to String.

For example, let's say that you want to read all tables in the US_WEST and US_EAST schemas except for tables that start with dept. You enter the following schema, table name pattern, and table exclusion pattern:

• Schema - US%
• Table Name Pattern - %
• Table Exclusion Pattern - dept.*

Or, let's say that you want to read all tables from all schemas, except for the sys and system schemas. You enter the following schema, table name pattern, and schema exclusion pattern and leave the table exclusion pattern blank:

• Schema - %
• Table Name Pattern - %
• Schema Exclusion Pattern - sys|system

By default, the source reads all available data from each table when you start the flow. The source generates SQL queries using the following syntax when you start the flow:

SELECT * FROM <table> ORDER BY <offset column_1>, <offset column_2>, ...

You can make the following changes to how the source handles offset columns and initial offset values:

Override the primary key as the offset column

You can override the primary key and define another offset column or columns. Or if the table doesn’t have a primary key, you can define the offset column or columns to use.

Important: As a best practice, a user-defined offset column should be an incremental and unique column that does not contain null values. If the column is not unique - that is, multiple rows can have the same value for this column - there is a potential for data loss upon flow restart.

Having an index on this column is strongly encouraged since the underlying query uses an ORDER BY and inequality operators on this column.

Define an initial offset value

The initial offset value is a value within the offset column where you want the JDBC Multitable Consumer source to start reading. When you define an initial offset value, you must first enter the offset column name and then the value. If you are using the default primary key as the offset column, enter the name of the primary key.

If you define an initial offset value for a single offset column, the source generates SQL queries using the following syntax:

SELECT * FROM <table> ORDER BY <offset column> WHERE <offset column> > ${offset}

If you defined multiple offset columns, you must define an initial offset value for each column, in the same order that the columns are defined. The source uses the initial offset values of all columns to determine where to start reading data. For example, you override the primary key with the following offset columns: p1, p2, p3 and define an initial offset value for each column. The source generates SQL queries using the following syntax:

SELECT * FROM <table> ORDER BY p1, p2, p3 WHERE (p1 > ${offset1}) OR (p1 = ${offset1} AND p2 > ${offset2}) OR (p1 = ${offset1} AND p2 = ${offset2} AND p3 > ${offset3})

Note: Data Collector stores offsets for Datetime columns as Long values. For offset columns with a Datetime data type, enter the initial value as a Long value. You can use the time functions to transform a Datetime value to a Long value. For example, the following expression converts a date entered as a String to a Date object, and then to a Long value:

${time:dateTimeToMilliseconds(time:extractDateFromString('2017-05-01 20:15:30.915','yyyy-MM-dd HH:mm:ss.SSS'))} 

Define additional offset column conditions

You can use the expression language to define additional conditions that the source uses to determine where to start reading data. The source adds the defined condition to the WHERE clause of the SQL query.

You can use the offset:column function in the condition to access an offset column by position. For example, if you have a table with offset columns p1 and p2, then offset:column(0) returns the value of p1 while offset:column(1) returns the value of p2.

Let's say that you defined a transaction_time column as the offset column. While the source reads the table, multiple active transactions are being written to the table with the current timestamp for the transaction_time column. When the source finishes reading the first record with the current timestamp, the source continues reading with the next offset and skips some rows with the current timestamp. You can enter the following offset column condition to ensure that the source reads from all offset columns with a timestamp less than the current time:

${offset:column(0)} < ${time:now()}

If your database requires the datetime in a specific format, you can use the time:extractStringFromDate function to specify the format. For example:

${offset:column(0)} < '${time:extractStringFromDate(time:now(), "yyyy-MM-dd HH:mm:ss")}'

When performing multithreaded table processing, the Oracle Multitable Consumer source retrieves the list of tables defined in the table configuration when you start the flow. The source then uses multiple concurrent threads based on the Number of Threads property. Each thread reads data from a single table, and each table can have a maximum of one thread read from it at a time.

The Process All Available Rows from the Table batch strategy differs slightly depending on whether the source is processing full tables or partitions within a table.

Multithreaded table processing

When the source performs multithreaded table processing for all tables, each thread processes all available rows from a table. A thread runs a SQL query and processes all of the results for a table. Then, the thread switches to the next available table.

For example, let's say the source has a maximum batch size of 100 and uses two threads to read from four tables, each of which contains 1000 rows. Thread 1 runs a SQL query to create 10 batches of 100 rows each from table A, while thread 2 uses the same strategy to read data from table B.

When table A and table B are fully read, the threads switch to table C and table D and complete the same process. When thread 1 finishes reading from table C, it switches back to the next available table to read any data that has arrived since the table was processed, starting from the last saved offset.

The number of threads that can process the tables is limited by the Number of Threads property for the source.

When the tables being processed use both table and partition processing, the threads query the partitions as described below. For details on how the tables and partitions rotate through the processing queue, see Processing queue.

Multithreaded partition processing

Multithreaded partition processing is similar to multithreaded table processing, except that it works at a partition level. Each thread runs a SQL query for a partition and processes multiple batches of data from the results. When all data in the partition is processed, the thread switches to the next available partition.

The number of threads that can process partitions for each table is limited by the Number of Threads property for the source and the Max Active Partitions table property.

For details on how the tables and partitions rotate through the processing queue, see Processing queue.

The Switch Tables batch strategy differs depending on whether the source performs full table or partition processing. The number of batches that the source creates before closing a result set is based on the Batches from Result Set property.

Multithreaded table processing

When the source performs multithreaded table processing for all tables, each thread processes one batch of data, then switches to an available table and repeats the process. When a flow starts, each thread runs a SQL query, generates a result set, and processes a batch of records from the result set. The database driver caches the remaining records for the same thread to access again. Then, the thread switches to the next available table.

A table is available in the following situations:

• The table does not have a result set created by another thread.

  In this case, the thread runs a SQL query to generate a result set and processes a batch of records. The database driver caches the remaining records for later use.

• The table has a cached result set created by the same thread.

  In this case, the thread creates the batch from the result set, rather than running another SQL query.

A table is not available when the table has a result set created by another thread. A table with a cached result set can only be processed by the thread that created the result set. And a table cannot have multiple cached result sets.

The Max Batch Size property limits the number of records included in a batch. The Batches from Result Set property limits the number of batches that a thread can create from the result set. After a thread creates the configured number of batches, the thread closes the result set. This allows a different thread to read from the table.

For example, let's say a source uses two concurrent threads to read from three tables. Max Batch Size is set to 100 records, and Batches from Result Set is set to 2.

Thread 1 runs a SQL query on table A, which contains 80 records. The thread processes the 80 records in a single batch. In the meantime, thread 2 runs a SQL query on table B, which contains 5000 records, and caches the result set. The thread creates a batch to process the first 100 records in the table.

After processing table A, thread 1 looks for the next table to process. Thread 1 skips table B because the table has a cached result set from thread 2, so thread 1 runs a SQL query on table C, which contains 1000 records, and caches the result set. Thread 1 processes the first 100 records in the table.

Similarly, when thread 2 completes the batch on table B, it skips table C because thread 1 has a result set cached for the table. If table A had new data that came in, thread 2 would process that data, since there is no result set for the table. But if table A has no additional data, thread 2 returns to table B, where it has a cached result set.

Since thread 2 has a cached result set on table B, it creates a new batch from the result set instead of running a SQL query. Since this is the second batch from the result set, thread 2 closes the result set after processing the batch. This allows either thread to process data from table B the next time they check the table.

When the tables being processed use both table and partition processing, the threads query the partitions as described below. For details on how the tables and partitions rotate through the processing queue, see Processing queue.

Multithreaded partition processing

Multithreaded partition processing is similar to multithreaded table processing. The behavior around caching the result set and the number of batches to process from the result set is the same, but at a partition level. Threads also skip processing partitions that already have a result set cached from a different thread.

The difference is, when a thread works on a partition, all partitions from the same table are moved to the end of the processing queue. This allows the source to switch to the next available table.

For examples of how tables and partitions rotate through the processing queue, see Processing queue.

The source performs only multithreaded table processing in either of the following situations:

• Multithreaded Partition Processing Mode property is set to Off.
• Multithreaded Partition Processing Mode property is set to On (Best Effort) and no tables meet the partition processing requirements.

Say you have tables A, B, C and D that you configure for table processing. When you start the flow, the source adds all tables to the queue. If configured, the Initial Table Order Strategy advanced property can affect the order. Let's assume we have no referential constraints and use alphabetical order:

A  B  C  D

When a thread becomes available, it processes data from the first table in the queue. The processing of the tables depends on how you define the Per Batch Strategy property:

Process All Available Rows in the Table

With this batch strategy, threads process all data in a table before proceeding to the next table.

For example, say you have two threads. Thread 1 processes table A, thread 2 processes table B. That leaves the remaining tables in the queue, as follows:

C  D

The first available thread then processes table C, and the next processes table D.

If additional data is written to one of the processed tables, the table is placed at the end the queue.

Switch Tables

With this batch strategy, each thread processes a single batch from a table before moving to the next available table, so the threads cycle through the tables quickly.

After a thread performs a SQL query, it creates a batch and caches a result set for future use. As long as a thread has a result set cached for a table, other threads skip that table. The result set is closed based on the Batches from Result Set property. So, if the property is set to 2, then the thread must cycle back and create a second batch for the table before closing the result set. Closing the result set enables any thread to process the table.

Say you have two threads, one processes table A, the other processes table B. They each create one batch and cache a result set. Then, those tables move to the back of the queue, as follows:

C  D  A  B

The threads repeat the same process for table C and D, and those tables move to the back of the queue:

A  B  C  D  

So, after four batches are processed, the queue looks like it did in the beginning.

When cycling back to table A and B, each thread returns to the original table that they processed, creating a batch from its cached result set. That is, if thread 1 cached a result set on table A and thread 2 cached a result set on table B, they both go back to those tables to process a second batch.

After each thread processes a second batch from their result sets, they close the result sets. Then, the next time the tables appear at the front of the queue, either thread can process them.

The source performs only multithreaded partition processing in either of the following situations:

• Multithreaded Partition Processing Mode property is set to On (Required).
• Multithreaded Partition Processing Mode property is set to On (Best Effort) and all tables meet the partition processing requirements.

A partition remains in the queue until the source confirms that there is no more data in the partition. When a thread becomes available, it creates a batch from the first available partition of the first table in the queue. The order of tables and partitions in the queue depends on how you define the Per Batch Strategy, as follows:

Process All Available Rows in the Table

As when processing tables, when processing partitions with this batch strategy, threads process all data in a partition before proceeding to the next partition.

Note: In practice, this means that rows from subsequent tables can be processed before a previous table is completed, since available threads continue to pick up partitions from the queue.

For example, say we configure the flow for 4 threads. That means table C can have up to 8 partitions in the queue at any given time. So the initial queue looks like this:

A1  A2  A3  B1  B2  B3  C1  C2  C3  C4  C5  C6  C7  C8 

The four threads start processing on the first four partitions in the queue: A1, A2, A3, and B1. This puts B2 at the front of the queue, ready for the next available thread.

B2  B3  C1  C2  C3  C4  C5  C6  C7  C8

Note that in this situation, the processing of table B data begins before table A is fully processed.

After the four threads process another four sets of batches, the queue looks like this:

C3  C4  C5  C6  C7  C8

If additional data is written to the processed partitions, those partitions are placed at the end of the queue. So if data is written to A3, the queue looks like this:

C3  C4  C5  C6  C7  C8  A3

Switch Tables

As when processing tables, when processing partitions with this batch strategy, each thread processes a single batch from a partition before moving to the next available partition, so the threads cycle through the partitions quickly.

Also like processing tables, when a thread performs a SQL query to create a batch, it also caches a result set for future use. As long as a thread has a result set cached for a partition, other threads skip that partition. The result set is closed based on the Batches from Result Set property. So, if the property is set to 2, then the thread must cycle back and create a second batch for the partition before closing the result set. Closing the result set enables any thread to access the partition.

Unique to partition processing, this batch strategy forces all subsequent, consecutive partitions from the same table to the end of the queue each time a thread processes a batch from a partition.

Let's start again with the initial batch order, but this time let's say we have three threads, so table C is limited to 6 partitions:

A1  A2  A3  B1  B2  B3  C1  C2  C3  C4  C5  C6

When the flow starts, thread 1 takes A1, processing one batch and caching a result set. To switch tables, the rest of the table A partitions move to the end of the queue, and A1 takes the last position because it still has additional data to be processed:

B1  B2  B3  C1  C2  C3  C4  C5  C6  A2  A3  A1 

Similarly, thread 2 takes B1, processing one batch and caching a result set. The rest of the table B partitions move to the end of the queue, with B1 at the very end:

C1  C2  C3  C4  C5  C6  A2  A3  A1  B2  B3  B1

Thread 3 takes C1, processing one batch and caching the rest of the data in a result set. The rest of the table C partitions move to the end of the queue, with C1 at the very end:

A2  A3  A1  B2  B3  B1  C2  C3  C4  C5  C6  C1

Available threads continue taking the first partition in the queue as follows:

B2  B3  B1  C2  C3  C4  C5  C6  C1  A3  A1  A2
C2  C3  C4  C5  C6  C1  A3  A1  A2  B3  B1  B2
A3  A1  A2  B3  B1  B2  C3  C4  C5  C6  C1  C2
B3  B1  B2  C3  C4  C5  C6  C1  C2  A1  A2  A3
C3  C4  C5  C6  C1  C2  A1  A2  A3  B1  B2  B3
A1  A2  A3  B1  B2  B3  C4  C5  C6  C1  C2  C3

At this point, all of the A and B partitions, and partitions C1, C2, and C3, have one batch of data processed and cached result sets. The remaining C partitions have yet to be touched.

A1 was processed by thread 1. Say, A2 and A3 were processed by thread 3, and thread 2 is now available. Which partition does thread 2 take?

Thread 2 skips all of the A partitions because they have existing cached result sets from other threads. Instead, thread 2 takes B1, which has a result set that it created. Then, the other B partitions move to the end with B1 following. The A partitions remain at the front of the queue:

A1  A2  A3  C4  C5  C6  C7  C8  C1  C2  C3  B2  B3  B1

After processing this second batch from the B1 result set, thread 2 closes the result set. This allows any thread to process data from B1 the next time it comes to the front of the queue.

If thread 3 becomes available next, it also skips A1 and takes A2, which has a result set that it created. This results in the following order:

C4  C5  C6  C7  C8  C1  C2  C3  B2  B3  B1  A1  A3  A2

After processing this second batch from the A2 result set, thread 3 closes the result set. This allows any thread to process data from A2 the next time it comes to the front of the queue.

The source performs both multithreaded partition processing and multithreaded table processing in the following situation:

• Multithreaded Partition Processing Mode property is set to On (Best Effort) and some tables meet the partition processing requirements while other tables do not.

Say we have table A being processed without partitions, and table B configured with a maximum of 3 partitions, and table C with no limit. The flow has 3 threads to work with which grants 6 partitions to table C. Using the alphabetical initial table ordering, the initial queue looks like this:

A  B1  B2  B3  C1  C2  C3  C4  C5  C6

When a thread becomes available, it processes the first table or partition in the queue. The order of the queue depends on how you define the Per Batch Strategy, as follows:

Process All Available Rows in the Table

When processing tables and partitions with this batch strategy, threads process all data in a table or partition before proceeding to the next partition. An unpartitioned table, like table A, is processed like a table with a single partition.

When the flow starts, the 3 threads process a set of batches from table A and from partitions B1 and B2, leaving the queue like this:

B3  C1  C2  C3  C4  C5  C6

As each thread completes processing, it picks up the next table or partition at the front of the queue. After each of the 3 threads takes another table or partition, the queue looks like this:

C3  C4  C5  C6  

If additional data is written to one of the processed tables or partitions, the table or partition is placed at the end of the queue.

Switch Tables

When processing tables and partitions with this batch strategy, each thread processes a single batch from a table or partition before moving to the next available table or partition, so the threads cycle through the queue quickly.

When a thread performs a SQL query to create a batch, it also caches a result set for future use. As long as a thread has a result set cached for a table or partition, other threads skip that table or partition. The result set is closed based on the Batches from Result Set property. So, if the property is set to 2, then the thread must cycle back and create a second batch for the table or partition before closing the result set. This enables any thread to access the table or partition.

When processing tables and partitions, this batch strategy forces all subsequent, consecutive partitions from the same table to the end of the queue. It also treats unpartitioned tables as a single partition. As a result, the queue rotation is a simplified version of processing only partitioned tables.

So we have this initial order:

A  B1  B2  B3  C1  C2  C3  C4  C5  C6

Thread 1 processes a batch from table A and caches a result set. Then, table A moves to the end of the queue:

B1  B2  B3  C1  C2  C3  C4  C5  C6  A

Thread 2 processes batch from B1 and caches a result set. The rest of the table B partitions move to the end of the queue, with B1 at the end because it contains more data to be processed:

C1  C2  C3  C4  C5  C6  A  B2  B3  B1

Thread 3 processes a batch from C1 and caches a result set. The rest of the table C partitions move to the end of the queue, with C1 at the end because it contains more data to be processed:

A  B2  B3  B1  C2  C3  C4  C5  C6  C7  C8  C1

Now, say thread 2 is the first thread to become available. Though table A is at the front of the queue, table A also has a result set cached by thread 1. Only thread 1 can process table A while the result set remains open, so thread 2 skips table A and takes B2.

This leaves A in the front of the queue. The B partitions get pushed to the end of the queue, with B2 at the very end, as follows:

A C2  C3  C4  C5  C6  C7  C8  C1  B3  B1  B2

Thread 1 then becomes available. Since it has a cached result set on table A, it creates a batch from the result set and moves A to the end of the queue:

B2  B3  B1  C2  C3  C4  C5  C6  C7  C8  C1  A

After processing this second batch from the table A result set, thread 1 closes the result set. This allows any thread to process data from table A the next time it comes to the front of the queue.

The Oracle Multitable Consumer source specifies the operation type for generated records in the following record header attributes:

sdc.operation.type

The source evaluates the operation type associated with each entry that it processes. When appropriate, it writes the operation type to the sdc.operation.type record header attribute.

The source uses the following values in the attribute to represent the operation type:

• 1 for INSERT
• 2 for DELETE
• 3 for UPDATE
• 4 for MERGE

If you use a CRUD-enabled target in the flow such as JDBC Producer or Elasticsearch, the target can use the operation type when writing to target systems. When necessary, you can use an Expression Evaluator processor or any scripting processor to manipulate the value in the header attribute. For an overview of Data Collector changed data processing and a list of CRUD-enabled targets, see Processing changed data.

When using CRUD-enabled targets, the target looks for the operation type in this attribute before checking the following attribute.

oracle.cdc.operation

The source also writes the CRUD operation type to an oracle.cdc.operation attribute. This attribute was implemented in an earlier release and is supported for backward compatibility.

The source uses the following values in the attribute to represent the operation type:

• INSERT
• UPDATE
• DELETE
• MERGE

CRUD-enabled targets check this attribute for the operation type only if the sdc.operation.type attribute is not set.

The source can provide the following JDBC header attributes:

The following data type conversions do not include all information in the corresponding Data Collector data type:

• Decimal and Numeric data types are converted to the Data Collector Decimal data type, which does not store scale and precision.
• The Timestamp data type is converted to the Data Collector Datetime data type, which does not store nanoseconds.

To preserve this information during data type conversion, the source generates the following field attributes for these Data Collector data types:

Event records generated by Oracle Multitable Consumer source have the following event-related record header attributes: