streamclient: add random stream client

[pbardea]

See individual commits, but this PR does some cleanup while introducing a
stream client implementation that randomly generates rows.

I broke down each change into its own commit for reviews, but let me know
if splitting this into separate PRs would be helpful.

[cockroach-teamcity]

This change is 

[adityamaru]

Reviewed 18 of 18 files at r1, 7 of 7 files at r2, 5 of 5 files at r3, 10 of 10 files at r4, 3 of 3 files at r5, 1 of 1 files at r6, 1 of 1 files at r7.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @pbardea)

---

pkg/ccl/streamingccl/streamclient/client.go, line 30 at r3 (raw file):

the partition and close the event channel

nit: last part of the sentence is repeated.

---

pkg/ccl/streamingccl/streamclient/client_test.go, line 89 at r6 (raw file):

				panic(fmt.Sprintf("unexpected event type %v", event.Type()))
			}
			numReceivedEvents++

can we delete?

---

pkg/ccl/streamingccl/streamclient/random_stream_client.go, line 186 at r4 (raw file):

			}

			// TODO: Consider keeping an in-memory copy so that tests can verify

+1 this was on my mind too.

---

pkg/ccl/streamingccl/streamclient/random_stream_client.go, line 199 at r4 (raw file):

}

func (m *randomStreamClient) makeRandomKey(r *rand.Rand, minTs time.Time) roachpb.KeyValue {

love this new version!

---

pkg/kv/bulk/sst_batcher.go, line 146 at r4 (raw file):

// Keys must be added in order.
func (b *SSTBatcher) AddMVCCKey(ctx context.Context, key storage.MVCCKey, value []byte) error {
	if len(b.batchEndKey) > 0 && bytes.Equal(b.batchEndKey, key.Key) && b.disallowShadowing {

nice catch

[pbardea]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @adityamaru and @pbardea)

---

pkg/ccl/streamingccl/streamclient/client.go, line 30 at r3 (raw file):

Previously, adityamaru (Aditya Maru) wrote…

the partition and close the event channel

nit: last part of the sentence is repeated.

Done.

---

pkg/ccl/streamingccl/streamclient/client_test.go, line 89 at r6 (raw file):

Previously, adityamaru (Aditya Maru) wrote…

can we delete?

Nice catch. Done.

---

pkg/ccl/streamingccl/streamclient/stream_client.go, line 36 at r7 (raw file):

Previously, pbardea (Paul Bardea) wrote…

This doesn't need to be a select.

Done.

---

pkg/kv/bulk/sst_batcher.go, line 146 at r4 (raw file):

Previously, adityamaru (Aditya Maru) wrote…

nice catch

I don't think this was a bug before since we always ingested a snapshot of the data rather than the entire revision history.

[pbardea]

The tests against CI were flaking so I through together a change that allows the tests to inject "observers" into the client. Curious what you think about this approach. It allows the tests to run deterministically and we could leverage this by adding another "observer" in the test that keeps the in-memory state like we were talking about.

[adityamaru]

The tests against CI were flaking so I through together a change that allows the tests to inject "observers" into the client. Curious what you think about this approach. It allows the tests to run deterministically and we could leverage this by adding another "observer" in the test that keeps the in-memory state like we were talking about.

Just looked at it, and it's a good idea. I guess we still rely on being able to get a hold of the client to be able to invoke client.RegisterInterceptor, which will be hard when we write job level tests. We'd probably still need to add a testing knob where we deposit our "interceptors" and then the ingestion processor can register those for us in the NewStreamClient call for test:// URIs?

[adityamaru]

LGTM!

[pbardea]

TFTR!
bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)

[shermanCRL]

cc @HonoreDB this might be informative, as its the client of the producer stream