opt, sql: LEFT OUTER and ANTI spatial joins cannot be index-accelerated

[rytaft]

Inner and semi joins with a spatial ON condition such as ST_Intersects(g1, g2) can take advantage of an inverted index on either g1 or g2 to perform an inverted join and get better performance than the alternative approach of performing a cross join followed by a filter.

For example, running explain on the the following inner join query shows that it will use an inverted join:

> EXPLAIN SELECT *
FROM nyc_census_blocks AS c
JOIN nyc_neighborhoods AS n
ON ST_Covers(c.geom, n.geom);
         tree        |         field         |                 description
---------------------+-----------------------+-----------------------------------------------
                     | distribution          | full
                     | vectorized            | false
  lookup join        |                       |
   │                 | table                 | nyc_census_blocks@primary
   │                 | equality              | (gid) = (gid)
   │                 | equality cols are key |
   │                 | pred                  | st_covers(geom, geom)
   └── inverted join |                       |
        │            | table                 | nyc_census_blocks@nyc_census_blocks_geom_idx
        └── scan     |                       |
                     | estimated row count   | 129
                     | table                 | nyc_neighborhoods@primary
                     | spans                 | FULL SCAN
(13 rows)

Time: 1.599ms

However, this is not the case for left outer or anti joins. For example, running explain on the following left join query shows that it performs an inefficient cross join:

> EXPLAIN SELECT *
FROM nyc_census_blocks AS c
LEFT JOIN nyc_neighborhoods AS n
ON ST_Covers(c.geom, n.geom);
           tree           |        field        |        description
--------------------------+---------------------+----------------------------
                          | distribution        | full
                          | vectorized          | true
  cross join (left outer) |                     |
   │                      | pred                | st_covers(geom, geom)
   ├── scan               |                     |
   │                      | estimated row count | 38794
   │                      | table               | nyc_census_blocks@primary
   │                      | spans               | FULL SCAN
   └── scan               |                     |
                          | estimated row count | 129
                          | table               | nyc_neighborhoods@primary
                          | spans               | FULL SCAN
(12 rows)

Time: 2.357ms

The reason left joins cannot use the inverted index is because as shown in the first explain, the inverted join must be wrapped in a lookup join with the primary index of the table. This is necessary because inverted joins for spatial predicates produce some false positives. To eliminate the false positives, we need the original geometry or geography column. This column is not stored in the inverted index, so we always need to fetch it from the primary index in order to run the original predicate function and remove the false positives.

In order for this to work correctly for a left join, the inverted join would need to return all of the rows returned by an inner join, plus any rows from the left side that didn't match, with the right columns NULL-extended. Then the lookup join would also need to return all of the rows returned by an inner join, plus any rows from the original left side that didn't match after removal of false positives.

The problem is that the lookup join join currently has no way of knowing that the rows passed in from the inverted join actually represent one or more copies of each of the original left side rows. As a result, it incorrectly returns some rows that should have been eliminated as false positives.

For example, suppose we have the following two tables, with an inverted index on g2:

      t1                        t2

 k1 |      g1               k2 |      g2 
----+-------------------   ----+------------------
  1 | POINT(1.0 1.0)        10 | POINT(1.2 1.2)
  2 | POINT(1.5 1.5)        20 | POINT(1.5 1.5)
  3 | POINT(40.0 40.0)      30 | POINT(40.0 40.0)
  4 | POINT(-10.0 40.0) 

We want to execute the following query:

SELECT k1, k2 FROM t1 LEFT JOIN t2 ON ST_Intersects(g1, g2);

A left inverted join would return the following rows:

 k1 |        g1         |  k2
----+-------------------+------
  1 | POINT(1.0 1.0)    |   10   // false positive
  1 | POINT(1.0 1.0)    |   20   // false positive
  2 | POINT(1.5 1.5)    |   10   // false positive
  2 | POINT(1.5 1.5)    |   20
  3 | POINT(40.0 40.0)  |   30
  4 | POINT(-10.0 40.0) | NULL

The lookup join (prior to projecting away the unneeded columns) would then return the following rows:

 k1 |        g1         |  k2  |        g2
----+-------------------+------+------------------
  1 | POINT(1.0 1.0)    |   10 | NULL 
  1 | POINT(1.0 1.0)    |   20 | NULL 
  2 | POINT(1.5 1.5)    |   10 | NULL 
  2 | POINT(1.5 1.5)    |   20 | POINT(1.5 1.5)
  3 | POINT(40.0 40.0)  |   30 | POINT(40.0 40.0)
  4 | POINT(-10.0 40.0) | NULL | NULL

This is incorrect. What it should be returning is:

 k1 |        g1         |  k2  |        g2
----+-------------------+------+------------------
  1 | POINT(1.0 1.0)    | NULL | NULL 
  2 | POINT(1.5 1.5)    |   20 | POINT(1.5 1.5)
  3 | POINT(40.0 40.0)  |   30 | POINT(40.0 40.0)
  4 | POINT(-10.0 40.0) | NULL | NULL

