Type the instant cell: one unit from storage to every protocol edge

nodedb-cluster-tests/tests/common_suite/cases/shuffle_join_end_to_end.rs

@@ -24,6 +24,14 @@ use nodedb_types::DatabaseId;
 
 use crate::common::cluster_harness::{TestCluster, wait_for};
 
+/// The instant [`a_shuffle_join_renders_a_time_key_as_the_stored_instant`]
+/// stores in its timeseries collection.
+const EARLY: &str = "2020-03-05 10:00:00";
+/// `EARLY` as a declared `TIMESTAMP` time key renders it. The engine stores
+/// 1583402400000 epoch milliseconds and emits the cell as a typed instant,
+/// which the pgwire encoder writes as ISO-8601 UTC.
+const EARLY_ISO: &str = "2020-03-05T10:00:00.000000Z";
+
 /// Run `sql` and collect the `id` column of every returned data row, sorted, so
 /// the result is order-independent for equality assertions.
 async fn collect_ids(client: &tokio_postgres::Client, sql: &str, col: &str) -> Vec<String> {
@@ -211,3 +219,132 @@ async fn distributed_shuffle_join_matches_inner_join() {
 
     cluster.shutdown().await;
 }
+
+/// A timeseries `TIMESTAMP` time key projected through a shuffle join denotes
+/// the stored instant, the same as a direct read does. The shuffle path
+/// repartitions rows through its own coordinator-side encode/decode hop, so
+/// it is a route to the cell distinct from the local join and the direct
+/// `SELECT` both.
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn a_shuffle_join_renders_a_time_key_as_the_stored_instant() {
+    const LEFT: &str = "ts_shuffle_events";
+    const RIGHT: &str = "ts_shuffle_hosts";
+    assert_ne!(
+        vshard_for_collection(DatabaseId::DEFAULT, LEFT),
+        vshard_for_collection(DatabaseId::DEFAULT, RIGHT),
+        "test collections must hash to different vShards to exercise cross-node shuffle"
+    );
+
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {LEFT} \
+             (captured_at TIMESTAMP TIME_KEY, host TEXT, v FLOAT) \
+             WITH (engine='timeseries')"
+        ))
+        .await
+        .expect("CREATE COLLECTION for the timeseries side");
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {RIGHT} (id TEXT PRIMARY KEY, region TEXT) \
+             WITH (engine='document_strict')"
+        ))
+        .await
+        .expect("CREATE COLLECTION for the document side");
+
+    wait_for(
+        "all 3 nodes see both collections",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 2)
+        },
+    )
+    .await;
+
+    cluster.nodes[0]
+        .client
+        .simple_query(&format!(
+            "INSERT INTO {LEFT} (captured_at, host, v) VALUES ('{EARLY}', 'h1', 1.5)"
+        ))
+        .await
+        .expect("insert timeseries row");
+    cluster.nodes[0]
+        .client
+        .simple_query(&format!(
+            "INSERT INTO {RIGHT} (id, region) VALUES ('h1', 'eu')"
+        ))
+        .await
+        .expect("insert document row");
+
+    for (idx, node) in cluster.nodes.iter().enumerate() {
+        wait_for(
+            &format!("node {idx} sees the timeseries row"),
+            Duration::from_secs(15),
+            Duration::from_millis(50),
+            || {
+                let n = tokio::task::block_in_place(|| {
+                    tokio::runtime::Handle::current().block_on(count_rows(
+                        &node.client,
+                        &format!("SELECT host FROM {LEFT}"),
+                    ))
+                });
+                n >= 1
+            },
+        )
+        .await;
+        wait_for(
+            &format!("node {idx} sees the document row"),
+            Duration::from_secs(15),
+            Duration::from_millis(50),
+            || {
+                let n = tokio::task::block_in_place(|| {
+                    tokio::runtime::Handle::current()
+                        .block_on(count_rows(&node.client, &format!("SELECT id FROM {RIGHT}")))
+                });
+                n >= 1
+            },
+        )
+        .await;
+    }
+
+    cluster.nodes[1]
+        .client
+        .simple_query("SET nodedb.force_shuffle_join = on")
+        .await
+        .expect("SET nodedb.force_shuffle_join");
+    cluster.nodes[1]
+        .client
+        .simple_query("SET nodedb.shuffle_num_parts = 4")
+        .await
+        .expect("SET nodedb.shuffle_num_parts");
+
+    let join_sql = format!(
+        "SELECT {LEFT}.captured_at FROM {LEFT} \
+         INNER JOIN {RIGHT} ON {LEFT}.host = {RIGHT}.id"
+    );
+    let msgs = cluster.nodes[1]
+        .client
+        .simple_query(&join_sql)
+        .await
+        .expect("a shuffle join over a time key must succeed");
+    let values: Vec<String> = msgs
+        .into_iter()
+        .filter_map(|m| match m {
+            tokio_postgres::SimpleQueryMessage::Row(row) => row.get(0).map(str::to_string),
+            _ => None,
+        })
+        .collect();
+
+    assert_eq!(values.len(), 1, "one event matches one host: {values:?}");
+    assert_eq!(
+        values[0], EARLY_ISO,
+        "a shuffle-joined time key must denote {EARLY}: got {values:?}"
+    );
+
+    cluster.shutdown().await;
+}

