opt: create inverted or lookup joins for left/anti joins in more cases

[rytaft]

As described in #59615, we can no longer plan left/anti lookup/inverted joins for certain types of queries where the prefix columns in the index are constrained to multiple constant values, since doing so could result in too many rows in the output. However, the alternative of requiring a hash or merge join is suboptimal for performance, especially for mutation queries that rely on using left/anti joins for UPSERT/INSERT ... ON CONFLICT execution and foreign key checks.

The most efficient way to support these joins safely would require execution support to allow performing each lookup with multiple spans instead of a single equality condition. However, as a stopgap, we could support them in the optimizer by constructing a UNION of several lookup joins. We may also be able to use the infrastructure of the paired joins.

As described in #59615, this has implications for multi-region support, because these types of queries will show up often due to the ENUM type crdb_region column.

cc @mgartner

Jira issue: CRDB-3262

[mgartner]

However, as a stopgap, we could support them in the optimizer by constructing a UNION of several lookup joins.

I'm not sure this will work because UNION will remove duplicates that exist within just the left or just the right side:

select * from (values (1), (1)) union select * from (values (2));
  column1
-----------
        1
        2

In a case like below, I think we'd incorrectly de-duplicate the (1, 1, 100) rows:

show create table t;
  table_name |          create_statement
-------------+--------------------------------------
  t          | CREATE TABLE public.t (
             |     a INT8 NULL,
             |     b INT8 NULL,
             |     INDEX t_b_a_idx (b ASC, a ASC),
             |     FAMILY "primary" (a, b, rowid)
             | )
select * from t;
  a |  b
----+------
  1 | 100
  1 | 100

select * from (values (1), (2)) as s (a) left join t on s.a = t.a and b IN (100, 200);
  a |  a   |  b
----+------+-------
  1 |    1 |  100
  1 |    1 |  100
  2 | NULL | NULL

But it might work if we ensured there was a strict key in the input? In that case, would a distinct-on above the lookup join with all columns as grouping column be equivalent?

[rytaft]

Yea, I agree that we have to ensure that a strict key exists on the input. I think we have functions that do this by either adding the primary key back to a scan or wrapping everything in an ordinality operator.

In that case, would a distinct-on above the lookup join with all columns as grouping column be equivalent?

Do we need the distinct-on if we use a UNION (assuming the input has a strict key)? We would just need to project away the key if it was added.

EDIT: Oh I see -- yea, your suggestion might be better than a union.

[mgartner]

Something still feels fishy about a distinct-on or a UNION to me. A left-join can produce multiple rows from the left with the same PK, so a distinct-on or UNION don't seem valid.

For example:

marcus=# create table t (a int primary key, b int);
CREATE TABLE

marcus=# create table s (b int);
CREATE TABLE

marcus=# insert into t values (1, 100), (2, 200);
INSERT 0 2

marcus=# insert into s values (100), (100);
INSERT 0 2

marcus=# select * from t left join s on t.b = s.b;
 a |  b  |   b
---+-----+--------
 1 | 100 |    100
 1 | 100 |    100
 2 | 200 |   NULL
(3 rows)

The presence of the PK a doesn't allow us to eliminate distinct rows in this case.

[rytaft]

Hmm good point. Another issue is that there can be some null-extended rows that appear for rows that do have a match. Here's an example with the buggy code from before the fix merged:

[email protected]:26257/movr> CREATE TABLE t (
  x INT NOT NULL CHECK (x IN (1, 3, 5)),
  y INT NOT NULL,
  z INT,
  PRIMARY KEY (x, y)
);
CREATE TABLE

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

[email protected]:26257/movr> INSERT INTO t SELECT 1, generate_series(1, 1000);
INSERT 1000

Time: 9ms total (execution 8ms / network 0ms)

[email protected]:26257/movr> INSERT INTO t SELECT 3, generate_series(1, 1000);
INSERT 1000

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

[email protected]:26257/movr> ANALYZE t;
ANALYZE

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

[email protected]:26257/movr> SELECT * FROM (VALUES (1), (2000)) AS u (y) LEFT JOIN t ON u.y = t.y;
   y   |  x   |  y   |  z
-------+------+------+-------
     1 |    1 |    1 | NULL
     1 |    3 |    1 | NULL
  2000 | NULL | NULL | NULL
  2000 | NULL | NULL | NULL
     1 | NULL | NULL | NULL
  2000 | NULL | NULL | NULL
(6 rows)

Notice the null-extended row starting with 1 in addition to the null-extended rows starting with 2000. We have to make sure that the null-extended row starting with 1 doesn't appear at all.

