The collect aggregator is receiving null values for some reason in ConsumeAsPartial().
Let's investigate this and either eliminate the call or document it better.
Recreate with
collect(x) by key with -limit 1 in.zson
where in.zson is
{key:"a",x:1(int32)}
{key:"a",x:-1(int32)}
{key:"b",x:2(int32)}
{key:"b",x:1(int32)}
{key:"a",x:8(int32)}
{key:"b",x:1(int32)}
{key:"a"}
{key:"a"}
{key:"a"}