Blend & fuse aggregate functions can vary results on the same input

[philrz]

Given the synthetic input data below, the single value returned by the blend or fuse aggregate functions varies on repeated runs (field order is different). While nondeterminism among the order of output values (i.e., "rows") is well-documented and encountered in familiar systems like legacy SQL, having variation within the contents of a single-valued result would probably be unexpected by a user and tricky to defend as desirable behavior.

$ seq 1 240000 | awk '{ n = int(($1-1)/20000); printf "{\"k\":%d,\"x%d\":%d}\n", $1, n, n }' > disjoint.json &&
  super -version &&
  for i in $(seq 1 12); do super -s -c "from 'disjoint.json' | aggregate blend(this)"; done | sort -u &&
  for i in $(seq 1 12); do super -s -c "from 'disjoint.json' | aggregate fuse(this)"; done | sort -u

Version: v0.3.0-410-g888a49943

<{k:int64,x0?:int64,x1?:int64,x6?:int64,x7?:int64,x2?:int64,x3?:int64,x8?:int64,x9?:int64,x4?:int64,x10?:int64,x5?:int64,x11?:int64}>
<{k:int64,x0?:int64,x1?:int64,x6?:int64,x7?:int64,x3?:int64,x9?:int64,x2?:int64,x8?:int64,x4?:int64,x10?:int64,x5?:int64,x11?:int64}>
<{k:int64,x0?:int64,x1?:int64,x6?:int64,x7?:int64,x3?:int64,x9?:int64,x4?:int64,x10?:int64,x2?:int64,x8?:int64,x5?:int64,x11?:int64}>
<{k:int64,x1?:int64,x6?:int64,x7?:int64,x2?:int64,x3?:int64,x8?:int64,x9?:int64,x0?:int64,x4?:int64,x10?:int64,x5?:int64,x11?:int64}>
<{k:int64,x3?:int64,x9?:int64,x1?:int64,x7?:int64,x4?:int64,x10?:int64,x0?:int64,x6?:int64,x2?:int64,x8?:int64,x5?:int64,x11?:int64}>

<fusion({k:int64,x0:fusion(int64|none),x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x2:fusion(int64|none),x3:fusion(int64|none),x8:fusion(int64|none),x9:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x0:fusion(int64|none),x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x3:fusion(int64|none),x9:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x0:fusion(int64|none),x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x3:fusion(int64|none),x9:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x0:fusion(int64|none),x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x3:fusion(int64|none),x9:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x0:fusion(int64|none),x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x3:fusion(int64|none),x9:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x2:fusion(int64|none),x3:fusion(int64|none),x8:fusion(int64|none),x9:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x0:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x1:fusion(int64|none),x6:fusion(int64|none),x7:fusion(int64|none),x3:fusion(int64|none),x9:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x0:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>
<fusion({k:int64,x3:fusion(int64|none),x9:fusion(int64|none),x0:fusion(int64|none),x1:fusion(int64|none),x7:fusion(int64|none),x2:fusion(int64|none),x8:fusion(int64|none),x4:fusion(int64|none),x10:fusion(int64|none),x6:fusion(int64|none),x5:fusion(int64|none),x11:fusion(int64|none)})>

Details

Repro is with super commit 888a499.

I originally bumped into this symptom while trying to verify #7259 using the "logical equivalent" shown in the blend operator docs to work around the current lack of spill-to-disk functionality in the blend operator.

I used Claude to help craft the simplified repro above, and it also gave its take on root cause, so that's available in a Gist. Also noted is that the blend and fuse operators don't seem to have this problem, and that the blend and fuse aggregate functions don't have this problem if run with GOMAXPROCS=1 to disable parallel processing, e.g.,

$ GOMAXPROCS=1 super -s -c "from 'disjoint.json' | aggregate blend(this)"
<{k:int64,x0?:int64,x1?:int64,x2?:int64,x3?:int64,x4?:int64,x5?:int64,x6?:int64,x7?:int64,x8?:int64,x9?:int64,x10?:int64,x11?:int64}>

where we see the field order matches how they appeared top-to-bottom of the input data.

In discussing this issue with the team, it's been proposed that we may just make the aggregate functions deterministic in the short term by making them single-threaded and then design a more sophisticated approach later that is able to produce a deterministic result while still doing parallel processing.