readyset-dataflow: Fix unquery eviction for straddled joins.

During on_replay_misses, we add remapped keys for left and right
ancestors if the node is partial. However during unquery eviction,
we were triggering evictions for all ancestors that were not a
base node. This condition is not guaranteed to be the same, as a node
without a state for example, is not a base node but is also not
partial, in this situation we would trigger eviction for a node that
does not have a remapping, causing the downstream nodes to not find
a tag to fullfill the eviction key.

Fixes: REA-6145

Release-Note-Core: Fixed an issue during eviction of straddled joins
  causing it to attempt to evict a key that does not have a remapping.
Change-Id: I86ea30e7cc5881f2fa895f4d872ec0264211bca4
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10985
Tested-by: Buildkite CI
Reviewed-by: Sidney Cammeresi <sac@readyset.io>

Author: altmannmarcelo

readyset-client-test-helpers/src/lib.rs

@@ -130,6 +130,7 @@ pub struct TestBuilder {
     replication_server_id: Option<ReplicationServerId>,
     mixed_comparisons: bool,
     topk: bool,
+    straddled_joins: bool,
     parsing_preset: ParsingPreset,
 }
 
@@ -158,6 +159,7 @@ impl TestBuilder {
             replication_server_id: None,
             mixed_comparisons: true,
             topk: false,
+            straddled_joins: false,
             parsing_preset: ParsingPreset::for_tests(),
         }
     }
@@ -250,6 +252,11 @@ impl TestBuilder {
         self
     }
 
