sql: enable lookup joins to lookup spans, not just individual keys

[andy-kimball]

Imagine you have tables like this:

CREATE TABLE metrics (
	id   SERIAL PRIMARY KEY,
	name STRING,
	INDEX name_index (name)
);
CREATE TABLE metric_values (
	metric_id INT8,
	time      TIMESTAMPTZ,
	value     INT8,
	PRIMARY KEY (metric_id, time)
);

Now say I want to look up a 10 minute interval of metric values for a particular metric:

SELECT *
FROM metric_values
INNER JOIN metrics
ON metric_id=id
WHERE
	time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
	name='cpu.percent'
;

This is the resulting plan:

inner-join (lookup metric_values)
 ├── scan metrics@name_index
 │    └── constraint: /6/5: [/'cpu.percent' - /'cpu.percent']
 └── filters
      └── (time >= '2020-01-01 00:00:00+00:00') AND (time <= '2020-01-01 00:10:00+00:00')

Notice that the time filter, which is actually the most important thing about this query, is not part of the lookup join criteria. That's because lookup joins only allow equality conditions for lookup. But what if lookup joins allowed (multi-column) spans? It'd make dynamic lookup cases like this vastly more efficient. In this example, what we really want is for the lookup join to operate on spans like this:

/id/time
[/1/'2020-01-01 00:00:00+00:00' - /1/'2020-01-01 00:10:00+00:00']
[/5/'2020-01-01 00:00:00+00:00' - /5/'2020-01-01 00:10:00+00:00']
[/6/'2020-01-01 00:00:00+00:00' - /6/'2020-01-01 00:10:00+00:00']
...

rather than simply as it does today:

/id
[/1 - /1]
[/5 - /5]
[/6 - /6]

Jira issue: CRDB-4031
Epic: CRDB-14091

gz#11736

[awoods187]

This is a cool enhancement idea!

[RaduBerinde]

A solution inside the optimizer would be for LookupJoin to contain a Constraint that is similar to an index constraint, but applies to the index columns after the equality columns. I think the opt part wouldn't be hard (we can reuse code from index constraints), but we'd also need execution support.

[andy-kimball]

Note that if we don't support this, then it puts the burden on the user to do 2 separate queries for cases where the lookup keys are not known beforehand. For example, the user could manually issue these two queres:

1. Query # 1: Fetch the list of interesting metrics and store in a temporary app variable:

SELECT id FROM metrics WHERE name='cpu.percent';

2. Query # 2: Pass the list gathered in step # 1 as a parameter to a query that gets the needed time ranges:

SELECT *
FROM metric_values
WHERE
	time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
	metric_id IN $1
;

The difference in cost between doing it with lookup join vs. doing two manual queries is substantial. In one experiment, the lookup join costs is a whopping 204,149, whereas the combined manual queries cost only 27,343 - an order of magnitude difference.

[andy-kimball]

A solution inside the optimizer would be for LookupJoin to contain a Constraint that is similar to an index constraint, but applies to the index columns after the equality columns. I think the opt part wouldn't be hard (we can reuse code from index constraints), but we'd also need execution support.

Adding support for this would definitely require a partnership between the SQL Execution and Optimizer teams. I don't think it'd be terribly hard and would bring a lot of "bang for the buck".

CC @jordanlewis

[jordanlewis]

Cool, this seems like a great idea. So what would the API look like in the JoinReader proto definition? An "extra constraint column set", and a pair of constant values that fill in that set?

I suppose it'd also be possible to make this dynamic, if you wanted to construct the span range based on values that you find in the data... maybe this is a marginal use case in this example, but I bet we could find some real world ones as well.

CREATE TABLE time_spans(
   id INT PRIMARY KEY,
   start TIMESTAMPTZ,
   end TIMESTAMPTZ
);

SELECT *
FROM time_spans,
metric_values
INNER JOIN metrics
ON metric_id=id
WHERE
	time BETWEEN start AND end AND
	name='cpu.percent' AND
    time_spans.id=$1
;

[RaduBerinde]

I think we handle constant columns today (by projecting the values before the lookup join).

It would be a list of pseudo-spans, i.e. spans with key pieces that get appended to the keys in each span generated in the lookup joiner. In the example above, the pseudo-spans would be [/'2020-01-01 00:00:00+00:00' - /'2020-01-01 00:10:00+00:00']. The lookup joiner would take a span like [/1 - /1] (note that all spans generated by the lookup joiner have equal start and end keys) and append the pseudo-spans to get [/1/'2020-01-01 00:00:00+00:00' - /1/'2020-01-01 00:10:00+00:00'].

