distsql: uncertainty reads under DistSQL don't benefit from read span refresh mechanism

[andreimatei]

When a regular Scan encounters a ReadWithinUncertaintyInterval error, the TxnCoordSender will immediately try to refresh the txn's read spans and, if successful, retry the batch. This doesn't apply to DistSQL reads which don't go through the TxnCoordSender.
We should figure out another level at which to retry.
Separately, if the whole flow is scheduled on the gateway, everything could go through the TxnCoordSender, I think.

Jira issue: CRDB-5744

[andreimatei]

As the only-visible-to-crdb referenced issue above shows, this is suspected to cause a significant regression in higher percentage select latency on a customer workload.

[jordanlewis]

Hmm, @andreimatei can we talk about this? Seems like something we should tackle soon.

[andreimatei]

Months later, DistSQL reads go through the TxnCoordSender, but the txnSpanRefresher is neutered.
Remote flows do return their read spans to the gateway so, amusingly, the root can attempt to refresh if an error is encountered later by some other query, but not when an error is encountered by DistSQL itself.

[andreimatei]

I've been thinking about this again, because I'm trying to also think through how refreshes should work in a world where we use transactions concurrently (and DistSQL is concurrent + distributed).
The crux of the problem here is the fundamental difference between having a single TxnCoordSender that all reads go through before their results are presented to clients, versus not having it. In the local case, when one requests wants the read timestamp to be advanced, the TCS can refresh all the previous writes and then simply retry the respective batch. However, in the distributed case, nobody knows all the read spans until DistSQL drains all the metadata at the end of the query.

Distilling more, if we want our queries to use a consistent snapshot of the data, then a read r can only use a forwarded read timestamp ts2 if:
a) all the results that have been delivered to the client before any results resulting from r can be refreshed to ts2.
b) all the results that will be delivered to the client after any results resulting from r either come from reads at ts2 or can be refreshed to ts2.

How can we get that in a DistSQL setting, where nodes read independently and results flow through various paths from where they're read to the gateway? The only way I see is by attaching a token to every row flowing through DistSQL, tracking the highest timestamp of a read that contributed to that row. For example, if a row was computed by joining something read at ts 10 with something read at ts 20, we'd say that the resulting row is tagged with 20.
Upon receiving such a row, the DistSQLReceiver would have to make sure that all the nodes that contributed to it can refresh their reads to 20, by contacting all the nodes in the flow and asking them to refresh. Obviously, all the tracking here would be very course. Then, leaves could be allowed to refresh independently (and it can also be the leaves that initiate the refreshing on all the other leaves).

It's worth mentioning that there appears to be a way to completely eliminate the need for refreshes mid-DistSQL flows. Being read-only, it's only uncertainty that causes a DistSQL flow to want to forward its read timestamp. So, before starting a distributed query we could observe the timestamps on all the nodes involved (one round trip), then ratchet everybody to the highest observed timestamp (second rounds trip), then forward our txn's timestamp to the highest observed one (generally, by refreshing) and only then start the flow with no uncertainty remaining. But, it seems pretty pessimistic and expensive...

cc my friends @knz, @bdarnell, @nvanbenschoten, @tbg to see if there's opinions

[knz]

The only way I see is by attaching a token to every row flowing through DistSQL, tracking the highest timestamp of a read that contributed to that row.

What do you do with a filter, an aggregation or an anti-join, where the row carrying the tag is filtered out?

[andreimatei]

I would carry forward the tag even when a row is filtered by infecting all the next rows. I'd have each processor keep track of the highest timestamp that any of its input rows have been tagged with, and I'd tag every output row with that (and also tag the "absence of any output rows" by including this timestamp in each processor's trailing metadata (collected when processors drain). I think that works?

Now that I think about it again, I'm not sure why I phrased this as "tagging rows" rather than describing it in terms of broadcasting metadata and taking advantage of the DistSQL ordered message streams: processors that do KV operations (TableReader, IndexJoiner, etc) would notice when a scan they've done was actually performed at a new (higher) timestamp and would broadcast this information to all their consumers as a ProducerMetadata before sending any more rows to any consumer (the "all consumers" part is important; for example a hash router would send this along on all its output streams). Then, every other processor would respect the convention that such a metadata record is forwarded immediately (as opposed to how we currently handle metadata by deferring its forwarding to later).

[knz]

A distsql processor can have no output row. Indeed it seems like something that's not part of the flow but instead part of the "metadata"

(I really think that this word "metadata" is really bad and should never have been used. The better abstraction is a difference between control plane and data plane. You're playing with the control plane here regardless of what flows data-wise.)

[knz]

There's another challenge in there though. Suppose you have two concurrent processors A and B.

Processor A fails with a logic error (says some SQL built-in function errors out).
Concurrently B is scanning ahead some data.

Today the repatriation of the "metadata" payload will cause the logic error to cancel out whatever result comes from B. That would trash the information bits needed in your algorithm.

If we ever implement savepoint rollbacks in combination with txn refreshes, it's important that the magic that you want to implement does not get invalidated by such a logic error.

[github-actions]

We have marked this issue as stale because it has been inactive for
18 months. If this issue is still relevant, removing the stale label
or adding a comment will keep it active. Otherwise, we'll close it in
5 days to keep the issue queue tidy. Thank you for your contribution
to CockroachDB!

[github-actions]

We have marked this issue as stale because it has been inactive for
18 months. If this issue is still relevant, removing the stale label
or adding a comment will keep it active. Otherwise, we'll close it in
10 days to keep the issue queue tidy. Thank you for your contribution
to CockroachDB!

[knz]

seems still relevant