[rytaft]

Here's something that I think would work:

For anti joins, we would use the existing cross-join approach and simply wrap it with a group by + having clause. Using the example from above with 3 possible values in the check constraint:

CREATE TABLE t (
  x INT NOT NULL CHECK (x IN (1, 3, 5)),
  y INT NOT NULL,
  z INT,
  PRIMARY KEY (x, y)
);

The following two queries are equivalent:

SELECT * FROM (VALUES (1), (2000)) AS u (y) WHERE NOT EXISTS (
  SELECT * FROM t WHERE u.y = t.y
);

SELECT u.y FROM (
  (VALUES (1), (2000)) INNER JOIN (VALUES (1), (3), (5)) ON true
) AS u(y, x) WHERE NOT EXISTS (
  SELECT * FROM t WHERE u.y = t.y AND u.x = t.x
) GROUP BY u.y HAVING count(*) = 3;

We would need to ensure that the input has a strict key and the count in the having clause is equal to the number of values in the right side of the cross join.

For a left join, we could simply take the output of the anti join, null-extend it, and UNION ALL with the output of an inner join.

It's not ideal, but I haven't been able to think of any other correct solution yet that doesn't require execution changes. I also haven't looked at how hard it would be to make the execution changes -- maybe it would be better to just bite the bullet and do that.

[rytaft]

Probably a good idea to get @yuzefovich's opinion on this. @yuzefovich, how hard do you think it would be to support looking up several spans in the lookup joiner, instead of just a single equality span? For example, if the index has columns (x, y) and the join condition is ON (x = 3 OR x = 5) AND y = input.y, each row of the input would require looking up two spans, [/3/val - /3/val] [/5/val - /5/val].

[yuzefovich]

I think @asubiotto knows the joinReader code better, but I'd guess it won't be that hard of a change.

[asubiotto]

Mechanically I don't think it'll be that difficult (looks like we can just put the defaultSpanGenerator behind an interface and add another implementation). What I'm less sure of is how we're going to actually generate these spans. I assume the optimizer can hand down spans to be "filled in" with the values? Would that cover all cases?

[rytaft]

Ah ok perfect. What probably makes sense is for the optimizer to provide the new implementation for the SpanGenerator interface (or whatever we decide to call it), similar to how the optimizer implements the DatumsToInvertedExpr interface for the invertedJoiner. If we use that model, we would just pass ON conditions to the joinReader as expressions with an IndexedVar, which the joinReader can use to initialize the optimizer implementation of the SpanGenerator.

[sumeerbhola]

Reading this issue and #59615, it seems very similar to the problem that paired joins is solving in other contexts: the rows generated by the join that applies the check constraint are grouped corresponding to the original left row. In the case that nothing in the group matches in the lookup join, only one row should be output (for left outer join and left semi join). It would need the cross join to output groups with the continuation column, which seems straightforward. And this should also distribute cleanly since each group will be produced in one place.
Enumerating the rows after applying the check constraint seems easier on the execution and optimizer machinery than carrying the potentially complex check constraint into the lookup join, which IIUC is the current alternative being explored.
Thoughts?

[rytaft]

I did briefly consider using the paired joins machinery for this, and I think we could probably make it work, but I've already opened a PR for looking up multiple spans in lookup joins. Feel free to take a look: #60302. The advantage of the multi-span approach is that we should be able to extend it to support other use cases described in #51576.

[rytaft]

#60302 will solve this issue for lookup joins, however we still need to solve this issue for inverted joins. I'm guessing the solution will look similar, but it will still require a non-trivial amount of work to update both the optimizer and execution sides. Therefore this is very unlikely to make it into the 21.1 release.

I'm adding the docs-known-limitation tag (cc @ianjevans, @taroface) in order to document this limitation: any left or anti join involving JSON, Array, or spatial columns with a multi-column or partitioned inverted index will not be able to take advantage of the index if the prefix columns are unconstrained or constrained to multiple constant values. Instead, we will need to plan a cross join with the filter applied after the fact, which may be significantly slower than an inverted join.

Possible workarounds: ensure that the prefix columns are constrained to single constant values or equated with input columns. For example, consider the following database:

CREATE DATABASE multi_region_test_db PRIMARY REGION "europe-west1" REGIONS "us-west1", "us-east1" SURVIVE REGION FAILURE;
USE multi_region_test_db;

CREATE TABLE t1 (
  k INT PRIMARY KEY,
  geom GEOMETRY
);

CREATE TABLE t2 (
  k INT PRIMARY KEY,
  geom GEOMETRY,
  INVERTED INDEX geom_idx (geom)
) LOCALITY REGIONAL BY ROW;