It's a little more tricky with multiple pseudo-spans - we would effectively be breaking up one span [/1 - /1] into multiple spans, e.g. [/1/1 - /1/5] [/1/10 - /1/15] [/1/20-/1/25]. This might require a bit more book-keeping in the lookup joiner. I think we would benefit even from supporting just one pseudo-span at first though.

Conceptually, if we have an index on (a,b,c,d) and a lookup join with equality columns (a,b), we think of each lookup as a scan into an index on (c,d) (and the pseudo-spans allow us to constrain that scan).

[jordanlewis]

Flagging as a potential project for @helenmhe if she has time after her current work on join reader.

[nvanbenschoten]

This would be a nice win for TPC-C. It shows up in the Stock-Level transaction. Right now, we issue two read-only statements in a read-only transaction. If we were able to combine them into a single statement then we'd be able to use an implicit transaction instead of an explicit transaction. So doing so would not only avoid one statement but would actually drop the transaction down from 6 statements (BEGIN; SAVEPOINT; SELECT ... district; SELECT ... order_line; RELEASE SAVEPOINT; COMMIT) to 1 statement (SELECT ... (SELECT district) ... stock). As we keep seeing, one of the optimizations with the biggest bang for its buck for workloads that bottleneck on CPU is minimizing the statement count.

For reference:

-- Currently
root@:26257/tpcc> EXPLAIN SELECT d_next_o_id
FROM district
WHERE d_w_id = 0 AND d_id = 3;
  tree |        field        |   description
-------+---------------------+-------------------
       | distribution        | local
       | vectorized          | false
  scan |                     |
       | estimated row count | 1
       | table               | district@primary
       | spans               | [/0/3 - /0/3]

root@:26257/tpcc> EXPLAIN SELECT count(DISTINCT s_i_id)
FROM order_line
JOIN stock
ON s_w_id = 0 AND s_i_id = ol_i_id
WHERE ol_w_id = 0
  AND ol_d_id = 3
  AND ol_o_id BETWEEN 3095 - 20 AND 3095 - 1
  AND s_quantity < 15;
            tree           |         field         |                    description
---------------------------+-----------------------+-----------------------------------------------------
                           | distribution          | full
                           | vectorized            | false
  group (scalar)           |                       |
   └── distinct            |                       |
        │                  | distinct on           | s_i_id
        └── lookup join    |                       |
             │             | table                 | stock@primary
             │             | equality              | (project_const_col_@14, ol_i_id) = (s_w_id,s_i_id)
             │             | equality cols are key |
             │             | pred                  | s_quantity < 15
             └── render    |                       |
                  └── scan |                       |
                           | estimated row count   | 117
                           | table                 | order_line@primary
                           | spans                 | [/0/3/3094 - /0/3/3075]


-- With Subquery (today)
root@:26257/tpcc> EXPLAIN SELECT count(DISTINCT s_i_id)
  FROM (SELECT d_w_id, d_id, d_next_o_id FROM district WHERE d_w_id = 0 AND d_id = 3)
  JOIN order_line ON  ol_w_id = d_w_id
                  AND ol_d_id = d_id
                  AND ol_o_id BETWEEN (d_next_o_id - 20) AND (d_next_o_id - 1)
  JOIN stock ON s_w_id = d_w_id AND s_i_id = ol_i_id
 WHERE s_quantity < 15;

             tree            |         field         |                            description
-----------------------------+-----------------------+---------------------------------------------------------------------
                             | distribution          | full
                             | vectorized            | false
  group (scalar)             |                       |
   └── distinct              |                       |
        │                    | distinct on           | s_i_id
        └── lookup join      |                       |
             │               | table                 | stock@primary
             │               | equality              | (d_w_id, ol_i_id) = (s_w_id,s_i_id)
             │               | equality cols are key |
             │               | pred                  | s_quantity < 15
             └── lookup join |                       |
                  │          | table                 | order_line@primary
                  │          | equality              | (d_w_id, d_id) = (ol_w_id,ol_d_id)
                  │          | pred                  | (ol_o_id >= (d_next_o_id - 20)) AND (ol_o_id <= (d_next_o_id - 1))
                  └── scan   |                       |
                             | estimated row count   | 1
                             | table                 | district@primary
                             | spans                 | [/0/3 - /0/3]

-- With Subquery (in future with this issue addressed)
root@:26257/tpcc> EXPLAIN SELECT count(DISTINCT s_i_id)
  FROM (SELECT d_w_id, d_id, d_next_o_id FROM district WHERE d_w_id = 0 AND d_id = 3)
  JOIN order_line ON  ol_w_id = d_w_id
                  AND ol_d_id = d_id
                  AND ol_o_id BETWEEN (d_next_o_id - 20) AND (d_next_o_id - 1)
  JOIN stock ON s_w_id = d_w_id AND s_i_id = ol_i_id
 WHERE s_quantity < 15;

             tree            |         field         |                            description
