In the process of the Lager brewing we use similar kind of transformations as we do in bulk processes. In majority of cases these queries are executed for few rows so should work based on other execution plans than these bulk operations. At the same time we run similar queries molded by requirements of identification with different leading data sets what impacts necessary plans too. And by the other hand we can always expect some source bulk when we cannot run these transformations row-based, but the optimizer has to work similarly to the bulk processing.
When I teach optimization of SQL queries, I classify type of tasks as either row based, bulk or hybrid ones.
Task of Lager-wise Real-time ODS is the Hybrid one and is the most complicated because of the requirements volatility.

Following edges should be considered:

• They are volatile - Every execution depends on volume of queued data (hops, IDs) and there could be various requirement for execution plans for each of them.
• They are plenty - Despite I have illustrated steps of the brewing as particular SQL queries, that are not few individual queries. There are thousands of queries running every second and the solution should be stable and scaleable. Any wrong plan could shake the fluency. That is why the solution should be well monitored and discrepancies tracked and solved immediately.
• They are multifaceted - Every transformation query is molded and used with different predicates i.e with different plans. In the row based execution we have to support much more indexes than in classical OLTP solutions to support all of them. We as well have to assure the predicate is always pushed to detail levels of the execution what could be tricky in cases of analytical function or aggregations.
• They are concurrent - We cannot lock any queue during our execution because we have to be fluent. We have to count on new items appearing in every queue during our execution of previous batch.

• Periodical business bulk operations in source systems - Despite of 24x7 characteristics of most source systems there are still periodical actions like Close of day, Close of month, Close of year, so operations triggered in the source system by a time period closure. They are usually well planned and the scenario to stop real-time processing during these actions (well, at 24x7 system that would be temporary breach of consistency of real-time ODS), but there are not always firm time frames and these organizational steps would require additional communication between source systems and Real-time ODS process management to stop and then to restart in time. So it is much better (I cannot say easier) to make your Real-time ETL processing able to absorb bulks organically without external process management actions.
• Periodical technical bulk operations in source systems - Differently from the previous category these actions are related to backup and clean-up periodical actions resulting in data changes. They have not always severe impact to data integration, but it is necessary to count on them.
• Unintended small bulks (bubbles) caused by data processing complexity - Data processing is not always smooth however we want to. Small regular data change can cause huge changes of dependent tables thanks to multiplication effect. Especially changes in lists of values what are relatively stable and almost immutable, changes in master data (e.g. repairs of wrong consolidation) etc.
• Reconciliations and repairs of source system after their failures or downtime - These things cannot be planned and are not considered as standard processing, but they happens. And we do not build our Real-time Operational data store for standard situations only, we build it for real life.
• Reconciliation and repairs after our system failure or downtime - Is the same as the one above. There are failures, breaches, issues in real life.

The problem is not that row based was not efficient with large load. It is the same way efficient with small and large amounts of record. The problem is in the part "the same". Because we need our system to work more efficient with large bulks.
The problem is not that row based is slow but that bulks are large. And that we need then processed much faster and more efficiently than particular rows.
Following picture shows the reason. The dissolving phase of row-based processing takes long time. All is proportional, but because of the size of the bulk too slow. We can afford bigger latency during the night Close of month out of business hours usually. But we want our system working properly in the morning.

From the graph above you can see why we use the SMALL - LARGE switch.

And now an important question:

How to decide what is the SMALL/LARGE rows count?

In some very serious resources you can read it is 42. I think it can differ statement by statement, by my experience the default could be set to 40000-100000 records on Oracle server but it is always good to allow designers to change the value for particular statements (pattern directive, hint) if necessary.

Following graph shows various paths of working with learned queue size and setup of statistics (spilling the number to optimizer). It shows both feasible and not feasible variants:

Second, whenever the execution gets into the volatile zone (it would be not often because it is off of the most frequent part of the small zone), randomize the execution plan (small/large) and collect supplemental log of method (plan), size of queue and execution time.
Third, after gathering enough statistics reconsider the particular limit SMALL-LARGE moving it either lower or higher along with the volatile area. After gathering enough sample of results (you have to consider some level of available system resources volatility too, so let the sample be huge enough) just compute the Linear regression for both sets.

TYPE B: - Small - Large approach
Following graph shows comparison of the row-based processing and the bulk processing efficiency. Bulk (set-based) processing is a total looser around small numbers of processed records. The difference is as significant that it is easy to seek and alert wrong plans, missing indexes, not pushed predicates etc. There is no doubt we need the fine grain row based processing 90 - 99% of time.
By the other hand bulk processing using full scan of tables and hash joins is much more proficient in big amounts of data. So the task is to be able to provide every statement either row or set based depending on coming amount of data.
And there is one big advantage we get during the Hops and IDs processing:
We know how many rows is in the queue! In each one!

So, how to reach that. How to make queries work based on the amount of coming data?
There is one quick and simple answer. Optimizer of the DB engine will do the work based on statistics. Like so many quick and simple answer it is not much real. I cannot speak bad about DB optimizer itself. In the case of our solutions it does unbelievable things, unbelievable pushing of predicates and other incredible things. No organic brewing described in last article would be possible without the great work of optimizer. But...

• Optimizer works based on statistics - but we cannot recalculate statistics before each run of an atomic statement and size of queues can volatile. We can of course rely on dynamic sampling, but it gets the database core into overhead too.
• Optimizer remembers last resolutions - Optimizer does not do hard parse and does not optimize again and again at repeating statements. So running the same statement several times in row automatically stuck on the first plan based on the first queue size.
• The previous point is reasonable - Causing hard parse before each atomic transaction would boil the batch.
• Two ranges is enough - you have to run each statement in two modes, the SMALL one and the LARGE one. There are several ways how to do that and they depends partially on features of the DB engine. So you can:
  • Use two distinguished statements instead of one each forced to another plan. Switch between these could be decided based on size of leading queue. There is no fear of redundancy as these statements are generated from the same metadata.
  • Set statistics on queue at the moment the size is switching between SMALL and LARGE and vice versa. The idea of setting realistic statistics based on before learned size of queue before each step fails on the fact setting of statistics causes hard parse of statement (when it should have any effect) an that would be an overhead at many atomic executions, so the number of switches should be minimized.
  • Use two queues, one for SMALL and one for LARGE - may be combined with difference of generated code for small and for large (different early pruning, LARGE statement has less limitations so sometimes it could be reasonable). This solution is more complicated, you have to solve the problem how and when to decide the queue will be the LARGE one, but it helps to dissolve bulks at the same time as new changes are processed.

Explanation:

• Row by row - That is the lazy way we decide to keep all the processing row based (as described in the chapter above). I classified that as not feasible way because of inability to dissolve bulks in time (in some extent it can make the system stuck), but it can work somehow and at least you have spared the effort of switching between small and large and having two execution plans and at least the most frequent SMALL phase work efficiently when it does not dissolve bulks of previous hard night.
• Realistic statistics - Another one I classified as non feasible. By one hand I trust optimizer (a bit) and it would work well with plans, but it increases number of hard-parses in the most frequent zone of SMALL operations. It causes huge performance issue when the system becomes big enough.
• White lie - My favorite solution. I trust the optimizer a bit, but not as much to let it decide in volatile zone around SMALL-LARGE boundary. This type of truth bending makes the SMALL-LARGE decision binary and counts on two plans, two ways. In the case you decide for two different statements for SMALL and LARGE executions (small overhead while generating, but much more possibilities of hard-core tuning), that is the best way. The advantage is we hold our finger on the execution process firmly, disadvantage the impact of wrongly set SMALL-LARGE limit.
• Partial white lie - This way is a compromise between the White lie and the Realistic statistics. It combines both advantage of trusted optimizer and reduction of hard parses in the most frequent SMALL zone. The advantage is less sensitivity of SMALL-LARGE limit set too low, disadvantage could be not perfect work of optimizer in the volatile zone.

• Assure possibility to setup overriding value for each transformation as a pattern directive or hint and to fill it in the manual tuning process when necessary.
• Let the system self-tune the limit based on statistic of previous runs.

Forth - After computing linear regression just find the intersection and push the SMALL-LARGE limit that way. Our defined volatile area will move too. Results of previous statistics should not be deleted. Shifting the SMALL-LARGE limit we changed nothing about transformations so our statistics data are still valid. Only reason for deleting statistics should be significant change of transformation logic or patterns used.

Note: There could appear a question why we do not use the capability of optimizer to solve the mess in the volatile area instead of such a linear regression circus. This specialized action is based on our knowledge of character and shape of our operations. The optimizer has more advanced resources, but we "natives" know the terrain of our land much better. That is why my pessimistic approach to tuning is usually so successful.

• Number of transformations (T) - means number of transformation modules, mappings ...
• Number of source tables (S) - number of event processing enabled source tables attending in transformations as sources (e.g. tables of the L0 layer, further or recursively used tables of L1 layer).
• Number of listening processes (L) - number of permanently listening processes.
• Number of processing threads (R) - number of threads the system allows to listen and run transformation and all necessary about our brewing.

R = L L = S + T

Thread groups - Transformations and source processes are collected into thread groups. The size and organization could be organized based on priority and SLA of particular transformations. All the signal processing is then aggregated to less number of groups. The solution is more complex but you can better scale the performance, diversify preferences and suppress unimportant parts of solution.

R = L L < S + T

Dispatchers - Listening and processing can be separated The solution brings some overhead about starting new processes if not organized in pools.

• Standard queue - We can use some still proven queue solutions, but aside the our needed they usually support many other roles, what makes them a bit heavy solution for our reason and we can hit on performance limits easily, because "they are plenty" is still valid.
• Bookmark - Bookmarking the batch start is not bad solution, but the problem is in the world of concurrent writing transactions in read-committed isolation you cannot relay neither on sequence nor on timestamp. Other isolation levels are out of thinking for the task.
• Flagging - We can set flags on records and decide the which records will be part of the micro-batch. Unfortunately the flag settings is a bit expensive operation for our reason.
• Statement isolation - We can use statement isolation - read-use-delete in one statement (maybe using some memory structure to help). That requires totally homologous environment.
• In-memory indexing - Less expensive method of flagging. It is relatively flexible (allowing to limit maximum bulk size).

• Bulk processing in the database - In our previous graphs we have presented bulk processes as stable in big numbers of records with and slow growing characteristics in big numbers. Bulk processing uses HASH JOIN method excessively and even there there are some limitations in huge numbers (we speak about 10'000'000 multiplications range).

• In-memory bulk index limits - there is some overhead in the case of huge bulks if you use this method.
• Fail-over - too large and long running bulks cause longer rollback time and repeat time when the process is interrupted.
• Process continuity - In the case of chained processes smaller bulks allow continuing process start earlier so the overall latency gets smaller.