kv: scan doesn't return row added in the same txn

[RaduBerinde]

I've been experimenting with TPCC and the new foreign key path. When I made an improvement that runs multiple FK checks in parallel, I started getting occasional violation errors. I was able to reproduce this locally and capture some traces.

Repro steps:

• set up a local cluster using the origin/fk-violation-repro branch; it feels like the error happens only when the server is under heavy load so I used GOMAXPROCS=4.

GOMAXPROCS=4 ./cockroach start --insecure --logtostderr --max-sql-memory 4GB

• import TPCC fixtures with 100 warehouses

bin/workload fixtures import tpcc --warehouses 100

• run

bin/workload run tpcc --warehouses=100 --duration=10m --split --wait=false --method=simple --mix=newOrder=100

The error is always the same:

Error: error in newOrder: ERROR: insert on table "order_line" violates foreign key constraint "fk_ol_w_id_ref_order" (SQLSTATE 23503)

This is the parent row in the order table which should have been inserted in the same transaction.

I set up a local zipkin and below is the span for the join reader. You can find the relevant span by looking for the omfg tag ("annotation query" in zipkin UI). All 9 rows reference the same order row (57, 2, 3029) so we scan 1 span. The KV request returns no rows (0 loop iterations).

This is the corresponding distsender span:

Earlier spans in the transaction confirm that we did in fact insert a row with these values.

[nvanbenschoten]

Could you expand on

made an improvement that runs multiple FK checks in parallel

Are you running multiple KV checks in the same batch or are you issuing multiple batches in parallel?

[RaduBerinde]

Different batches, in parallel. When the batches are run serially I don't see the issue. They are checking different parent tables so it's a bit puzzling.

[irfansharif]

diff --git i/pkg/kv/txn_coord_sender.go w/pkg/kv/txn_coord_sender.go
index 5a14058eda..44beb1bb4d 100644
--- i/pkg/kv/txn_coord_sender.go
+++ w/pkg/kv/txn_coord_sender.go
@@ -453,6 +453,7 @@ func NewTxnCoordSenderFactory(
 	if tcf.heartbeatInterval == 0 {
 		tcf.heartbeatInterval = base.DefaultTxnHeartbeatInterval
 	}
+	tcf.heartbeatInterval = 5 * base.DefaultTxnHeartbeatInterval
 	if tcf.metrics == (TxnMetrics{}) {
 		tcf.metrics = MakeTxnMetrics(metric.TestSampleInterval)
 	}

Discovered this yesterday. Increasing the heartbeat interval to trigger more aborted transactions drastically reduces how long it takes for the consistency error to manifest (from taking upwards of 60s to 7-9s, consistently). This points to our current flow cancellation procedure: when the root txn is aborted/abandoned, we set it up to not return results to the client so as to not miss seeing our own writes. Suspiciously we avoid doing this if the flow is local and is using the root txn, both of which apply here.

[RaduBerinde]

CC @andreimatei

[andreimatei]

Suspiciously we avoid doing this if the flow is local and is using the root txn

The local can't miss to see its own writes; when using the RootTxn, the transaction cleanup process (that cleans up intents, which is how one might miss to see its own writes) is synchronized with the sending and receiving of batches, and so a client is not supposed to get results back after this cleanup. At least in theory.

@irfansharif, are you looking at this? Would you like me to take over? And if so, are Radu's instructions still the best repro?

[irfansharif]

Banged my head over this all morning again to no avail, happy to have you take over. The best repro is with the patch I added above. The missing key error only ever occurs after a flurry of aborted transactions now to be retried. Naively this diff seemed awfully suspect, though I'm not familiar enough with pkg/sql to understand the reasoning as stated here.

[irfansharif]

I was also going to pull on this thread next, though our usage of planner seems fine despite parallelizing the two FK checks.

[RaduBerinde]

I think that comment refers to executing different RETURNING NOTHING statements in parallel which we don't do anymore.

[jordanlewis]

Yeah, that was about RETURNING NOTHING.

[jordanlewis]

I think the root/leaf stuff might be a red herring. None of the queries in question get run on remote nodes. The main query is a mutation, which can't be distributed. The postqueries contain nodes that prevent distribution (errorIfRowsNode, scanBufferNode).

[andreimatei]

Will look at this tomorrow. The money in this cubicle is on some variation of
#25329 and #22615

[andreimatei]

So I think this is simply #25329; one can't have concurrent uses of a root transaction. And so I understand that the distsql union processor is probably broken.

I think there's also funky something funky here:

cockroach/pkg/sql/distsql_running.go

Line 543 in 2109039

if r.txn != nil && r.txn.Type() == client.LeafTxn {

That check was changed to only affect leaf transactions, but I think it's useless for leaves.

More tomorrow.

[RaduBerinde]

Weird.. I assume you can have multiple scans in parallel in a Leaf txn (or distsql would be badly broken)? We could maybe set up a Leaf txn even if it's all local (?)

[andreimatei]

Yeah, Leaf transactions are fine. That's part of their point. But you can't do writes in leaf transactions.

So, we have a couple of requirements:

1. mutations need to use the root
2. roots don't support concurrent use
3. processors that might execute concurrently with others need to use leaves

Concurrency is introduced by processors inputs with more than one stream, which require synchronizers (or simply a RowChannel if for an "unordered synchronizer"). We're helped, however, by the fact that a mutation forces everything to be planned on the gateway. When everything is forced on the gateway, only a union can introduce a synchronizer (I hope).

So the plan I've discussed with @jordanlewis is the following:

1. Disallow mutations under unions. This means that everything under a union can use a leaf.
2. Have different processors in a flow use different transactions: everything from the top of the plan above a synchronizer will use the Root (which satisfies the mutations), and everything below will use the leaf (which satisfies the need for concurrency for unions, or generally for other TableReaders when combined with remote flows).

[RaduBerinde]

To clarify requirement 2: is it allowed to have read-only queries on leaves concurrently with a root mutation?

[andreimatei]

is it allowed to have read-only queries on leaves concurrently with a root mutation?

yes

[andreimatei]

I've worked on this and made progress. Have yet to deal with the columnar execution world, but @jordanlewis helped me figure out how stuff works there.

@RaduBerinde , do you want to disallow mutations under a union? You seem like the right guy for the job. And also, maybe you can think about whether there's other cases where one might end up with a mutation under a synchronizer (which synchronizer introduces concurrent execution).

[RaduBerinde]

Yeah, very easy to do from opt. Will think about other cases.

[andreimatei]

I've made good progress on this, but I've now hit a something I wasn't prepared for: I naively thought that all vectorizedFlows have exactly one processor - a Materializer - under which a bunch of other Operators lie. And so I thought that, by starting from that root Materializer, I can visit the tree of Operators/OpNodes and tell each one what transaction they can run in.
But lo and behold, you can also have vectorizedFlows with no processors at all. You can have outboxes and routers under which there's a line of Operators too. These roots are represented in the Flow as Startables - added here.
So now I'm thinking how best to propagate the transaction through these guys. Should I pass the transaction through Startable.Start() and let each such columnar startable pass it along to its child Operators? Or should I make the vectorized flow setup process keep track of what operators need the txn regardless of where they'll end up in the final flow? All operators are created by NewColOperator, so I could have that guy maintain a list of operators that need the transaction...

[andreimatei]

So here's where we are on this:
Originally I thought that we have the following problems within this issue:
a) concurrent mutations can end up committing unintended data. That's what Radu fixed in #40975
b) concurrent reads on the gateway can miss seeing previous writes from the same txn. Among other consequences, this can cause the surprising failure to validate FKs that was reported here. I have #41102 open to fix that. However, such reads would be retried, so... that somewhat limits the consequences.

But I've recently realized that there's also another problem, worse than b).
c) possible write skew. Described in #41173. This has to do with races with operations done by the RootTxn (on the gateway) and remote flows. So only queries that were not fully planned on the gateway were affected.

Unfortunately, my fix for b) in #41102 I believe actually makes c) more likely. Because the PR introduces uses of the LeafTxn on the gateway too, now queries that execute fully on the gateway but are not completely "fused" are also susceptible. This means queries with a union in them. And also more weird queries with an outbox on the gateway; these would be queries that are distributed, but previously the part of the gateway flow leading into the outbox would not contribute to the problem.

In other words, I've lied when I answered @RaduBerinde's question:

is it allowed to have read-only queries on leaves concurrently with a root mutation?

I said yes. I now believe the answer is no. Except the question was implicitly scoped with concurrent uses of the root and leaves on the gateway. I believe even (existing) uses of remote leaves are unsafe.