Sort position links for faster non-equi joins

[pnowojski]

This is a implementation of:

#6922

From preliminary benchmarks this change didn't introduce regression in equality joins and for some in-equality joins this gives orders of magnitude speed ups.

BenchmarkInequalityJoin speeds up from ~210ms down to ~50ms and existing BenchmarkHashBuildAndJoinOperators do not show a visible performance regression:

AFTER

Benchmark                                              (hashColumns)  (matchRate)  Mode  Cnt    Score    Error  Units
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash        varchar          N/A  avgt   10   56.792 ±  7.107  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash         bigint          N/A  avgt   10   35.530 ±  2.304  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash            all          N/A  avgt   10   69.249 ±  7.824  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar          0.1  avgt   10  100.534 ±  8.280  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar            1  avgt   10  127.782 ±  5.132  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar            2  avgt   10  118.533 ± 13.199  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint          0.1  avgt   10   79.058 ±  4.773  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint            1  avgt   10  100.670 ± 10.200  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint            2  avgt   10   74.568 ±  2.282  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all          0.1  avgt   10  119.906 ±  1.313  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all            1  avgt   10  156.546 ± 12.703  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all            2  avgt   10  138.223 ±  7.201  ms/op

BEFORE:

Benchmark                                              (hashColumns)  (matchRate)  Mode  Cnt    Score    Error  Units
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash        varchar          N/A  avgt   10   54.626 ±  5.888  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash         bigint          N/A  avgt   10   35.253 ±  1.619  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkBuildHash            all          N/A  avgt   10   73.002 ±  8.299  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar          0.1  avgt   10   93.097 ±  4.915  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar            1  avgt   10  130.529 ±  7.421  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash         varchar            2  avgt   10  114.086 ± 10.210  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint          0.1  avgt   10   71.442 ±  4.633  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint            1  avgt   10   92.942 ± 12.446  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash          bigint            2  avgt   10   77.394 ±  2.632  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all          0.1  avgt   10  111.427 ±  3.545  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all            1  avgt   10  150.452 ± 15.728  ms/op
BenchmarkHashBuildAndJoinOperators.benchmarkJoinHash             all            2  avgt   10  130.621 ±  3.306  ms/op

There are couple of TODOs (future work?):

• compare method may not fit to PagesHashStrategy
• improve decision between choosing SortedPositionLinks over PositionLinksList
• maybe add NoopPositionLinks (that returns always -1) if all rows in build table have unique key (this would reduce memory usage and it might slightly improve performance)
• add similar logic to cross join

[losipiuk]

Why do you need this. It incomplete and unused.

[pnowojski]

Why is it incomplete? It is used both in SortedPositionLinks and in LookupSource directly, in case if there is no SortChannel for filter function.

[losipiuk]

Add short javadoc what it is.

[pnowojski]

please check comment if it suffice

[losipiuk]

Either change signature to take "JoinFilterFunction" or allow empty() and set null to the field in that case.

[pnowojski]

changed method signature

[losipiuk]

don't do that. Extract separate isDescending(ComparisonType) method.

[pnowojski]

fixed

[losipiuk]

drop those and just add default: section with return Optional.empty()

[pnowojski]

fixed

[losipiuk]

rename to asBuildFieldReference

[pnowojski]

done

[losipiuk]

Shouldn't you check some conditions about other side which is not buildFieldReference.
It seems like you are making some non-written assumption on shape of expression which is passed to this method.
Those should be exposed as Preconditions.checkArgument calls.

[pnowojski]

yes, I have missed one thing. For such simple extractor, I have to check that there is no build field references on the other side of the inequality. Fixed

[losipiuk]

why this change?

[kokosing]

+1 for undo

[pnowojski]

unintended change. Reverted

[losipiuk]

Please add as a reference implementation.

[pnowojski]

This also should be marked as //TODO

[kokosing]

you were able to write compiler code, so it should be easy to write regular code, shouldn't it?

[pnowojski]

At a time it wasn't crucial for performance tests and validation whether whole idea has any sense. I have added it now though.

[losipiuk]

This cannot be part of state as class is shared between threads.

[pnowojski]

Yes, that's one of the reasons why I think decoupling PositionLinks from JoinHash is a way to go. I should have marked this as a //TODO

[kokosing]

initial review

[kokosing]

You can use multimap. Does it influence that you are using boxed types?

[pnowojski]

I can not use Multimap, because I need to have last linked element as a starting point for the position links. Thus on each linking I would have to copy all collection returned by Multimap.get(...)

[kokosing]

You can use computeIfAbsent - https://docs.oracle.com/javase/8/docs/api/java/util/Map.html#computeIfAbsent-K-java.util.function.Function-

[pnowojski]

done

[kokosing]

Generally, I think these tests are not enough. It is similar with the current problem related to Exchanges.
The query will pass regardless you other commits are merged or not.

I am sure there are already some tests form inequality join.

[pnowojski]

Tests are very WIP.

Btw, there doesn't seems to be tests, that cover those test cases with simple enough comparison expression :)

[kokosing]

I think you can easily add unit tests for that class. That would be nice because it would show expressions are supported.

[pnowojski]

added

[kokosing]

+1 for undo

[kokosing]

why empty? I think it would be nice to test the things here.

[pnowojski]

I didn't want to spend a lot of time adding tests without initial approval that this thing will go into code base.

[losipiuk]

Is that still valid? I mean there is approval on approach. Right?

[pnowojski]

I think is not worth testing, since writing such test would be quite painful.

There are unit tests for both SortedPositionLinks and SortChannelExtractor and integration between JoinHash and PositionLinks interface is tested here regardless of this Optional.empty(). Also there are integration tests in AbstractTestQueries for this .

[kokosing]

you were able to write compiler code, so it should be easy to write regular code, shouldn't it?

[kokosing]

extract this as separate class

[pnowojski]

done

[pnowojski]

@dain I have fixed concurrency issues without refactor (as discussed) and it's read for review.

[dain]

@losipiuk and @kokosing are you two done with your reviews?

[losipiuk]

@dain I will give it one more read but feel free to go through it at the same time.

[losipiuk]

Looks ok up to my understanding of the code.
I have some outstanding questions to @pnowojski to be asked offline.
Though, It should not block your review @dain - at least to extent to see if general approach is something we want to push forward or not.

[losipiuk]

Maybe extract 1 variable (e.g MAX_MEMORY_USED). And set the pools as fractions of this one. So if we need to change sth here we just change one thing. Also set max-data-per-node config prop based on that.

[pnowojski]

I would rather expect that in the future we will need more customization of those parameters. In other words, we will need passing those values as parameters. Binding them to one static value doesn't seem like a good idea, since they are quite independent.

[losipiuk]

don't we need to save the results of query into some field to be sure that JVM does not need some weird code-removal optimizations?

[pnowojski]

I have added collecting output into List<Page> to MemoryQueryRunner

[losipiuk]

Why not return just new head instead boolean (either left or right)

[losipiuk]

How end of list is signalled?

[losipiuk]

Commit message suggests that you use Javas LinkedList as an implementation.
Maybe use Implement PositionLinks as link array as commit message and name the class PostionLinksArray?

[losipiuk]

Or maybe LinkedPositionLinks will be good and match the SortedPositionLinks from latter commit?

[pnowojski]

I have renamed it to ArrayPositionLinks, since it matches with Java's convention of ArrayList vs List.

[losipiuk]

InelliJ says Can be replaced with Comparator.comparingLong

[pnowojski]

yes, but I don't like it. This seems more easier to understand for me and it matches to reversed comparator in test below.

[losipiuk]

Name this testJoinWithLessThanInJoinClause

Inequality is everything which is not equailty. E.g. LIKE, complex expressions etc.
At least it used that way in test name convention so far.

[losipiuk]

ditto

[losipiuk]

ditto

[losipiuk]

Is that still valid? I mean there is approval on approach. Right?

[dain]

Review status:

• a87c482 Bump memory limits for memory benchmarks
• 5fd096a Add BenchmarkInequalityJoin
• edc1c75 Add benchmark for using position links in join
• 6b538a7 Add PositionLinks interface
• ca87b79 Implement ArrayPositionLinks
• b2e85b6 Implemented SortedPositionLinks with binary search
• 3d33e0b Add unit tests for PositionLinks
• 2ee7111 Add SortExpressionExtractor
• 3f9d0df Add unit tests for SortExpressionExtractor
• 87b119a Add tests queries for sorting position links
• 29b2f67 Extract sort channel from filter function
• f20ac3a Add compare method to PagesHashStrategy
• 4486d5e Use SortedPositionLinks in LookupJoinOperator
• 3533940 Collect output of the memory benchmarks

[dain]

Nice. I had a bunch of minor comments, but it is generally looking good. I noted one major concern about the stateful filter functions, but I think this will be easy to remedy.

This PR was pretty difficult to read because the features were broken into many small commits containing a single file. The is especially the case when a new class is added in one commit and then the test is in the next. It helps to read the code quickly if the commits grouped into features, so I can see how the different bits relate to each other and see usages in code and tests. For this review, I had to flip back and forth between commits. The collection into features (maybe mini-features) helps be focus on a single context which makes reading faster. Here is how I would have grouped the commits, and how I think you should squash them before we merge them:

Fix benchmarks using MemoryLocalQueryRunner

• 3533940 Collect output of the memory benchmarks
• Also apply the "output" fix to the other benchmarks using MemoryLocalQueryRunner

Increase memory limits for benchmarks using MemoryLocalQueryRunner

• a87c482 Bump memory limits for memory benchmarks

Add inequality join benchmarks

• 5fd096a Add BenchmarkInequalityJoin
• edc1c75 Add benchmark for using position links in join

Add direct handling of inequality conditions in JoinHash

• 6b538a7 Add PositionLinks interface
• ca87b79 Implement ArrayPositionLinks
• b2e85b6 Implemented SortedPositionLinks with binary search
• 3d33e0b Add unit tests for PositionLinks
• 2ee7111 Add SortExpressionExtractor
• 3f9d0df Add unit tests for SortExpressionExtractor
• 87b119a Add tests queries for sorting position links
• 29b2f67 Extract sort channel from filter function
• f20ac3a Add compare method to PagesHashStrategy
• 4486d5e Use SortedPositionLinks in LookupJoinOperator

[dain]

Maybe add an implementation of OutputFactory an inner class, so you don't have to have this crazy 5 argument lambda :) ?

[dain]

I think you might be able to use PageConsumerOutputFactory instead of this.

[dain]

I still "you might be able to use PageConsumerOutputFactory instead of this".

[dain]

I'm am concerned about this change. The reason we have a FilterFunctionFactory is because filter functions are stateful and not thread safe. Each driver (and thus threads) calls get to fetch a private version of the JoinHash for their thread. I am surprised there are no join tests that use a stateful function in the join condition (or maybe the tests are not concurrent enough to see the problem).

[pnowojski]

Grrr... :( This makes it more complicated. I can not build positionLinks in this get method, because it would be too costly to sort position links every time get is called.

But on the other hand, I can not return positionLinks created in the constructor with it's private final filterFunction here, because then we hit the same problem you are describing.

The only solution that I can see is to build/sort positionLinks in the constructor with temporarily created filterFunction, but before returning it in in get, this filter function must be overwritten by newly created one.

[dain]

Why not use IntComparator from fastutils?

[dain]

I think we normally put this call in the @Benchmark method and have the context just expose a getter for the runner.

[dain]

It might be easier to read if you named the tables a and b or if you dropped the aliases and rewrite the query to:

SELECT count(*) FROM t1 JOIN t2 using (bucket) WHERE t1.val1 < t2.val2

[dain]

most of these fields are no longer used

[pnowojski]

after fixing thread safety of FilterFunction they are now used again

[dain]

this can be final

[dain]

This is a utility class (all static methods), so it should be final

[dain]

How about extractSortOrder?

[pnowojski]

actually it should be extractSortExpression. SortOrder was a remanent after a refactor/rename.

[dain]

Can we add a kill switch for this feature incase we find a bug?

[pnowojski]

I have applied changes, please check my responses to some of your comments.

I have also merged some of those commits, but IMO it's much harder to review one 1000 commit from 1200 lines PR. Indeed, smaller commits are often missing some greater context, but in such case if I have doubts I like to fallback to diff of all commits by whole PR to grasp what PR is about or to search for method usages.

On the other hand smaller commits allows for bundling some changes that belong to some topic. Like in this PR commit "Add compare method to PagesHashStrategy". It would just make unnecessary noise if squashed with other commits.

[pnowojski]

fixed, Intellij was auto folding this so I missed it

[pnowojski]

I have moved filterFunction to SortedPositionLinks constructor and dropped it from ArrayPositionLinks.

[pnowojski]

Thanks, indeed much better to understand.

[pnowojski]

I have replaced it and didn't notice performance improvement in benchmarks. However code is almost the same, so fastutil version can stay.

[pnowojski]

that's not the only place that uses this assumption. This lines uses it to terminate iteration over position links. Same assumption is used to find a starting point of iteration.

I have added javadoc comment at the top of this class.

[pnowojski]

I have added some explanation to the java doc, because I can not come up with an idea for better naming. Keep in mind that smaller and larger are defined using an external comparator, which is a transformation of the filterFunction

[pnowojski]

wow, nice bug... It must have slipped through since it was just a poc...

[pnowojski]

actually it should be extractSortExpression. SortOrder was a remanent after a refactor/rename.

[pnowojski]

They execute faster in H2 then in presto.

[pnowojski]

Grrr... :( This makes it more complicated. I can not build positionLinks in this get method, because it would be too costly to sort position links every time get is called.

But on the other hand, I can not return positionLinks created in the constructor with it's private final filterFunction here, because then we hit the same problem you are describing.

The only solution that I can see is to build/sort positionLinks in the constructor with temporarily created filterFunction, but before returning it in in get, this filter function must be overwritten by newly created one.

[kokosing]

@dain I have just approved. Thanks for notification ;)

[pnowojski]

Applied comments.

I have added some more benchmark cases to see if there is a performance drop for very short position links (larger number of buckets) and surprising I haven't seen that. So indeed I think we can turn this on by default.

Here are the results. Number of elements in build table is 60175, so position links length is equal to 60175 / buckets. Of course the larger position links chain, the better case for the binary search.

Benchmark                              (buckets)  (fastInequalityJoin)  (filterOutCoefficient)  Mode  Cnt     Score    Error  Units
BenchmarkInequalityJoin.benchmarkJoin        100                   true                      10  avgt   30    79.657 ±  4.545  ms/op
BenchmarkInequalityJoin.benchmarkJoin        100                  false                      10  avgt   30  1367.266 ± 28.193  ms/op
BenchmarkInequalityJoin.benchmarkJoin       1000                   true                      10  avgt   30    70.507 ±  4.654  ms/op
BenchmarkInequalityJoin.benchmarkJoin       1000                  false                      10  avgt   30   204.847 ±  6.114  ms/op
BenchmarkInequalityJoin.benchmarkJoin      10000                   true                      10  avgt   30    65.660 ±  4.196  ms/op
BenchmarkInequalityJoin.benchmarkJoin      10000                  false                      10  avgt   30    82.278 ±  3.722  ms/op
BenchmarkInequalityJoin.benchmarkJoin      60000                   true                      10  avgt   30    61.385 ±  3.970  ms/op
BenchmarkInequalityJoin.benchmarkJoin      60000                  false                      10  avgt   30    65.466 ±  5.425  ms/op

[dain]

The "performance drop" you are seeing is in the margin of error, so it doesn't mean anything.

[pnowojski]

Maybe I wasn't clear enough. I meant, that I expected to see some performance regression in some edge cases, but it wasn't visible in benchmarks (which is good :) )

[dain]

Merged, thanks!

[pnowojski]

Thanks :)