However, there is no way to get that result without adding additional bookkeeping to the inverted joiner and/or the lookup joiner.

Workarounds

Left join queries like the one described above return correct results (because we do not use the inverted index), but they may be very slow. An alternative is to rewrite the query so that it first performs an inner join followed by a left join. For example, the above left join can be rewritten as follows:

> EXPLAIN SELECT inr.*
FROM nyc_census_blocks as c_outer LEFT JOIN (SELECT * FROM nyc_census_blocks AS c
JOIN nyc_neighborhoods AS n ON ST_Covers(c.geom, n.geom)) AS inr ON c_outer.gid = inr.gid;
           tree           |         field         |                 description
--------------------------+-----------------------+-----------------------------------------------
                          | distribution          | full
                          | vectorized            | false
  hash join (right outer) |                       |
   │                      | equality              | (gid) = (gid)
   │                      | right cols are key    |
   ├── lookup join        |                       |
   │    │                 | table                 | nyc_census_blocks@primary
   │    │                 | equality              | (gid) = (gid)
   │    │                 | equality cols are key |
   │    │                 | pred                  | st_covers(geom, geom)
   │    └── inverted join |                       |
   │         │            | table                 | nyc_census_blocks@nyc_census_blocks_geom_idx
   │         └── scan     |                       |
   │                      | estimated row count   | 129
   │                      | table                 | nyc_neighborhoods@primary
   │                      | spans                 | FULL SCAN
   └── scan               |                       |
                          | estimated row count   | 38794
                          | table                 | nyc_census_blocks@primary
                          | spans                 | FULL SCAN
(20 rows)

Time: 2.362ms

Notice that it now uses an inverted join, but also requires an additional scan of the nyc_census_blocks table. Depending on the query and data, this may or may not be faster than the cross join.

Possible Solutions

One possible solution would be to create an exploration rule that performs the transformation described in the previous section: convert the original left join to an inner join, then wrap it in an additional left join. We could also do something similar in which the left input is buffered, so it doesn't need to be re-scanned. This would be like moving the left input to a CTE.

Another alternative is to create a new execution operator or augment the existing joinReader and invertedJoiner operators. These operators would need to do some extra bookkeeping to keep track of how many rows had been output from the left side, and ensure that each row with no match on the right side was only output once.

Once left joins are supported, anti joins can be supported by converting them to a left join wrapped in a Select operator followed by a Project.

[sumeerbhola]

augment the existing joinReader and invertedJoiner operators. These operators would need to do some extra bookkeeping to keep track of how many rows had been output from the left side, and ensure that each row with no match on the right side was only output once.

I am leaning towards this (as I mentioned in-person) -- @yuzefovich and @asubiotto do you have concerns about adding another mode to joinReader (I can prototype something to show you). The invertedJoiner would add a count column to its output, representing the size of a group, since it knows all the matches for a left row before outputting anything for a left row. So using the earlier example, the invertedJoiner would output

 k1 |        g1         |  k2   | count
----+-------------------+-------+----
  1 | POINT(1.0 1.0)    |   10  | 2  // false positive
  1 | POINT(1.0 1.0)    |   20  | 2  // false positive
  2 | POINT(1.5 1.5)    |   10  | 2  // false positive
  2 | POINT(1.5 1.5)    |   20  | 2
  3 | POINT(40.0 40.0)  |   30  | 1
  4 | POINT(-10.0 40.0) | NULL  | 1

The joinReader would be configured with the knowledge that k1, g1 are the "real left" columns, and that count is a special book-keeping column. The group of input rows for a "real left" row will be consecutive (I am assuming that when distributing, both an invertedJoiner and joinReader are in the same node and exactly one invertedJoiner is feeding a joinReader), but they can be split across multiple batches (hence the count book-keeping column). The joinReader can continue outputting rows that match the predicate as usual (all the joinReaderStrategy implementations are applicable), and would keep track of whether anything has matched for the current group. This tracking of whether anything matched would extend beyond the batch boundary if the group was not complete -- there would be at most one such group. Only when all the rows of a group were done would it potentially output the NULL-extended row.

[yuzefovich]

I think as long as the performance of the joinReader is not impacted in other cases, it's fine to add another mode for this use case.

[asubiotto]

I agree, but it makes me slightly uncomfortable that we'll be introducing a dependency on the invertedJoiner. Is there a way to fix this issue by only changing the invertedJoiner behavior?

[sumeerbhola]

Is there a way to fix this issue by only changing the invertedJoiner behavior?

I can't think of any. The problem here is that the inverted joiner (the first joiner in the pair of joins, in the PR I just sent for review), is one that is producing false positives. So the second joiner, which has perfect information (but we used the first joiner for efficiency, because it had an index that was suited to the join expression), has to view the stream of input rows as groups corresponding to an original left row.

[sumeerbhola]

With #55216, #55742, #55986, we are now using paired joins for left outer/semi/anti joins.