-----------------------------+-----------------------+---------------------------------------------------------------------
                             | distribution          | full
                             | vectorized            | false
  group (scalar)             |                       |
   └── distinct              |                       |
        │                    | distinct on           | s_i_id
        └── lookup join      |                       |
             │               | table                 | stock@primary
             │               | equality              | (d_w_id, ol_i_id) = (s_w_id,s_i_id)
             │               | equality cols are key |
             │               | pred                  | s_quantity < 15
             └── lookup join |                       |
                  │          | table                 | order_line@primary
                  │          | equality              | (d_w_id, d_id) = (ol_w_id,ol_d_id)
                  │          | inequality            | (ol_o_id >= (d_next_o_id - 20)) AND (ol_o_id <= (d_next_o_id - 1))
                  └── scan   |                       |
                             | estimated row count   | 1
                             | table                 | district@primary
                             | spans                 | [/0/3 - /0/3]

[rytaft]

There are really two different issues here:

1. Support lookup joins to ranges of keys where the range is known at optimization time. This is what is described in @andy-kimball's issue description with the predicate time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00'. This should be relatively straightforward to add as an extension of the work done in sql,opt: add support for lookup joins with multiple spans #60302 -- we just need to add support for inequality operators (>,>=,<,<=) in addition to the support that already exists for = and IN.
2. Support lookup joins where the join condition is an inequality between a column in the index and an expression over one or more columns in the input. This is what is described in @nvanbenschoten's comment about TPC-C. The joinReader will need to learn how to evaluate expressions to produce spans, and the optimizer will need to be able to detect that a join condition fits the requirements. In the case of TPC-C, this is as simple as noting that ol_o_id is a single variable on the left hand side, while d_next_o_id is the only outer column on the right hand side. However, we could also support more complex predicates such as CASE WHEN b THEN a > c ELSE a > d END where a is a column in the index and b, c, and d are columns in the input. The optimizer would need to be able to do some sort of constant folding simulation to determine that this is a valid lookup join predicate.

[michae2]

This is blocking an important POC and I think we should prioritize it.

[michae2]

Here's a demonstration of a kind of query that would benefit:

CREATE TABLE t (i PRIMARY KEY) AS SELECT generate_series(0, 999999);
CREATE TABLE u (j PRIMARY KEY, k) AS SELECT n, n FROM generate_series(0, 999999) AS s(n);
SELECT (SELECT k FROM u WHERE j > i ORDER BY j LIMIT 1) FROM t WHERE i > 999997 ORDER BY i LIMIT 1;

With > in the subquery, we have to use a cross join over a full scan of u:

[email protected]:26257/defaultdb> EXPLAIN SELECT (SELECT k FROM u WHERE j > i ORDER BY j LIMIT 1) FROM t WHERE i > 999997 ORDER BY i LIMIT 1;
                                                   info
-----------------------------------------------------------------------------------------------------------
  distribution: local
  vectorized: true

  • render
  │ estimated row count: 1
  │
  └── • top-k
      │ estimated row count: 1
      │ order: +i
      │ k: 1
      │
      └── • distinct
          │ distinct on: i
          │
          └── • sort
              │ estimated row count: 67
              │ order: +j
              │
              └── • cross join (right outer)
                  │ estimated row count: 67
                  │ pred: j > i
                  │
                  ├── • scan
                  │     estimated row count: 1,000,000 (100% of the table; stats collected 2 minutes ago)
                  │     table: u@u_pkey
                  │     spans: FULL SCAN
                  │
                  └── • scan
                        estimated row count: 0 (<0.01% of the table; stats collected 2 minutes ago)
                        table: t@t_pkey
                        spans: [/999998 - ]
(31 rows)


Time: 2ms total (execution 2ms / network 0ms)

But we really want to use a lookup join, which we can see if we change the > in the subquery to =:

[email protected]:26257/defaultdb> EXPLAIN SELECT (SELECT k FROM u WHERE j = i ORDER BY j LIMIT 1) FROM t WHERE i > 999997 ORDER BY i LIMIT 1;
                                          info
-----------------------------------------------------------------------------------------
  distribution: local
  vectorized: true

  • render
  │ estimated row count: 1
  │
  └── • lookup join (left outer)
      │ estimated row count: 1
      │ table: u@u_pkey
      │ equality: (i) = (j)
      │ equality cols are key
      │
      └── • scan
            estimated row count: 0 (<0.01% of the table; stats collected 3 minutes ago)
            table: t@t_pkey
            spans: [/999998 - ]
            limit: 1
(17 rows)


Time: 2ms total (execution 1ms / network 1ms)

[DrewKimball]

Closing as fixed by #85597.