nodedb-cluster-tests/tests/sql_cluster_cross_node_dml_tests/join_cross_node.rs

@@ -26,6 +26,18 @@ async fn count_rows(client: &tokio_postgres::Client, sql: &str) -> usize {
         .count()
 }
 
+/// Helper: run a simple query and return the first column of every data row,
+/// as the pgwire text encoder rendered it.
+async fn first_column_text(client: &tokio_postgres::Client, sql: &str) -> Vec<String> {
+    let msgs = client.simple_query(sql).await.expect("simple_query");
+    msgs.into_iter()
+        .filter_map(|m| match m {
+            tokio_postgres::SimpleQueryMessage::Row(row) => row.get(0).map(str::to_string),
+            _ => None,
+        })
+        .collect()
+}
+
 /// A cross-node inner join between two single-vShard-homed collections must
 /// return every matching row, regardless of which node the SELECT is issued
 /// from. Before the build-side gather fix this returned 0 rows when the build
@@ -151,3 +163,144 @@ async fn cross_node_join_returns_all_matches() {
 
     cluster.shutdown().await;
 }
+
+/// A distributed join gathers the remote (build) side through the coordinator
+/// while scanning the local (probe) side directly. When the `ON` predicate
+/// compares two TIME_KEY columns, both sides must reach that comparison
+/// expressed in the same unit — the gathered side must not arrive pre-decoded
+/// into a different representation than the locally scanned side. This holds
+/// for every node the query runs from, whichever side that node hosts.
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn cross_node_join_compares_time_keys_in_one_unit() {
+    use nodedb_cluster::routing::vshard_for_collection;
+    use nodedb_types::DatabaseId;
+    const EVENTS: &str = "ts_events";
+    const FEATURES: &str = "ts_features";
+    assert_ne!(
+        vshard_for_collection(DatabaseId::DEFAULT, EVENTS),
+        vshard_for_collection(DatabaseId::DEFAULT, FEATURES),
+        "test collections must hash to different vShards to exercise cross-node join"
+    );
+
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {EVENTS} \
+             (captured_at TIMESTAMP TIME_KEY, host TEXT, v FLOAT) \
+             WITH (engine='timeseries')"
+        ))
+        .await
+        .expect("CREATE COLLECTION ts_events");
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {FEATURES} \
+             (captured_at TIMESTAMP TIME_KEY, host TEXT, v FLOAT) \
+             WITH (engine='timeseries')"
+        ))
+        .await
+        .expect("CREATE COLLECTION ts_features");
+
+    wait_for(
+        "all 3 nodes see both collections",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 2)
+        },
+    )
+    .await;
+
+    cluster.nodes[0]
+        .client
+        .simple_query(&format!(
+            "INSERT INTO {EVENTS} (captured_at, host, v) VALUES ('2020-03-05 10:00:00', 'h1', 1.5)"
+        ))
+        .await
+        .expect("insert event row");
+    cluster.nodes[0]
+        .client
+        .simple_query(&format!(
+            "INSERT INTO {FEATURES} (captured_at, host, v) VALUES ('2020-03-05 09:00:00', 'h1', 2.5)"
+        ))
+        .await
+        .expect("insert feature row");
+
+    for (idx, node) in cluster.nodes.iter().enumerate() {
+        wait_for(
+            &format!("node {idx} sees the event row"),
+            Duration::from_secs(15),
+            Duration::from_millis(50),
+            || {
+                let n = tokio::task::block_in_place(|| {
+                    tokio::runtime::Handle::current().block_on(count_rows(
+                        &node.client,
+                        &format!("SELECT host FROM {EVENTS}"),
+                    ))
+                });
+                n >= 1
+            },
+        )
+        .await;
+        wait_for(
+            &format!("node {idx} sees the feature row"),
+            Duration::from_secs(15),
+            Duration::from_millis(50),
+            || {
+                let n = tokio::task::block_in_place(|| {
+                    tokio::runtime::Handle::current().block_on(count_rows(
+                        &node.client,
+                        &format!("SELECT host FROM {FEATURES}"),
+                    ))
+                });
+                n >= 1
+            },
+        )
+        .await;
+    }
+
+    // The join's ON predicate compares the two TIME_KEY columns directly
+    // (`FEATURES.captured_at <= EVENTS.captured_at`). This only matches the
+    // single feature row when both sides land in the comparison expressed
+    // in the same unit, regardless of which node the coordinator gathers the
+    // build side from.
+    let value_sql = format!(
+        "SELECT {FEATURES}.v FROM {EVENTS} INNER JOIN {FEATURES} \
+         ON {EVENTS}.host = {FEATURES}.host \
+         AND {FEATURES}.captured_at <= {EVENTS}.captured_at"
+    );
+    for (idx, node) in cluster.nodes.iter().enumerate() {
+        let rows = first_column_text(&node.client, &value_sql).await;
+        assert_eq!(
+            rows,
+            vec!["2.5".to_string()],
+            "node {idx}: join comparing time keys must return the one matching feature row"
+        );
+    }
+
+    // The joined TIME_KEY is a typed instant on every node, so the
+    // coordinator's gather step renders it as ISO-8601 UTC. The stored
+    // 1583402400000 milliseconds read as a number would denote 1970-01-19.
+    const EARLY_ISO: &str = "2020-03-05T10:00:00.000000Z";
+    let time_key_sql = format!(
+        "SELECT {EVENTS}.captured_at FROM {EVENTS} INNER JOIN {FEATURES} \
+         ON {EVENTS}.host = {FEATURES}.host"
+    );
+    for (idx, node) in cluster.nodes.iter().enumerate() {
+        let rows = first_column_text(&node.client, &time_key_sql).await;
+        assert_eq!(
+            rows.len(),
+            1,
+            "node {idx}: the join must yield one row: {rows:?}"
+        );
+        assert_eq!(
+            rows[0], EARLY_ISO,
+            "node {idx}: joined time key must denote 2020-03-05T10:00:00Z: got {rows:?}"
+        );
+    }
+
+    cluster.shutdown().await;
+}

