Adds new WINDOW JOIN syntax for time-based aggregations:

SELECT t.*, avg(p.bid) avg_big, avg(p.ask) avg_ask
FROM trades t
WINDOW JOIN prices p ON p.sym = t.sym
  RANGE BETWEEN 1 second PRECEDING and 1 second FOLLOWING INCLUDE PREVAILING;

Here, each trades row is joined with all prices rows with the matching sym value from the [-1s, 1s] interval, inclusive, and then aggregations are calculated for the joined prices rows.

In case of symbol-based join, the execution may use SIMD instructions to calculate the aggregates. Other than that, the execution may use multiple threads as long as left-hand table has many rows.

Both INCLUDE PREVAILING (the default) and EXCLUDE PREVAILING clauses of WINDOW JOIN is supported. In case of the EXCLUDE PREVAILING clause, only the right-hand table rows from the window interval are aggregated. The INCLUDE PREVAILING clause also includes the latest right-hand table row (think, price) before the interval (think, ASOF JOIN combined with EXCLUDE PREVAILING WINDOW JOIN).

Requirements:

• Right-hand side of the query must be a "physical" table, i.e. not a sub-query.
• WINDOW JOINs can be chained in the same query level, but can't be mixed with other JOIN types, e.g. OUTER JOIN. Although, it's possible to use other JOIN types if WINDOW JOIN is moved to a sub-query.

GA Readiness Checklist

• Replace hardcoded 1024 symbol limit with dynamic hash tables to support arbitrary symbol value sizes @kafka1991
• Support more vectorized aggregate functions for AsyncFastWindowJoinRecordCursorFactory @RaphDal
• Support all fixed-size types (that have aggregate functions) @RaphDal
• Single-thread WindowJoinFactory @RaphDal
• FastWindowJoin support join filter @kafka1991
• NormalWindowJoin supports vectorized aggregate when joinFilter is null @kafka1991
• AsyncWindowJoin support case when slave doesn't support TimeFrameCursor @RaphDal
• Small page frame size for window join @kafka1991
• Apply time intrinsics from the left-hand table to the right-hand table @kafka1991
• remove TODO comments throughout the codebase.@kafka1991
• Support limit in master table @RaphDal
• Fix memory leak in SqlCodeGenerator around window join factory creation: when the code throws SqlException, we need to make sure that filter and groupByFunctions (and any other intermediate structs) are closed
• Double check that vectorized reducers kick in when all aggregate functions use right-hand columns only and cover left-hand/right-hand column expression case with tests
• Support window join chains @kafka1991
• Add single-threaded Fast factory - @RaphDal
• Add parallel tests - @puzpuzpuz
• Add parallel tests that involve functions which use group by allocator - @puzpuzpuz
• Fix missing casts in column sink @kafka1991
• Cover binary search in TimeFrameHelper with tests
• Add tests with multiple window joins involving the same left-hand table
• Add parallel test for vectorized impl that satisfies the perWorkerGroupByFunctions != null condition
• Add more tests, specially for edge case validation.
• Measure performance with the latest changes - the vectorized perf should be same as it was in PoC
• Time frame support for single threaded implementation @RaphDal
• [Nice to have] build segment tree aggregate prototype
• [Nice to have] AsyncWindowJoin support negative limit

Benchmarks

0.5B trades and 1.6B prices rows within a single day:

-- 123s on Ryzen 7900x + 64GB RAM box
SELECT t.*, avg(p.bid) avg_big, avg(p.ask) avg_ask
FROM trades t
WINDOW JOIN prices p ON p.sym = t.symbol
  RANGE BETWEEN '1' second PRECEDING and '1' second FOLLOWING INCLUDE PREVAILING
LIMIT -1;

-- 115s
SELECT t.*, avg(p.bid) avg_big, avg(p.ask) avg_ask
FROM trades t
WINDOW JOIN prices p ON p.sym = t.symbol
  RANGE BETWEEN '1' second PRECEDING and '1' second FOLLOWING EXCLUDE PREVAILING
LIMIT -1;

Microbenchmark results for the new DirectIntIntHashMap hash table used in the symbol column equality operator:

Benchmark                                  (size)  Mode  Cnt   Score    Error  Units
IntIntHashMapBenchmark.testDirectHashMap     5000  avgt    3   4.953 ±  4.042  ns/op
IntIntHashMapBenchmark.testDirectHashMap    50000  avgt    3   4.783 ±  0.197  ns/op
IntIntHashMapBenchmark.testDirectHashMap   500000  avgt    3   4.683 ±  0.135  ns/op
IntIntHashMapBenchmark.testDirectHashMap  5000000  avgt    3  20.119 ±  3.876  ns/op
IntIntHashMapBenchmark.testHashMap           5000  avgt    3   4.682 ± 10.045  ns/op
IntIntHashMapBenchmark.testHashMap          50000  avgt    3   5.380 ±  0.804  ns/op
IntIntHashMapBenchmark.testHashMap         500000  avgt    3   5.165 ±  0.623  ns/op
IntIntHashMapBenchmark.testHashMap        5000000  avgt    3  16.193 ±  6.691  ns/op
IntIntHashMapBenchmark.testStdHashMap        5000  avgt    3   9.468 ±  0.282  ns/op
IntIntHashMapBenchmark.testStdHashMap       50000  avgt    3  11.244 ±  1.683  ns/op
IntIntHashMapBenchmark.testStdHashMap      500000  avgt    3  30.416 ± 32.222  ns/op
IntIntHashMapBenchmark.testStdHashMap     5000000  avgt    3  82.271 ± 13.793  ns/op

New EqSymFunctionFactory speeds up filters like sym_col1 = sym_col2:

create table x as (
  select rnd_symbol('1','3','5',null) a, rnd_symbol('1','4','5',null) b
  from long_sequence(100_000_000)
);

-- master 1.24s
-- patch 65ms
select count() from x where a = b;

Microbenchmark results for Vect#sum*Acc() functions (they're now vectorized):

# before
Benchmark                                    Mode  Cnt     Score   Error  Units
SumDoubleBenchmark.testVectSumDouble         avgt    2   165.845          ns/op
SumDoubleBenchmark.testVectSumDoubleAcc      avgt    2  1122.349          ns/op

# after
Benchmark                                Mode  Cnt     Score   Error  Units
SumDoubleBenchmark.testVectSumDoubleAcc  avgt    2   151.589          ns/op

# before
Benchmark                            Mode  Cnt     Score   Error  Units
SumLongBenchmark.testVectSumLong     avgt    2   125.216          ns/op
SumLongBenchmark.testVectSumLongAcc  avgt    2  1128.825          ns/op

# after
Benchmark                            Mode  Cnt    Score   Error  Units
SumLongBenchmark.testVectSumLongAcc  avgt    2  159.327          ns/op

# before
Benchmark                          Mode  Cnt    Score   Error  Units
SumIntBenchmark.testVectSumInt     avgt    2   89.560          ns/op
SumIntBenchmark.testVectSumIntAcc  avgt    2  558.675          ns/op

# after
Benchmark                          Mode  Cnt    Score   Error  Units
SumIntBenchmark.testVectSumIntAcc  avgt    2  122.994          ns/op

# before
Benchmark                            Mode  Cnt     Score   Error  Units
SumShortBenchmark.testJavaHeapSum    avgt    2  4861.724          ns/op
SumShortBenchmark.testJavaNativeSum  avgt    2  4852.106          ns/op
SumShortBenchmark.testNativeSum      avgt    2   300.381          ns/op

# after
Benchmark                        Mode  Cnt    Score   Error  Units
SumShortBenchmark.testNativeSum  avgt    2   97.380          ns/op