[SPARK-52873][SQL] Further restrict when SHJ semi/anti join can ignore duplicate keys on the build side

[bersprockets]

What changes were proposed in this pull request?

After e861b0d, shuffle hash join for left semi/anti/existence will ignore duplicate keys if the join condition is empty or refers to the same parent attributes as the join keys. This PR proposes that duplicate keys should be ignored only when the join condition has these properties:

1. a subtree that is a semantic match to a build-side key, and/or
2. all attributes, outside of any subtree that is a semantic match to a build-side join key, should be from the stream-side.

Why are the changes needed?

e861b0d causes a correctness issue when a column is transformed in the build-side join keys and also transformed, but differently, in a join condition. As an example:

create or replace temp view data(a) as values
("xxxx1111"),
("yyyy2222");

create or replace temp view lookup(k) as values
("xxxx22"),
("xxxx33"),
("xxxx11");

-- this returns one row
select *
from data
left semi join lookup
on substring(a, 1, 4) = substring(k, 1, 4)
and substring(a, 1, 6) >= k;

-- this is the same query as above, but with a shuffle hash join hint, and returns no rows
select /*+ SHUFFLE_HASH(lookup) */ *
from data
left semi join lookup
on substring(a, 1, 4) = substring(k, 1, 4)
and substring(a, 1, 6) >= k;

When the join uses broadcast hash join, the hashrelation of lookup has the following key -> values:

Key xxxx:
  xxxx11
  xxxx33
  xxxx22

The join condition matches on the build side row with the value xxxx11.

When the join uses shuffle hash join, on the other hand, the hash relation of lookup has the following key -> values:

Key xxxx:
  xxxx22

Because the keys must be unique, an arbitrary row is chosen to represent the key, and that row does not match the join condition.

After 1f35577, a similar issue happens with integer keys:

create or replace temp view data(a) as values
(10000),
(30000);

create or replace temp view lookup(k) as values
(1000),
(1001),
(1002),
(1003),
(1004);

-- this query returns one row
select * from data left semi join lookup on a/10000 = cast(k/1000 as int) and k >=  a/10 + 3;

-- this is the same query as above, but with a shuffle hash join hint, and returns no rows
select /*+ SHUFFLE_HASH(lookup) */ * from data left semi join lookup on a/10000 = cast(k/1000 as int) and k >=  a/10 + 3;

Does this PR introduce any user-facing change?

No, except for fixing the correctness issue.

How was this patch tested?

Modified an existing unit test.

Was this patch authored or co-authored using generative AI tooling?

No.

[YuzhouSun]

Thanks for fixing this issue!

Instead of disable ignoreDuplicatedKey in this case, Is it possible to relax the requirement to that the references in conditions are all in streamedOutput ++ buildKeysThatAreAttributes ? E.g. diff:

-      val streamedOutputAndBuildKeys = AttributeSet(streamedOutput ++ buildKeys)
+      val attrBuildKeys = buildKeys.filter(_.isInstanceOf[Attribute])
+      val streamedOutputAndBuildKeys = AttributeSet(streamedOutput ++ attrBuildKeys)”

The current master branch allows buildKeys’ references in the condition (AttributeSet extracts references), while this diff limits it to streamedOutput and buildKeys that are Attributes

[peter-toth]

Or maybe something like:

      val streamedOutputSet = AttributeSet(streamedOutput)
      val buildKeysSet = ExpressionSet(buildKeys)
      condition.forall { c =>
        var valid = true
        c.transformDownWithPruning(c => valid && !buildKeysSet.contains(c.asInstanceOf[Expression])) {
          case a: Attribute if !streamedOutputSet.contains(a) =>
            valid = false
            a
        }
        valid
      }

to allow full expressions from buildKeys to appear in condition.

[bersprockets]

@peter-toth

It's hard for me to grok that traversal, but I think the gist is:

• Any subtree of condition that is not a semantic match to a build-side join key must be checked for naughty attributes.
• Set valid to false when we hit a naughty attribute in such a subtree.
• We can short circuit the traversal once valid is set to false.

I wish there was another way to express that, but I don't know how else to skip selected subtrees of condition.

[peter-toth]

Yes, exactly. But we can write a recursive function if that's easier to grasp.

[bersprockets]

Your's is nice and compact, but I'll try an explicit traversal and see how sprawling it gets.

[peter-toth]

This is a good catch!
Fix looks good to me, but maybe we can improve it a bit.

[peter-toth]

This doesn't need to be a top level function, but can be nested under validCondForIgnoreDupKey() and you don't need to pass in buildKeysSet and streamedOutputAttrs that way, but this is just a nit.

[bersprockets]

I will fix that.