SQL Server 2019 BDC Multitable Consumer (deprecated)

You can install the SQL Server 2019 Big Data Cluster stage library using any of the following methods:

• Install the library in an existing Data Collector as a custom stage library.
• If using Control Hub, install the library in a provisioned Data Collector container that is part of an orchestration framework, such as Kubernetes. Use a technique valid for your environment:
  • In a production environment, see the Control Hub topic Provision Data Collectors. You must install the SQL Server 2019 Big Data Cluster stage library in the customized StreamSets Data Collector Docker image.
  • In a development environment, you can run the StreamSets-developed deployment script to try SQL Server 2019 BDC with Data Collector through Control Hub.

    The script deploys a Control Hub Provisioning Agent and Data Collector on a Kubernetes cluster. The script automatically installs the SQL Server 2019 Big Data Cluster stage library in the deployed Data Collector. You can use that Data Collector as an authoring Data Collector to create and test SQL Server 2019 BDC pipelines.

    Use the script in development environments only. For more information, see the deployment script in Github.

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 origin reads all available data from each table when you start the pipeline. The origin generates SQL queries using the following syntax when you start the pipeline:

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

You can make the following changes to how the origin 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 pipeline restart. For details, see Multiple Offset Value Handling.

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 SQL Server 2019 BDC Multitable Consumer origin 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 origin 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 origin 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 origin 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 origin uses to determine where to start reading data. The origin 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 origin reads the table, multiple active transactions are being written to the table with the current timestamp for the transaction_time column. When the origin finishes reading the first record with the current timestamp, the origin 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 origin 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 SQL Serer 2019 BDC Multitable Consumer origin retrieves the list of tables defined in the table configuration when you start the pipeline. The origin 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 origin is processing full tables or partitions within a table:

Multithreaded table processing

When the origin 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 origin 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 origin.

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 origin 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 origin performs full table or partition processing. The number of batches that the origin creates before closing a result set is based on the Batches from Result Set property.

Multithreaded table processing

When the origin performs multithreaded table processing for all tables, each thread creates a set of batches from one table, and then switches to the next available table to create the next set of batches. The thread runs an initial SQL query to create the first set of batches from the table. The database caches the remaining rows in a result set in the database for the same thread to access again, and then the thread switches to the next available table. A table is available in the following situations:

• The table does not have an open result set cache. In this case, the thread runs an initial SQL query to create the first batch, caching the remaining rows in a result set in the database.
• The table has an open result set cache created by that same thread. In this case, the thread creates the batch from the result set cache in the database rather than running another SQL query.

A table is not available when the table has an open result set cache created by another thread. No other thread can read from that table until the result set is closed.

When you configure a switch table strategy, define the result set cache size and the number of batches that a thread can create from the result set. After a thread creates the configured number of batches, a different thread can read from the table.

Note: By default, the origin instructs the database to cache an unlimited number of result sets. A thread can create an unlimited number of batches from that result set.

For example, let's say an origin has a batch size of 100 and uses two concurrent threads to read from four tables, each of which contains 10,000 rows. You set the result set cache size to 500 and set the number of batches read from the result set to 5.

Thread1 runs an SQL query on table1, which returns all 10,000 rows. The thread creates a batch when it reads the first 100 rows. The next 400 rows are cached as a result set in the database. Since thread2 is similarly processing table2, thread1 switches to the next available table, table3, and repeats the same process. After creating a batch from table3, thread1 switches back to table1 and retrieves the next batch of rows from the result set that it previously cached in the database.

After thread1 creates five batches using the result set cache for table1. Thread1 then switches to the next available table. A different thread runs an SQL query to read additional rows from table1, beginning from the last saved offset.

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, with a twist - each thread creates a set of batches from one partition for a table, then all partitions from the same table are moved to the end of the processing queue. This allows the origin to switch to the next available table.

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.

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

The origin 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 pipeline, the origin 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 number of batches is based on the Batches from Result Set property. 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 do not start processing data in the next table until all available data is processed for the preceding table.

That is, table A remains at the front of the queue until all available data is processed. Then processing begins on table B. Table A moves to the back, remaining in the queue in case more data appears, as follows:

B  C  D  A  

Switch Tables

With this batch strategy, the order of the queue remains the same, but each thread performs a SQL query to create a set of batches based on the Batches from Result Set property. When it completes processing, it performs the same process with the next table in the queue.

After a thread takes a set of batches from table A, table A moves to the back of the queue:

B  C  D  A

The next thread takes a set of batches from table B. Then B moves to the back of the queue:

C  D  A  B  

So after processing 4 sets of batches, the queue looks like it did in the beginning:

A  B  C  D

The origin 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.

When you start the pipeline, each table is queued up with the maximum number of active partitions. And for table C, that means double the number of threads for the pipeline. So if we configure the pipeline for 4 threads, 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 

A partition remains in the queue until the origin confirms that there is no more data in the partition. When a thread becomes available, it creates a set of batches from the first partition of the first table in the queue. The number of batches is based on the Batches from Result Set property. 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

When processing partitions, this batch strategy retains the original order of the queue, but rotates through the partitions as each thread processes a set of batches.

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, 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. And since the four partitions being processed have additional data to process, they go to the back of the queue. So processing of table B data begins before table A is fully processed.

The rest of the partitions remain in the original order as follows:

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

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

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

Switch Tables

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

Let's start again with the initial batch order:

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

When a thread processes a set of batches from A1, it pushes the rest of the table A partitions to the end of the queue. This queues up the next table, table B, for processing. And since A1 still contains data, it takes the last spot, as follows:

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

As the second thread processes a set of batches from B1, the other B partitions are sent to the back, and since B1 still contains data, it takes the last spot as follows:

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

And as the third thread takes a set of batches from C1, the rest of the C partitions are pushed to the back, so the queue looks like this:

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

The origin 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. As in the example above, the pipeline has 4 threads to work with which allows 8 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  C7  C8 

When a thread becomes available, it processes a set of batches from the first table or partition in the queue. The number of batches is based on the Batches from Result Set property. The order of the queue depends on how you define the Per Batch Strategy, as follows:

Process All Available Rows in the Table

With this batch strategy, the queue remains in the basic initial order and rotates as each thread claims a set of batches from the next table or partition. The unpartitioned table A is processed like a table with a single partition.

Note that unpartitioned tables are not processed in full when they move to the front of the queue. For this behavior, configure all tables to be processed without partitions. Or, set the Batches from Result Set property to -1.

When the pipeline starts, the 4 threads process a set of batches from the A table and from partitions B1, B2, and B3. Since the table and partitions all still contain data, they then move to the end of the queue as follows:

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

As each thread completes processing, it processes a set of batches from the front of the queue. After each of the 4 threads takes another set of batches, the queue looks like this:

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

Switch Tables

When processing tables and partitions, this batch strategy forces all subsequent, consecutive partitions from the same table to the end of the queue. And it treats unpartitioned tables as a table with 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  C7  C8 

The first thread processes a set of batches from table A, and since there are no related partitions, it simply goes to the end of the queue:

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

The second thread processes a set of batches from B1, pushes the rest of the table B partitions to the end of the queue, and B1 lands at the end because it contains more data to be processed:

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

The third thread processes a set of batches from C1, pushes the rest of the table C partitions to the end, and C1 takes the last slot:

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

And then the fourth thread processes another set of batches from table A, and moves A to the end of the queue:

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

The origin can provide the following JDBC header attributes:

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

• SQL Server Decimal and Numeric data types are converted to the Data Collector Decimal data type, which does not store scale and precision.
• SQL Server Datetime, Datetime2, and Smalldatetime data types are converted to the Data Collector Datetime data type, which does not store nanoseconds.

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

Event records generated by SQL Server 2019 BDC Multitable Consumer origin have the following event-related record header attributes: