Enable rewriting certain inner joins as filters.

[gianm]

The main logic for doing the rewrite is in JoinableFactoryWrapper's
segmentMapFn method. The requirements are:

• It must be an inner equi-join.
• The right-hand columns referenced by the condition must not contain any
  duplicate values. (If they did, the inner join would not be guaranteed
  to return at most one row for each left-hand-side row.)
• No columns from the right-hand side can be used by anything other than
  the join condition itself.

HashJoinSegmentStorageAdapter is also modified to pass through to
the base adapter (even allowing vectorization!) in the case where 100%
of join clauses could be rewritten as filters.

In support of this goal:

• Add Query getRequiredColumns() method to help us figure out whether
  the right-hand side of a join datasource is being used or not.
• Add JoinConditionAnalysis getRequiredColumns() method to help us
  figure out if the right-hand side of a join is being used by later
  join clauses acting on the same base.
• Add Joinable getNonNullColumnValuesIfAllUnique method to enable
  retrieving the set of values that will form the "in" filter.
• Add LookupExtractor canGetKeySet() and keySet() methods to support
  LookupJoinable in its efforts to implement the new Joinable method.
• Add enableRewriteJoinToFilter feature flag to
  JoinFilterRewriteConfig. The default is disabled.

Testing strategy:

• Add join-to-filter conversion tests to JoinableFactorWrapperTests.
• Add getRequiredColumns tests to individual query engines.
• Add getNonNullColumnValuesIfAllUnique tests to LookupJoinable and IndexedTableJoinable.
• Extend BaseCalciteQueryTest's QueryContextForJoinProvider to also
  provide query contexts that enable this rewrite.
• Add some new tests to CalciteQueryTest that are designed to exercise
  this rewrite. (And some existing tests did, too.)

[abhishekagarwal87]

Suggested change

	DEFAULT_ENABLE_JOIN_FILTER_REWRITE_VALUE_COLUMN_FILTERS
	DEFAULT_ENABLE_REWRITE_JOIN_TO_FILTER

[gianm]

😱

[abhishekagarwal87]

Looks good. Out of curiosity, given that we exclude null values, how are null values in the left table matched?

[abhishekagarwal87]

Suggested change

	* @throws UnsupportedOperationException if {@link #canIterate()} returns false.
	* @throws UnsupportedOperationException if {@link #canGetKeySet()} returns false.

[abhishekagarwal87]

do we need not exclude empty strings here?

[gianm]

Ah, I think we should. I replaced it with NullHandling.isNullOrEquivalent(s).

[abhishekagarwal87]

maybe not here but should we cap the max size so that it can't result in an OOM?

[gianm]

I was thinking we should rely on the user setting this parameter "correctly" sort of like the subquery limit. I also think most people won't change it from the default, which is 10,000 and should be pretty safe. Unless the values are gigantic it's only going to be a few MB per query.

I thought a bit about measuring these limits in terms of bytes instead of rows, which has pros/cons:

• Pro of bytes: less likely to be misconfigured & cause OOME, more likely to use memory efficiently & maximally
• Con of bytes: harder for users to understand the limit. "10,000 rows" is easy to communicate & understand; "5MB" is harder because people won't be able to easily figure out if a particular data set fits in 5MB or not.

[gianm]

Looks good. Out of curiosity, given that we exclude null values, how are null values in the left table matched?

I was thinking that because it's an inner join, null values from the left table are supposed to be dropped anyway. (That's why I didn't allow this optimization to trigger for left joins.)

[abhishekagarwal87]

Looks good. Out of curiosity, given that we exclude null values, how are null values in the left table matched?

I was thinking that because it's an inner join, null values from the left table are supposed to be dropped anyway. (That's why I didn't allow this optimization to trigger for left joins.)

Got it. Thanks for clarifying.

[clintropolis]

👍

[gianm]

@abhishekagarwal87 any other thoughts?