Skip to content

Incorrect results for NOT IN subqueries with nulls #8241

Closed
Closed
Incorrect results for NOT IN subqueries with nulls#8241
Labels
bugSomething isn't working

Description

@msirek
msirek
opened on Nov 16, 2023

Describe the bug

Datafusion does not use three-value logic (true, false, unknown) for evaluating NOT IN predicates when a join is required, causing incorrect results in the presence of nulls.
An expression without a join evaluates correctly:

select 1 NOT IN (3, null) IS NULL;
+----------------------------------------------------------------------------------+
| Int64(1) NOT IN (Map { iter: Iter([Literal(Int64(3)), Literal(NULL)]) }) IS NULL |
+----------------------------------------------------------------------------------+
| true                                                                             |
+----------------------------------------------------------------------------------+

The expression 1 NOT IN (3, null) is unknown (null), which if used as WHERE or ON clause filter should disqualify the row:

select 'qualified filter' where 1 NOT IN (3, null);
0 rows in set. Query took 0.007 seconds.

Something similar involving a relation instead of a pure scalar expression returns wrong results:

select 'qualified filter' where 1 NOT IN (select * from (values (3), (null)) my_tab);
+--------------------------+
| Utf8("qualified filter") |
+--------------------------+
| qualified filter         |
+--------------------------+

To Reproduce

Here is an example involving base tables:

create table t1 (a int);
create table t2 (a int);
insert into t1 values (1);
insert into t2 values (3), (null);

-- The correct answer is no rows
select * from t1 where a not in (select a from t2);
+---+
| a |
+---+
| 1 |
+---+
1 row in set. Query took 0.015 seconds.

❯ explain select * from t1 where a not in (select a from t2);
+---------------+-------------------------------------------------------------------------+
| plan_type     | plan                                                                    |
+---------------+-------------------------------------------------------------------------+
| logical_plan  | LeftAnti Join: t1.a = __correlated_sq_1.a                               |
|               |   TableScan: t1 projection=[a]                                          |
|               |   SubqueryAlias: __correlated_sq_1                                      |
|               |     TableScan: t2 projection=[a]                                        |
| physical_plan | CoalesceBatchesExec: target_batch_size=8192                             |
|               |   HashJoinExec: mode=Partitioned, join_type=LeftAnti, on=[(a@0, a@0)]   |
|               |     CoalesceBatchesExec: target_batch_size=8192                         |
|               |       RepartitionExec: partitioning=Hash([a@0], 32), input_partitions=1 |
|               |         MemoryExec: partitions=1, partition_sizes=[1]                   |
|               |     CoalesceBatchesExec: target_batch_size=8192                         |
|               |       RepartitionExec: partitioning=Hash([a@0], 32), input_partitions=1 |
|               |         MemoryExec: partitions=1, partition_sizes=[1]                   |
|               |                                                                         |
+---------------+-------------------------------------------------------------------------+

Expected behavior

NOT IN should honor three-value logic.

From the ANSI SQL spec, given a row value predicand, RVC, and an in predicate value, IPV,
RVC NOT IN IPV is equivalent to NOT ( RVC IN IPV ).
For the given example,

1 NOT IN (3, null) -> NOT (1 IN (3, null))

1 IN (3, null)      -> 1 = 3 OR 1 = null
NOT(1 IN (3, null)) -> 1 != 3 AND 1 != NULL
                    -> true AND null
                    -> null

A filter which evaluates to null should not return the row.

Additional context

Velox handles nulls in NOT IN via a null-aware antijoin.
Substrait is adding null-aware antijoins as physical join methods an engine can support.
Datafusion could add null-aware antijoin support as a solution.

A less performant alternative would be to use a cross antijoin, with a modified join predicate, as is done in CockroachDB:

explain(opt) select * from t1 where a not in (select a from t2);                       
                  info
----------------------------------------
  anti-join (cross)
   ├── scan t1
   ├── scan t2
   └── filters
        └── (t1.a = t2.a) IS NOT false

which perhaps would translate into a NestedLoopJoinExec in Datafusion:

