sql: call DrainMeta and Next from a single goroutine

[asubiotto]

The inner-join logictest fails under stress with the fakedist configuration. Repro:

make stress PKG=./pkg/sql/logictest TESTS="TestLogic//inner-join" TESTFLAGS="-v -config=fakedist"

Looks like sometimes rows are being only partially returned:

                testdata/logic_test/inner-join:69: SELECT a, b, c FROM abc_decimal WHERE EXISTS (SELECT * FROM def_decimal WHERE a::string=d::string)
                expected:
                    1     1     1
                    1     1     1
                    1.0   1.0   1.0
                    1.00  1.00  1.00
                but found (query options: "rowsort" -> ignore the following ordering of rows) :
                    1.00  1.00  1.00

[blathers-crl]

Hi @asubiotto, please add a C-ategory label to your issue. Check out the label system docs.

While you're here, please consider adding an A- label to help keep our repository tidy.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is otan.}

[knz]

another occurrence here: https://teamcity.cockroachdb.com/buildConfiguration/Cockroach_UnitTests/1939915?

[asubiotto]

@yuzefovich based on a git bisect it looks like #48668 is the first bad commit. Would you be able to look into this?

[yuzefovich]

Sure.

[yuzefovich]

This issue is quite puzzling. I've been playing around with this slightly modified logic test:

# LogicTest: 5node

statement ok
CREATE TABLE abc_decimal (i INT PRIMARY KEY, a DECIMAL, b DECIMAL, c DECIMAL);
INSERT INTO abc_decimal VALUES (1, 1, 1, 1), (2, 1, 1, 1), (3, 1.0, 1.0, 1.0), (4, 1.00, 1.00, 1.00);
ALTER TABLE abc_decimal SPLIT AT VALUES (2), (3), (4);
ALTER TABLE abc_decimal EXPERIMENTAL_RELOCATE SELECT ARRAY[i+1], i FROM generate_series(1, 4) as g(i)

statement ok
CREATE TABLE def_decimal (i INT PRIMARY KEY, d DECIMAL, e DECIMAL, f DECIMAL);
INSERT INTO def_decimal VALUES (1, 1, 1, 1), (2, 1.0, 1.0, 1.0), (3, 1.00, 1.00, 1.00);
ALTER TABLE def_decimal SPLIT AT VALUES (2), (3);
ALTER TABLE def_decimal EXPERIMENTAL_RELOCATE SELECT ARRAY[i+1], i FROM generate_series(1, 3) as g(i)

query RRR rowsort
SELECT a, b, c FROM abc_decimal WHERE EXISTS (SELECT * FROM def_decimal WHERE a::string=d::string)
----
1     1     1
1     1     1
1.0   1.0   1.0
1.00  1.00  1.00

and what's interesting is that the failure seems to occur only if both tables do not have any data on the gateway node (that's the reason for i+1 when relocating).

I feel like the problem might be with parallel unordered synchronizer, and that PR to short-circuit the hash joiner just exposed it (namely, made it more likely to occur on this logic test). I think so because if it was the hash joiner to blame, then I'd expect the failure to occur every time, but it occurs quite rarely and only under stress which indicates that there some race going on.

[yuzefovich]

I stared at it for another hour or two but didn't make any progress.

Another logic test that be used to repro is

# LogicTest: fakedist

statement ok
CREATE TABLE twocolumn (x INT, y INT PRIMARY KEY);
INSERT INTO twocolumn(x, y) VALUES (44,51), (45,45)

query IIIIII rowsort
SELECT a.x AS s, b.x, c.x, a.y, b.y, c.y FROM (twocolumn AS a JOIN twocolumn AS b USING(x) JOIN twocolumn AS c USING(x))
----
44  44  44  51  51  51
45  45  45  45  45  45

and I wasn't successful in converting it into using 5node config with ranges relocation.

My theory still remains the same - that somehow parallel unordered synchronizer is to blame. @asubiotto do you mind taking a look?

[asubiotto]

Sure.

[jordanlewis]

------- Stdout: -------
=== RUN   TestLogic/fakedist/subquery_correlated
=== PAUSE TestLogic/fakedist/subquery_correlated
=== CONT  TestLogic/fakedist/subquery_correlated
        --- FAIL: TestLogic/fakedist/subquery_correlated (5.03s)
            logic.go:1952: 
                 
                testdata/logic_test/subquery_correlated:84: SELECT *
                FROM c
                WHERE
                    'CA' IN (SELECT ship FROM o WHERE o.c_id=c.c_id)
                    AND 'TX' NOT IN (SELECT ship FROM o WHERE o.c_id=c.c_id);
                expected:
                    1  CA
                    
                but found (query options: "rowsort" -> ignore the following ordering of rows) :
                    1  CA
                    2  TX
                    
            logic.go:2530: 
                testdata/logic_test/subquery_correlated:93: error while processing
            logic.go:2530: testdata/logic_test/subquery_correlated:93: too many errors encountered, skipping the rest of the input

