[SPARK-31606][SQL] Reduce the perf regression of vectorized parquet reader caused by datetime rebase

[cloud-fan]

What changes were proposed in this pull request?

Push the rebase logic to the lower level of the parquet vectorized reader, to make the final code more vectorization-friendly.

Why are the changes needed?

Parquet vectorized reader is carefully implemented, to make it more likely to be vectorized by the JVM. However, the newly added datetime rebase degrade the performance a lot, as it breaks vectorization, even if the datetime values don't need to rebase (this is very likely as dates before 1582 is rare).

Does this PR introduce any user-facing change?

no

How was this patch tested?

Run part of the DateTimeRebaseBenchmark locally. The results:
before this patch

[info] Load dates from parquet: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] after 1582, vec on, rebase off 2677 2838 142 37.4 26.8 1.0X
[info] after 1582, vec on, rebase on 3828 4331 805 26.1 38.3 0.7X
[info] before 1582, vec on, rebase off 2903 2926 34 34.4 29.0 0.9X
[info] before 1582, vec on, rebase on 4163 4197 38 24.0 41.6 0.6X
[info] Load timestamps from parquet: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] after 1900, vec on, rebase off 3537 3627 104 28.3 35.4 1.0X
[info] after 1900, vec on, rebase on 6891 7010 105 14.5 68.9 0.5X
[info] before 1900, vec on, rebase off 3692 3770 72 27.1 36.9 1.0X
[info] before 1900, vec on, rebase on 7588 7610 30 13.2 75.9 0.5X

After this patch

[info] Load dates from parquet: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] after 1582, vec on, rebase off 2758 2944 197 36.3 27.6 1.0X
[info] after 1582, vec on, rebase on 2908 2966 51 34.4 29.1 0.9X
[info] before 1582, vec on, rebase off 2840 2878 37 35.2 28.4 1.0X
[info] before 1582, vec on, rebase on 3407 3433 24 29.4 34.1 0.8X
[info] Load timestamps from parquet: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] after 1900, vec on, rebase off 3861 4003 139 25.9 38.6 1.0X
[info] after 1900, vec on, rebase on 4194 4283 77 23.8 41.9 0.9X
[info] before 1900, vec on, rebase off 3849 3937 79 26.0 38.5 1.0X
[info] before 1900, vec on, rebase on 7512 7546 55 13.3 75.1 0.5X

Date type is 30% faster if the values don't need to rebase, 20% faster if need to rebase.
Timestamp type is 60% faster if the values don't need to rebase, no difference if need to rebase.

[cloud-fan]

cc @MaxGekk@HyukjinKwon@kiszk@rednaxelafx

[kiszk]

While I have not understood this logic yet, this code sees the result only at the last iteration.
May it be rebase |= ... or something?

[cloud-fan]

good catch! I rerun the benchmark and no big difference. Because the benchmark data are either all values need to rebase, or none.

[kiszk]

same as at line 136.

[MaxGekk]

@cloud-fan Please, review and merge cloud-fan#15

[SparkQA]

Test build #122068 has finished for PR 28406 at commit 056e7f0.

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

[SparkQA]

Test build #122076 has finished for PR 28406 at commit 4d15a49.

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

[SparkQA]

Test build #122137 has finished for PR 28406 at commit 5a3009b.

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

[MaxGekk]

The byte code of the loop is:

 LINENUMBER 136 L6
ILOAD 6
ALOAD 5
ALOAD 5
INVOKEVIRTUAL java/nio/ByteBuffer.position ()I
ILOAD 7
BIPUSH 8
IMUL
IADD
INVOKEVIRTUAL java/nio/ByteBuffer.getLong (I)J
INVOKESTATIC org/apache/spark/sql/catalyst/util/RebaseDateTime.lastSwitchJulianTs ()J
LCMP
IFGE L7
ICONST_1
GOTO L8

We could avoid mul like

intpos = buffer.position();
intendPos = pos + total * 8;
longthreshold = RebaseDateTime.lastSwitchJulianTs();
while (pos < endPos) {
rebase |= buffer.getLong(pos) < threshold;
pos += 8;
}

Would it be faster?

[cloud-fan]

I tried it and the perf has no difference.

[MaxGekk]

Is it impossible to optimize the case too?

[cloud-fan]

I didn't optimize this case because the no-rebase code path looks not very fast. It has a if-else in the loop.

[cloud-fan]

The general idea is to add an extra loop to check if we need to rebase or not, and it's only worthwhile if the no-rebase code path is much faster than the rebase code path.

[HyukjinKwon]

Merged to master and branch-3.0.