INSERT INTO t1 SELECT generate_series(1, 1000), 'POINT(1.0 1.0)';
INSERT INTO t2 (crdb_region, k, geom) SELECT 'us-east1', generate_series(1, 1000), 'POINT(1.0 1.0)';
INSERT INTO t2 (crdb_region, k, geom) SELECT 'us-west1', generate_series(1001, 2000), 'POINT(2.0 2.0)';
INSERT INTO t2 (crdb_region, k, geom) SELECT 'europe-west1', generate_series(2001, 3000), 'POINT(3.0 3.0)';
ANALYZE t1;
ANALYZE t2;

If we try to perform a left join between these two tables only on the geometry columns, we will not be able to plan an inverted join due to the limitation described above:

> EXPLAIN SELECT * FROM t1 LEFT JOIN t2 ON st_contains(t1.geom, t2.geom);
                info
------------------------------------
  distribution: full
  vectorized: true

  • cross join (right outer)
  │ pred: st_contains(geom, geom)
  │
  ├── • scan
  │     estimated row count: 3,000
  │     table: t2@primary
  │     spans: FULL SCAN
  │
  └── • scan
        estimated row count: 1,000
        table: t1@primary
        spans: FULL SCAN
(15 rows)

However, if we can constrain the crdb_region column to a single value, we can plan an inverted join:

> EXPLAIN SELECT * FROM t1 LEFT JOIN t2 ON st_contains(t1.geom, t2.geom) AND t2.crdb_region = 'us-east1';
                       info
--------------------------------------------------
  distribution: full
  vectorized: true

  • lookup join (left outer)
  │ table: t2@primary
  │ equality: (crdb_region, k) = (crdb_region,k)
  │ equality cols are key
  │ pred: st_contains(geom, geom)
  │
  └── • inverted join (left outer)
      │ table: t2@geom_idx
      │
      └── • render
          │
          └── • scan
                estimated row count: 1,000
                table: t1@primary
                spans: FULL SCAN
(18 rows)

If the user doesn't know which region to use, the workaround is to construct a UNION ALL as follows:

> EXPLAIN SELECT * FROM t1 LEFT JOIN t2 ON st_contains(t1.geom, t2.geom) AND t2.crdb_region = 'us-east1'
UNION ALL SELECT * FROM t1 LEFT JOIN t2 ON st_contains(t1.geom, t2.geom) AND t2.crdb_region = 'us-west1'
UNION ALL SELECT * FROM t1 LEFT JOIN t2 ON st_contains(t1.geom, t2.geom) AND t2.crdb_region = 'europe-west1';
                           info
----------------------------------------------------------
  distribution: full
  vectorized: true

  • union all
  │
  ├── • union all
  │   │
  │   ├── • lookup join (left outer)
  │   │   │ table: t2@primary
  │   │   │ equality: (crdb_region, k) = (crdb_region,k)
  │   │   │ equality cols are key
  │   │   │ pred: st_contains(geom, geom)
  │   │   │
  │   │   └── • inverted join (left outer)
  │   │       │ table: t2@geom_idx
  │   │       │
  │   │       └── • render
  │   │           │
  │   │           └── • scan
  │   │                 estimated row count: 1,000
  │   │                 table: t1@primary
  │   │                 spans: FULL SCAN
  │   │
  │   └── • lookup join (left outer)
  │       │ table: t2@primary
  │       │ equality: (crdb_region, k) = (crdb_region,k)
  │       │ equality cols are key
  │       │ pred: st_contains(geom, geom)
  │       │
  │       └── • inverted join (left outer)
  │           │ table: t2@geom_idx
  │           │
  │           └── • render
  │               │
  │               └── • scan
  │                     estimated row count: 1,000
  │                     table: t1@primary
  │                     spans: FULL SCAN
  │
  └── • lookup join (left outer)
      │ table: t2@primary
      │ equality: (crdb_region, k) = (crdb_region,k)
      │ equality cols are key
      │ pred: st_contains(geom, geom)
      │
      └── • inverted join (left outer)
          │ table: t2@geom_idx
          │
          └── • render
              │
              └── • scan
                    estimated row count: 1,000
                    table: t1@primary
                    spans: FULL SCAN
(54 rows)

[github-actions]

We have marked this issue as stale because it has been inactive for
18 months. If this issue is still relevant, removing the stale label
or adding a comment will keep it active. Otherwise, we'll close it in
10 days to keep the issue queue tidy. Thank you for your contribution
to CockroachDB!