nodedb-columnar/src/materialize_rows/extract.rs

@@ -22,11 +22,15 @@ pub(crate) fn extract_row_value(
     use nodedb_types::value::Value;
 
     let v = match col {
-        DecodedColumn::Int64 { values, valid } => {
+        // The reader infers a column's physical kind from its codec, so a
+        // time column decodes as `Int64`. The declared type decides the cell:
+        // an instant column yields the variant its declared type names;
+        // every other type yields the integer stored.
+        DecodedColumn::Int64 { values, valid } | DecodedColumn::Timestamp { values, valid } => {
             if !valid[row_idx] {
                 Value::Null
             } else {
-                Value::Integer(values[row_idx])
+                col_type.time_cell(values[row_idx])
             }
         }
         DecodedColumn::Float64 { values, valid } => {
@@ -36,20 +40,6 @@ pub(crate) fn extract_row_value(
                 Value::Float(values[row_idx])
             }
         }
-        DecodedColumn::Timestamp { values, valid } => {
-            if !valid[row_idx] {
-                Value::Null
-            } else {
-                let micros = values[row_idx];
-                let dt = nodedb_types::datetime::NdbDateTime::from_micros(micros);
-                match col_type {
-                    nodedb_types::columnar::ColumnType::Timestamptz
-                    | nodedb_types::columnar::ColumnType::SystemTimestamp => Value::DateTime(dt),
-                    // Timestamp (naive) and anything else that maps to i64 storage.
-                    _ => Value::NaiveDateTime(dt),
-                }
-            }
-        }
         DecodedColumn::Bool { values, valid } => {
             if !valid[row_idx] {
                 Value::Null

nodedb-columnar/src/memtable/column_data/access.rs

@@ -2,6 +2,7 @@
 
 //! Read-only access methods on `ColumnData`: validity checks, value extraction.
 
+use nodedb_types::columnar::ColumnType;
 use nodedb_types::value::Value;
 use nodedb_types::value_from_msgpack;
 
@@ -57,17 +58,21 @@ impl ColumnData {
     }
 
     /// Extract a single row's value as `nodedb_types::Value`.
-    pub(crate) fn get_value(&self, row: usize) -> Value {
+    ///
+    /// A time cell is typed by the column's declared type: an instant column
+    /// yields `Value::NaiveDateTime` (`Timestamp`) or `Value::DateTime`
+    /// (`Timestamptz`) from the stored epoch microseconds, and every other
+    /// declared type backed by time storage (`SystemTimestamp`, `Duration`)
+    /// yields the integer stored.
+    pub(crate) fn get_value(&self, row: usize, declared: &ColumnType) -> Value {
         if self.is_null(row) {
             return Value::Null;
         }
         match self {
             Self::Int64 { values, .. } => Value::Integer(values[row]),
             Self::Float64 { values, .. } => Value::Float(values[row]),
             Self::Bool { values, .. } => Value::Bool(values[row]),
-            Self::Timestamp { values, .. } => Value::DateTime(
-                nodedb_types::datetime::NdbDateTime::from_micros(values[row]),
-            ),
+            Self::Timestamp { values, .. } => declared.time_cell(values[row]),
             Self::Decimal { values, .. } => {
                 Value::Decimal(rust_decimal::Decimal::deserialize(values[row]))
             }