persistent_state: Bound the memory used during index creation

During index creation, we parallelize the work in case we need to
create multiple indexes. We do this in a way where we create a mpsc
channel to pass the deserialized row data to the index creation
threads.
For btree indexes, we define our own comparator function, aka
make_compare_keys!. This function needs to perform extra work such
deserialization of keys, comparison of keys, ...
In comparison to hashmap indexes, btree indexes are slower to create
by the inner threads.

In such situation, where the inner threads are falling behind, we
end up storing an excessive amount of data in the channel, sometimes
even loading almost all table into memory.

This commit changes the channel to a sync channel, where we limit the
amount of data (rows) that can be in the buffer.

Fixes: REA-5767
Closes: #1528

Release-Note-Core: Fixed an issue where a CREATE CACHE using a range
  condition (Btree index) could consume excessive memory.
Change-Id: I7c70d4b4a398ee6c4c5891bdce23806e5c5e6db6
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9674
Reviewed-by: Sidney Cammeresi <sac@readyset.io>
Tested-by: Buildkite CI

Author: altmannmarcelo

dataflow-state/src/persistent_state/mod.rs

@@ -69,7 +69,7 @@ use std::cmp::Ordering;
 use std::io::{self, Read};
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
+use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender};
 use std::sync::{mpsc, Arc};
 use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
@@ -1836,11 +1836,11 @@ impl PersistentState {
     }
 
     #[allow(clippy::type_complexity)]
-    fn make_channels<T>(num: usize) -> (Vec<Sender<Arc<T>>>, Vec<Receiver<Arc<T>>>) {
+    fn make_channels<T>(num: usize) -> (Vec<SyncSender<Arc<T>>>, Vec<Receiver<Arc<T>>>) {
         let mut txs = Vec::new();
         let mut rxs = Vec::new();
         for _ in 0..num {
-            let (tx, rx) = mpsc::channel();
+            let (tx, rx) = mpsc::sync_channel(INDEX_BATCH_SIZE * 4);
             txs.push(tx);
             rxs.push(rx);
         }
@@ -1916,6 +1916,7 @@ impl PersistentState {
 
             while let (Some(pk), Some(value)) = (iter.key(), iter.value()) {
                 let row = deserialize_row(value);
+                // TODO: only pass data that will be used by any index.
                 let kv = Arc::new(IndexKeyValue::new(pk.to_vec(), row));
                 for tx in &txs {
                     tx.send(Arc::clone(&kv))