I just got this failure on a CI run. Is it related?

[yuzefovich]

Yeah, I think it's the same underlying problem.

[andreimatei]

In case it helps anything, I got

=== CONT  TestLogic/fakedist/join
rewrote:
CREATE TABLE abc (a INT8, b INT8, c INT8, PRIMARY KEY (a, c), FAMILY (a), FAMILY (b, c));
INSERT INTO abc VALUES (1, 1, 2), (2, 1, 1), (2, NULL, 2);


        --- FAIL: TestLogic/fakedist/join (45.88s)
            logic.go:1952: 
                 
                testdata/logic_test/join:429: SELECT * FROM (twocolumn AS a JOIN twocolumn AS b USING(x) JOIN twocolumn AS c USING(x)) ORDER BY x LIMIT 1
                expected:
                    x  y  y  y
                    42 53 53 53
                    
                but found (query options: "colnames") :
                    x   y   y   y
                    44  51  51  51

[asubiotto]

Forcing the use of a serial synchronizer (slower but simpler logic):

                } else {
-                       if opt == flowinfra.FuseAggressively {
+                       if true || opt == flowinfra.FuseAggressively {
                                op = colexec.NewSerialUnorderedSynchronizer(inputStreamOps, input.ColumnTypes)
                        } else {
                                op = colexec.NewParallelUnorderedSynchronizer(inputStreamOps, input.ColumnTypes, s.waitGroup)

still reproduces this problem so I don't think this is a parallel unordered synchronizer problem, and as soon as I comment out the short-circuiting logic in the hash joiner, this problem doesn't reproduce. Another change that makes this test pass is exhausting the left side:

                                if hj.spec.joinType == sqlbase.JoinType_INNER ||
                                        hj.spec.joinType == sqlbase.JoinType_RIGHT_OUTER ||
                                        hj.spec.joinType == sqlbase.JoinType_LEFT_SEMI {
+                                       for {
+                                               b := hj.inputOne.Next(ctx)
+                                               if b.Length() == 0 {
+                                                       break
+                                               }
+                                       }

Which indicates that not calling Next on the left input is having undesired side effects (which should not happen). @yuzefovich does anything come to mind? Will keep on digging.

[yuzefovich]

Hm, since we started populating batches of projecting operators upfront, we're able to short-circuit execution of operators when they receive zero-length batch from its inputs, however, in all previous scenarios we would always call Next on the inputs at least once, and this change in the hash joiner seems to be the first precedent of not doing so, but I can't think of anything that would break because of it - Init is still called as required by the interface.

I'm thinking that #49147 that shows a data race when cleaning up disk spilling infrastructure might be the root cause of this issue as well, but I haven't looked too closely into that issue.

[asubiotto]

I'm thinking that #49147 that shows a data race when cleaning up disk spilling infrastructure might be the root cause of this issue as well, but I haven't looked too closely into that issue.

I took a look at that and fixed the race and it doesn't make a difference for the erroneous test output.

[asubiotto]

Going to merge the single call to Next to fix the flake while I investigate further

[asubiotto]

Got to the bottom of this. This is the EXPLAIN (DISTSQL) plan for the repro shared in #48785 (comment)

In the repro, it looks like HashJoiner/9 only reads 1 row from the left side but 2 rows from the right side. It therefore only emits one result tuple when it should be emitting 2. This is cause by a missing row on the left side.

This row can be traced back to the inbox pointed out on the diagram. The row is correctly sent over the wire but DrainMeta is called before Next, so the Inbox returns zero rows. This is correct Inbox behavior, the problem is that DrainMeta is called prematurely.

HashJoiner/7 short circuits because it gets zero rows from its right side. The downstream outbox gets a zero length batch and therefore calls DrainMeta on its metadata sources, which includes the Inbox. This Inbox is in the outbox's input graph, but the problem is that we assume that the input graph is drained of rows when the outbox receives a zero-length batch. This is not the case with short-circuiting hash joiners, since the left input could be a hash router that still needs to send rows to another node. Calling Next once on the left side fixes this issue because it wait until the HashRouter has finished pushing all rows.

To properly fix this, I think we need to make the outbox call DrainMeta on its upstream HashRouter. The HashRouter would only cancel operation once all outputs have called DrainMeta.

[yuzefovich]

Nice find!

[knz]

That is a very nice find indeed.