Mate Czagany created FLINK-40959:
------------------------------------
Summary: Batch sort can emit unsorted rows if comparator
implementation is long enough to be split
Key: FLINK-40959
URL: https://issues.apache.org/jira/browse/FLINK-40959
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.1.3, 1.20.5, 2.2.1, 2.3.0, 2.4.0
Reporter: Mate Czagany
h2. Conditions:
* Async merge is enabled (table.exec.sort.async-merge-enabled)
* Intermediate merge happens while input is being sorted (At least
table.exec.sort.max-num-file-handles spill files before the input ends
* Generated RecordComparator method is longer than
table.generated-code.max-length. With the default of 4,000 it means about 6-7
fields, especially if they are nullable as those generate null-check blocks
h2. Root cause:
* A single RecordComparator will be used by all the steps (sort, spill, merge)
of BinaryExternalSorter.
** SortingThread uses it via QuickSort
** MergingThread uses it for intermediate merges
** SpillingThread uses it for the in-memory-merge path
* With async merge enabled, these threads will call compare() at the same time
* Code splitting makes the generated comparator impl stateful
** It will move the local variables from the compare() method to member fields
** This can be written and read by the different threads at any time
h2. Fix:
Make BinaryExternalSorter accept a Supplier<RecordComparator> instead of a
RecordComparator and create two separate instances:
* one for SortingThread
* one for MergingThread and SpillingThread, as these would never call the
compare() method at the same time
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
