[VL] Result mismatch when BHJ with left semi join

[NEUpanning]

Backend

VL (Velox)

Bug description

We are running into an issue where the probe table only contains a subset of rows that meet the build table, but the result of gluten BHJ is equal to the number of output rows of probe table. The join condition does not filter any rows. Through the velox plan we can see the join condition filter is pushed down to table scan and it also does not filter any rows.

vanilla plan:

gluten plan:

velox plan:

-- HashJoin[3][LEFT SEMI (FILTER) n0_14=n2_0] -> n0_0:BIGINT, n0_1:VARCHAR, n0_2:BIGINT, n0_3:BIGINT, n0_4:VARCHAR, n0_5:INTEGER, n0_6:INTEGER, n0_7:BIGINT, n0_8:VARCHAR, n0_9:BIGINT, n0_10:BIGINT, n0_11:BIGINT, n0_12:BIGINT, n0_13:INTEGER, n0_14:BIGINT, n0_15:BIGINT, n0_16:INTEGER, n0_17:VARCHAR, n0_18:VARCHAR, n0_19:VARCHAR, n0_20:VARCHAR, n0_21:VARCHAR, n0_22:VARCHAR
         Output: 135 rows (295.33KB, 1 batches), Cpu time: 129.13us, Wall time: 139.36us, Blocked wall time: 0ns, Peak memory: 68.00KB, Memory allocations: 2, CPU breakdown: B/I/O/F (85.06us/612ns/39.82us/3.63us)
         HashBuild: Input: 2 rows (32B, 1 batches), Output: 0 rows (0B, 0 batches), Cpu time: 13.00us, Wall time: 15.54us, Blocked wall time: 0ns, Peak memory: 68.00KB, Memory allocations: 2, Threads: 1, CPU breakdown: B/I/O/F (10.05us/0ns/1.50us/1.45us)
            distinctKey0                 sum: 3, count: 1, min: 3, max: 3
            hashtable.buildWallNanos     sum: 84.66us, count: 1, min: 84.66us, max: 84.66us
            hashtable.capacity           sum: 3, count: 1, min: 3, max: 3
            hashtable.numDistinct        sum: 2, count: 1, min: 2, max: 2
            hashtable.numRehashes        sum: 1, count: 1, min: 1, max: 1
            queuedWallNanos              sum: 0ns, count: 1, min: 0ns, max: 0ns
            rangeKey0                    sum: 3, count: 1, min: 3, max: 3
            runningAddInputWallNanos     sum: 0ns, count: 1, min: 0ns, max: 0ns
            runningFinishWallNanos       sum: 2.00us, count: 1, min: 2.00us, max: 2.00us
            runningGetOutputWallNanos    sum: 2.08us, count: 1, min: 2.08us, max: 2.08us
         HashProbe: Input: 135 rows (295.33KB, 1 batches), Output: 135 rows (295.33KB, 1 batches), Cpu time: 116.13us, Wall time: 123.82us, Blocked wall time: 0ns, Peak memory: 0B, Memory allocations: 0, Threads: 1, CPU breakdown: B/I/O/F (75.01us/612ns/38.32us/2.19us)
            dynamicFiltersProduced           sum: 1, count: 1, min: 1, max: 1
            queuedWallNanos                  sum: 1.00us, count: 1, min: 1.00us, max: 1.00us
            replacedWithDynamicFilterRows    sum: 135, count: 1, min: 135, max: 135
            runningAddInputWallNanos         sum: 885ns, count: 1, min: 885ns, max: 885ns
            runningFinishWallNanos           sum: 3.65us, count: 1, min: 3.65us, max: 3.65us
            runningGetOutputWallNanos        sum: 39.83us, count: 1, min: 39.83us, max: 39.83us

Spark version

Spark-3.5.x

Spark configurations

No response

System information

No response

Relevant logs

[NEUpanning]

cc @kecookier

[Yohahaha]

have you try force TableScan#canAddDynamicFilter return false?

[Yohahaha]

replacedWithDynamicFilterRows this metrics should be the root cause.

[JkSelf]

@NEUpanning
I tested the following code on the main branch, but it did not reproduce the exception mentioned above. Could you help by providing a unit test that reproduces the issue? Thanks.

import spark.implicits._
  val employees = Seq(
    (1, "Alice", 1),
    (2, "Bob", 2),
    (3, "Charlie", 3),
    (4, "David", 4)
  ).toDF("emp_id", "name", "dept_id")

  val departments = Seq(
    (1, "HR"),
    (2, "Engineering"),
    (3, "Marketing")
  ).toDF("dept_id", "dept_name")

  // filter condition
  val filteredDepartments = departments.filter($"dept_name" === "Engineering")

  // Left Semi Join
  val result = employees.join(filteredDepartments, employees("dept_id") === filteredDepartments("dept_id"), "leftsemi")
  result.show()

[NEUpanning]

@Yohahaha

have you try force TableScan#canAddDynamicFilter return false?

Yes, the results match vanilla Spark, so dynamic filter push down could be the root cause.
The output rows metric of HashProbe op without any dynamic filters push down is correct:

HashProbe: Input: 135 rows (341.73KB, 1 batches), Output: 131 rows (331.61KB, 1 batches), Cpu time: 822.08us, Wall time: 833.05us, Blocked wall time: 0ns, Peak memory: 50.25KB, Memory allocations: 26, Threads: 1, CPU breakdown: B/I/O/F (172.31us/3.27us/642.43us/4.08us)
        queuedWallNanos              sum: 1.00us, count: 1, min: 1.00us, max: 1.00us
        runningAddInputWallNanos     sum: 3.58us, count: 1, min: 3.58us, max: 3.58us
        runningFinishWallNanos       sum: 5.33us, count: 1, min: 5.33us, max: 5.33us
        runningGetOutputWallNanos    sum: 647.75us, count: 1, min: 647.75us, max: 647.75us

[NEUpanning]

@JkSelf

I tested the following code on the main branch, but it did not reproduce the exception mentioned above. Could you help by providing a unit test that reproduces the issue? Thanks.

I cannot reproduce this case with a unit test for now. I am working to find the reason for the issue. If I make any progress, I will keep you updated.

[Yohahaha]

@Yohahaha

have you try force TableScan#canAddDynamicFilter return false?

Yes, the results match vanilla Spark, so dynamic filter push down could be the root cause.

table scan op does not filtering out rows with pushed dynamic filter from join op on some cases, then join direct return scan's result.

[NEUpanning]

table scan op does not filtering out rows with pushed dynamic filter from join op on some cases, then join direct return scan's result.

@Yohahaha Do you mean this is a known issue?

[Yohahaha]

table scan op does not filtering out rows with pushed dynamic filter from join op on some cases, then join direct return scan's result.

@Yohahaha Do you mean this is a known issue?

it's the root cause of your issue, you can try follow and fix it.

[NEUpanning]

@JkSelf
I found the root cause of this issue. In our internal Velox branch, we support an internal data format reader that does not allow filter push down. However, TableScan#canAddDynamicFilter always returns true regardless of whether the format reader supports this feature. In this case, the filters were pushed down to the format reader and did not apply to rows. I've changed TableScan#canAddDynamicFilter to consider if the reader supports filter push down to resolve this issue.

Do you think we should open a PR in Velox, although all format readers support this feature in Velox now?

[JkSelf]

@NEUpanning Thanks for your information. It seems this is a bug in velox. And please help to open a PR in velox. Thanks.

[NEUpanning]

After implementing the approach in link , this issue has been resolved. Thanks for your help @JkSelf @Yohahaha