explain select * from t1 where a not in (select a from t2);
+---------------+----------------------------------------------------------+
| plan_type     | plan                                                     |
+---------------+----------------------------------------------------------+
| logical_plan  | LeftAnti Join:  Filter: (t1.a = __correlated_sq_1.a) IS NOT false     |
|               |   TableScan: t1 projection=[a]                           |
|               |   SubqueryAlias: __correlated_sq_1                       |
|               |     TableScan: t2 projection=[a]                         |
| physical_plan | NestedLoopJoinExec: join_type=LeftAnti, filter=a@0 = a@1 IS NOT false |
|               |   MemoryExec: partitions=1, partition_sizes=[1]          |
|               |   MemoryExec: partitions=1, partition_sizes=[1]          |
|               |                                                          |
+---------------+----------------------------------------------------------+

A null-aware antijoin would need to broadcast the subquery rows with a null key to all partitions so that each partition could "see" the null. A null in the subquery would cause no rows to be returned. Processing is more complicated for tuple types, e.g. (a, b) IN (SELECT a,b ..., where one of the keys could be null and the other non-null, but Datafusion doesn't currently support tuple types.

Proposed distribution/partitioning changes to support null-aware antijoin:

  1. Add a new HashPartitionedBroadcastNull distribution method:
/// How data is distributed amongst partitions. See [`Partitioning`] for more
/// details.
#[derive(Debug, Clone)]
pub enum Distribution {
    /// Unspecified distribution
    UnspecifiedDistribution,
    /// A single partition is required
    SinglePartition,
    /// Requires children to be distributed in such a way that the same
    /// values of the keys end up in the same partition
    HashPartitioned(Vec<Arc<dyn PhysicalExpr>>),
    /// Requires children to be distributed in such a way that the same
    /// values of the keys end up in the same partition, and the first row having a key with nulls
    /// is broadcast to all partitions (typically used for the inner relation of antijoin)
    HashPartitionedBroadcastNull(Vec<Arc<dyn PhysicalExpr>>),
}

Is this better than modifying HashPartitioned with a new flag, e.g. HashPartitioned(Vec<Arc<dyn PhysicalExpr>>, bool), to control whether nulls are distributed?

  1. Add a new flag to Partitioning::Hash:
 pub enum Partitioning {
     /// Allocate batches using a round-robin algorithm and the specified number of partitions
     RoundRobinBatch(usize),
-    /// Allocate rows based on a hash of one of more expressions and the specified number of
-    /// partitions
-    Hash(Vec<Arc<dyn PhysicalExpr>>, usize),
+    /// Allocate rows based on a hash of one of more expressions, the specified number of
+    /// partitions, and whether the first null in the input should be broadcast to all partitions
+    Hash(Vec<Arc<dyn PhysicalExpr>>, usize, bool),
     /// Unknown partitioning scheme with a known number of partitions
     UnknownPartitioning(usize),
 }

Same question here. Does it make more sense to add a flag to Hash partitioning, or make a totally separate enum value?

Alternative:

  • Instead of broadcasting the nulls, create a new mode for RepartitionExec, which broadcasts a special marker row indicating a null key was seen, then end the repartition operation. The antijoin would then have to detect that special row and quit the antijoin early.
    • This approach may not work well for the tuple NOT IN case. That case must broadcast every key combination containing a null. For example, all of the following would need to be broadcast: (1, null), (2, null), (null, 1)... Only if (null, null) is seen could the repartition end early. Also, there may be a filter in the join which filters out a given row with a null, so the presence of a null in the RepartitionExec doesn't necessarily mean the null will be processed by the join.

Activity

  1. msirek
    added
    bugSomething isn't working
    on Nov 16, 2023

  2. msirek commented on Nov 16, 2023

    @msirek
    msirek
    on Nov 16, 2023
    ContributorAuthor

    I've done a little prototyping on null-aware antijoin, but am wondering if there are ideas for alternatives.

  3. yukkit commented on Nov 17, 2023

    @yukkit
    yukkit
    on Nov 17, 2023
    Contributor

    @msirek Your detailed steps to reproduce the problem and your analysis are incredibly valuable. It's an intriguing bug indeed!

  4. yukkit commented on Nov 17, 2023

    @yukkit
    yukkit
    on Nov 17, 2023
    Contributor

    Is this better than modifying HashPartitioned with a new flag, e.g. HashPartitioned(Vec<Arc<dyn PhysicalExpr>>, bool), to control whether nulls are distributed?

    1. Add a new flag to Partitioning::Hash:
     pub enum Partitioning {
     /// Allocate batches using a round-robin algorithm and the specified number of partitions
     RoundRobinBatch(usize),
-    /// Allocate rows based on a hash of one of more expressions and the specified number of
-    /// partitions
-    Hash(Vec<Arc<dyn PhysicalExpr>>, usize),
+    /// Allocate rows based on a hash of one of more expressions, the specified number of
+    /// partitions, and whether the first null in the input should be broadcast to all partitions
+    Hash(Vec<Arc<dyn PhysicalExpr>>, usize, bool),
     /// Unknown partitioning scheme with a known number of partitions
     UnknownPartitioning(usize),
 }

    Same question here. Does it make more sense to add a flag to Hash partitioning, or make a totally separate enum value?

    I'm inclined towards adding a flag to Hash partitioning because broadcast null value only occurs in hash partitioning, even if the modification of partitioning will have a slight impact on downstream.

  5. msirek commented on Nov 17, 2023

    @msirek
    msirek
    on Nov 17, 2023
    ContributorAuthor

    I'm inclined towards adding a flag to Hash partitioning because broadcast null value only occurs in hash partitioning, even if the modification of partitioning will have a slight impact on downstream.

    Thanks for your input on this @yukkit. Any comment on whether you think it would be better to add a new enum to Distribution, or just similarly modify Distribution::HashPartitioned?

    Also wondering if it's better to implement a non-performant fix in the first pass (detect a nullable right table join key and use NestedLoopJoinExec in that case) vs. null-aware antijoin.

  6. msirek commented on Nov 17, 2023

    @msirek
    msirek
    on Nov 17, 2023
    ContributorAuthor

    Maybe adding an IS NOT false to the left_col = right_col predicate here, if right_col is nullable, would fix this:
    https://github.com/apache/arrow-datafusion/blob/05f0fa14628dab1cf27a3b15c3ae7d53cc01d1b2/datafusion/optimizer/src/decorrelate_predicate_subquery.rs#L264-L276

    Then optimizations could be added later.

  7. yukkit commented on Nov 18, 2023

    @yukkit
    yukkit
    on Nov 18, 2023
    Contributor

    Maybe adding an IS NOT false to the left_col = right_col predicate here, if right_col is nullable, would fix this:

    I completely agree with your @msirek idea. Currently, in datafusion, by adding IS NOT FALSE to the in_predicate, can transform an equijoin into a non-equijoin, thereby using NestedLoopJoinExec to execute and resolve the issue.

    In future development, when anti-join uncorrelated subqueries support the hash join algorithm, we can optimize it using broadcast null.

    @alamb, what are your thoughts on this?

  8. msirek commented on Nov 19, 2023

    @msirek
    msirek
    on Nov 19, 2023
    ContributorAuthor

    I'm working on the IS NOT FALSE fix.

  9. kumarUjjawal commented on Oct 24, 2026

    @kumarUjjawal
    kumarUjjawal
    on Oct 24, 2026
    Contributor
    +--------------------------+
| Utf8("qualified filter") |
+--------------------------+
+--------------------------+
0 row(s) fetched.
Elapsed 0.074 seconds.

0 row(s) fetched.
Elapsed 0.008 seconds.

0 row(s) fetched.
Elapsed 0.001 seconds.

+-------+
| count |
+-------+
| 1     |
+-------+
1 row(s) fetched.
Elapsed 0.003 seconds.

+-------+
| count |
+-------+
| 2     |
+-------+
1 row(s) fetched.
Elapsed 0.001 seconds.

+---+
| a |
+---+
+---+
0 row(s) fetched.
Elapsed 0.001 seconds.
  10. kumarUjjawal
    closed this as completedon Oct 24, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions