[SPARK-22984] Fix incorrect bitmap copying and offset adjustment in GenerateUnsafeRowJoiner

[JoshRosen]

What changes were proposed in this pull request?

This PR fixes a longstanding correctness bug in GenerateUnsafeRowJoiner. This class was introduced in #7821 (July 2015 / Spark 1.5.0+) and is used to combine pairs of UnsafeRows in TungstenAggregationIterator, CartesianProductExec, and AppendColumns.

Bugs fixed by this patch

1. Incorrect combining of null-tracking bitmaps: when concatenating two UnsafeRows, the implementation "Concatenate the two bitsets together into a single one, taking padding into account". If one row has no columns then it has a bitset size of 0, but the code was incorrectly assuming that if the left row had a non-zero number of fields then the right row would also have at least one field, so it was copying invalid bytes and and treating them as part of the bitset. I'm not sure whether this bug was also present in the original implementation or whether it was introduced in [SPARK-9518] [SQL] cleanup generated UnsafeRowJoiner and fix bug #7892 (which fixed another bug in this code).
2. Incorrect updating of data offsets for null variable-length fields: after updating the bitsets and copying fixed-length and variable-length data, we need to perform adjustments to the offsets pointing the start of variable length fields's data. The existing code was conditionally adding a fixed offset to correct for the new length of the combined row, but it is unsafe to do this if the variable-length field has a null value: we always represent nulls by storing 0 in the fixed-length slot, but this code was incorrectly incrementing those values. This bug was present since the original version of GenerateUnsafeRowJoiner.

Why this bug remained latent for so long

The PR which introduced GenerateUnsafeRowJoiner features several randomized tests, including tests of the cases where one side of the join has no fields and where string-valued fields are null. However, the existing assertions were too weak to uncover this bug:

• If a null field has a non-zero value in its fixed-length data slot then this will not cause problems for field accesses because the null-tracking bitmap should still be correct and we will not try to use the incorrect offset for anything.
• If the null tracking bitmap is corrupted by joining against a row with no fields then the corruption occurs in field numbers past the actual field numbers contained in the row. Thus valid isNullAt() calls will not read the incorrectly-set bits.

The existing GenerateUnsafeRowJoinerSuite tests only exercised .get() and isNullAt(), but didn't actually check the UnsafeRows for bit-for-bit equality, preventing these bugs from failing assertions. It turns out that there was even a GenerateUnsafeRowJoinerBitsetSuite but it looks like it also didn't catch this problem because it only tested the bitsets in an end-to-end fashion by accessing them through the UnsafeRow interface instead of actually comparing the bitsets' bytes.

Impact of these bugs

• This bug will cause equals() and hashCode() to be incorrect for these rows, which will be problematic in caseGenerateUnsafeRowJoiner's results are used as join or grouping keys.
• Chained / repeated invocations of GenerateUnsafeRowJoiner may result in reads from invalid null bitmap positions causing fields to incorrectly become NULL (see the end-to-end example below).
  • It looks like this generally only happens in CartesianProductExec, which our query optimizer often avoids executing (usually we try to plan a BroadcastNestedLoopJoin instead).

End-to-end test case demonstrating the problem

The following query demonstrates how this bug may result in incorrect query results:

set spark.sql.autoBroadcastJoinThreshold=-1; -- Needed to trigger CartesianProductExec

create table a as select * from values 1;
create table b as select * from values 2;

SELECT
  t3.col1,
  t1.col1
FROM a t1
CROSS JOIN b t2
CROSS JOIN b t3

This should return (2, 1) but instead was returning (null, 1).

Column pruning ends up trimming off all columns from t2, so when t2 joins with another table this triggers the bitmap-copying bug. This incorrect bitmap is subsequently copied again when performing the final join, causing the final output to have an incorrectly-set null bit for the first field.

How was this patch tested?

Strengthened the assertions in existing tests in GenerateUnsafeRowJoinerSuite. Also verified that the end-to-end test case which uncovered this now passes.

[SparkQA]

Test build #85775 has finished for PR 20181 at commit 6e879f1.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gatorsmile]

retest this please

[SparkQA]

Test build #85782 has finished for PR 20181 at commit 6e879f1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

LGTM, merging to master/2.3/2.2!