fix(stdlib): sort limit skips chunks with no rows (#5211)

The sort limit code did not skip chunks with no rows which caused it to
attempt to compare invalid chunks with each other. This skips those rows
but still ensures that the schema is created for them.

This is similar to the same problem that was identified in #4701. That
PR only addressed when the last or only chunk was zero while this one
ensures that the aggregation step also handles this situation properly.

Author: jsternberg

libflux/go/libflux/buildinfo.gen.go

@@ -576,7 +576,7 @@ var sourceHashes = map[string]string{
 	"stdlib/universe/simple_max_test.flux":                                                        "a63b3f530e4d81451e3c71a1abeea50cff02e743c9313ab3ffd5bc3b3ce9ad2e",
 	"stdlib/universe/skew_test.flux":                                                              "7782d41c563c77ba9f4176fa1b5f4f6107e418b7ea301e4896398dbcb514315a",
 	"stdlib/universe/sort2_test.flux":                                                             "1d2043c0d0b38abb8dc61fc1baa6d6052fae63fea55cc6e67fd1600911513bdb",
-	"stdlib/universe/sort_limit_test.flux":                                                        "5ae162efb25cb25c3f18b871ab30bfd5e827acb329fd92d8926011658df8bd82",
+	"stdlib/universe/sort_limit_test.flux":                                                        "80a7a25384b86db401ffd6fd2ea58c8e4b0646f522fa121e92a00793c0a28039",
 	"stdlib/universe/sort_rules_test.flux":                                                        "0770ae42e99b04167ca5bef8340a310b224baf1ba1928997273de9663b64684a",
 	"stdlib/universe/sort_test.flux":                                                              "f69ebb5972762078e759af3c1cd3d852431a569dce74f3c379709c9e174bfa31",
 	"stdlib/universe/spread_test.flux":                                                            "1ddf25e4d86b6365da254229fc8f77bd838c24a79e6a08c9c4c50330ace0a6a3",

stdlib/universe/sort.go

@@ -339,7 +339,7 @@ func (s *sortTableMergeHeap) ValueLen() int {
 
 func (s *sortTableMergeHeap) Table(limit int, mem memory.Allocator) (flux.Table, error) {
 	if s.ValueLen() == 0 {
-		// Degenerate case where there are no rows to merge sort
+		// Degenerate case where there are no rows to merge sort.
 		for len(s.items) > 0 {
 			s.Pop()
 		}

stdlib/universe/sort_limit.go

@@ -80,6 +80,13 @@ func (s *sortLimitTransformation) Aggregate(chunk table.Chunk, state interface{}
 
 	}
 
+	// If the chunk is empty, ignore this chunk.
+	// We still return the merge heap though as we still need to recognize that
+	// the group key/schema exist.
+	if chunk.Len() == 0 {
+		return mh, true, nil
+	}
+
 	if err := s.appendChunk(mh, chunk, mem); err != nil {
 		return nil, false, err
 	}

stdlib/universe/sort_limit_test.flux

@@ -170,3 +170,29 @@ testcase sort_limit_multi_successor {
 
     testing.diff(got: got, want: want)
 }
+
+testcase sort_limit_empty_chunk {
+    got =
+        array.from(
+            rows: [
+                {_time: 2022-01-11T00:00:00Z, t0: "aa", _value: 12.0},
+                {_time: 2022-01-11T00:00:00Z, t0: "ab", _value: 10.0},
+                {_time: 2022-01-11T00:00:00Z, t0: "ba", _value: 18.0},
+                {_time: 2022-01-11T00:00:00Z, t0: "bb", _value: 4.0},
+            ],
+        )
+            |> group(columns: ["t0"])
+            |> filter(fn: (r) => r.t0 =~ /b/, onEmpty: "keep")
+            |> group()
+            |> top(n: 2)
+
+    want =
+        array.from(
+            rows: [
+                {_time: 2022-01-11T00:00:00Z, t0: "ba", _value: 18.0},
+                {_time: 2022-01-11T00:00:00Z, t0: "ab", _value: 10.0},
+            ],
+        )
+
+    testing.diff(got: got, want: want)
+}