+    pub fn set_straddled_joins(mut self, straddled_joins: bool) -> Self {
+        self.straddled_joins = straddled_joins;
+        self
+    }
+
     pub fn parsing_preset(mut self, parsing_preset: ParsingPreset) -> Self {
         self.parsing_preset = parsing_preset;
         self
@@ -301,6 +308,7 @@ impl TestBuilder {
         };
         builder.set_persistence(persistence);
         builder.set_topk(self.topk);
+        builder.set_straddled_joins(self.straddled_joins);
         builder.set_pagination(false);
         builder.set_mixed_comparisons(self.mixed_comparisons);
         builder.set_parsing_preset(self.parsing_preset);

readyset-client-test-helpers/src/mysql_helpers.rs

@@ -39,6 +39,33 @@ pub async fn last_query_info(conn: &mut impl Queryable) -> QueryInfo {
         .unwrap()
 }
 
+/// Helper function that gets EXPLAIN MATERIALIZATIONS and ensure that nodes have a specific number of keys
+pub async fn assert_materializations_have_key(
+    conn: &mut impl Queryable,
+    skip_base_node: bool,
+    key: &str,
+) {
+    type Materialization = (usize, usize, String, String, String, usize, bool, String);
+    let materializations: Vec<Materialization> =
+        conn.query("EXPLAIN MATERIALIZATIONS").await.unwrap();
+    for (
+        _domain_index,
+        _node_index,
+        _node_name,
+        node_description,
+        keys,
+        _size_bytes,
+        _partial,
+        _indexes,
+    ) in materializations
+    {
+        if skip_base_node && node_description == "B" {
+            continue;
+        }
+        assert_eq!(keys, key);
+    }
+}
+
 pub async fn recreate_database<N>(dbname: N)
 where
     N: Display,

readyset-dataflow/src/domain/mod.rs

@@ -2708,7 +2708,12 @@ impl Domain {
                 let keys = if is_generated {
                     // No need to store remapping; if it's not there, we don't have the key.
                     let mut missed_keys = self.nodes[src].borrow_mut().handle_upquery(m)?;
-                    missed_keys.retain(|m| !self.nodes[m.node].borrow().is_base());
+                    missed_keys.retain(|upstream_miss| {
+                        self.state
+                            .get(upstream_miss.node)
+                            .map(|state| state.is_partial())
+                            .unwrap_or(false)
+                    });
                     missed_keys
                 } else {
                     vec![m]

readyset-e2e-tests/tests/straddled_joins.rs

@@ -0,0 +1,98 @@
+use mysql_async::prelude::Queryable;
+use readyset_adapter::backend::MigrationMode;
+use readyset_client_metrics::QueryDestination;
+use readyset_client_test_helpers::mysql_helpers::{
+    self, MySQLAdapter, assert_materializations_have_key, last_query_info,
+};
+use readyset_client_test_helpers::{TestBuilder, sleep};
+use test_utils::tags;
+
+/// REA-6145: no tag found for value
+#[tokio::test]
+#[tags(serial, mysql_upstream)]
+async fn test_sj_eviction_no_remapping() {
+    readyset_tracing::init_test_logging();
+    let db_name = "sj_eviction_no_remapping";
+    mysql_helpers::recreate_database(db_name).await;
+
+    let (rs_opts, mut handle, shutdown_tx) = TestBuilder::default()
+        .recreate_database(false)
+        .migration_mode(MigrationMode::OutOfBand)
+        .set_straddled_joins(true)
+        .replicate_db(db_name.to_string())
+        .build::<MySQLAdapter>()
+        .await;
+
+    let upstream_opts = mysql_helpers::upstream_config().db_name(Some(db_name));
+    let mut upstream_conn = mysql_async::Conn::new(upstream_opts).await.unwrap();
+    let mut rs_conn = mysql_async::Conn::new(rs_opts).await.unwrap();
+
+    upstream_conn
+        .query_drop(
+            "CREATE TABLE t1 (
+  id int NOT NULL AUTO_INCREMENT,
+  t2_id int DEFAULT NULL,
+  key_v varchar(255) DEFAULT NULL,
+  deleted_at datetime DEFAULT NULL,
+  PRIMARY KEY (id)
+);
+CREATE TABLE t2 (
+  id int NOT NULL AUTO_INCREMENT,
+  key_i int DEFAULT NULL,
+  deleted_at datetime DEFAULT NULL,
+  PRIMARY KEY (id)
+);",
+        )
+        .await
+        .unwrap();
+
+    sleep().await;
+
+    rs_conn
+        .query_drop("CREATE CACHE sj_cache FROM SELECT * FROM t1 INNER JOIN t2 ON (t1.t2_id = t2.id) WHERE t1.deleted_at IS NULL AND t2.deleted_at IS NULL AND t2.key_i = ? AND t1.key_v = ?;")
+        .await
+        .unwrap();
+
+    sleep().await;
+
+    let result: Vec<(i32, i32)> = rs_conn.query("SELECT * FROM t1 INNER JOIN t2 ON (t1.t2_id = t2.id) WHERE t1.deleted_at IS NULL AND t2.deleted_at IS NULL AND t2.key_i = 1 AND t1.key_v = '3';").await.unwrap();
+    assert_eq!(result.len(), 0);
+
+    let last_statement = last_query_info(&mut rs_conn).await;
+    assert_eq!(
+        last_statement.destination,
+        QueryDestination::Readyset(Some("sj_cache".into()))
+    );
+
+    let result: Vec<(i32, i32)> = rs_conn.query("SELECT * FROM t1 INNER JOIN t2 ON (t1.t2_id = t2.id) WHERE t1.deleted_at IS NULL AND t2.deleted_at IS NULL AND t2.key_i = 1 AND t1.key_v = '2';").await.unwrap();
+    assert_eq!(result.len(), 0);
+
+    let last_statement = last_query_info(&mut rs_conn).await;
+    assert_eq!(
+        last_statement.destination,
+        QueryDestination::Readyset(Some("sj_cache".into()))
+    );
+
+    let result: Vec<(i32, i32)> = rs_conn.query("SELECT * FROM t1 INNER JOIN t2 ON (t1.t2_id = t2.id) WHERE t1.deleted_at IS NULL AND t2.deleted_at IS NULL AND t2.key_i = 1 AND t1.key_v = '1';").await.unwrap();
+    assert_eq!(result.len(), 0);
+
+    let last_statement = last_query_info(&mut rs_conn).await;
+    assert_eq!(
+        last_statement.destination,
+        QueryDestination::Readyset(Some("sj_cache".into()))
+    );
+
+    assert_materializations_have_key(&mut rs_conn, true, "3").await;
+
+    handle.flush_partial().await.unwrap();
+    handle.flush_partial().await.unwrap();
+    handle.flush_partial().await.unwrap();
+    sleep().await;
+
+    shutdown_tx.shutdown().await;
+
+    upstream_conn
+        .query_drop(format!("DROP DATABASE {db_name}"))
+        .await
+        .unwrap();
+}