Subscriptions impl, server side #1997

lcodes · 2024-11-15T17:01:14Z

Description of Changes

Moving subscriptions from one batch-subscribe call to one call per subscribe with matching unsubscribe call.

API and ABI breaking changes

Updates to the websocket API and introduction of new messages.

Expected complexity level and risk

3: changes the way subscriptions are stored and used.

Testing

In progress

bfops · 2024-11-19T09:50:43Z

(Seems to be failing tests?)

Centril · 2024-11-19T14:38:36Z

crates/cli/src/subcommands/subscribe.rs

@@ -121,7 +121,7 @@ struct SubscriptionTable {
 }

 pub async fn exec(config: Config, args: &ArgMatches) -> Result<(), anyhow::Error> {
-    let queries = args.get_many::<String>("query").unwrap();
+    let query = args.get_one::<String>("query").unwrap();


In this case and because it is not part of the subscription proposal, I would consider not changing the CLI, as there is no way to add another query once you have added the first. Rather, you can issue N subscribe commands. We could also do this as future work if you prefer and make an issue for that.

Centril · 2024-11-19T14:42:30Z

crates/client-api-messages/src/websocket.rs

    /// Register SQL queries on which to receive updates.
    Subscribe(Subscribe),
+    /// Unregister SQL queries which are receiving updates.
+    Unsubscribe(Unsubscribe),


Suggested change

/// Register SQL queries on which to receive updates.

Subscribe(Subscribe),

/// Unregister SQL queries which are receiving updates.

Unsubscribe(Unsubscribe),

/// Register a SQL query on which to receive an initial and subsequent updates for.

/// The subscribed set is mutable, that is, this `Subscribe` message

/// can be followed by more, or `Unsubscribe` messages.

Subscribe(Subscribe),

/// Unregister a SQL query which the client is receiving updates for.

Unsubscribe(Unsubscribe),

Centril · 2024-11-19T14:43:40Z

crates/client-api-messages/src/websocket.rs

+    /// After connecting, to inform client of its identity.
+    IdentityToken(IdentityToken),


(This will conflict with the ids-not-names stuff.)

What is the ids-not-names stuff?

Centril · 2024-11-19T17:21:05Z

crates/core/src/subscription/execution_unit.rs

+        let ptr = &self.data as *const [u8; 32] as *const u128;
+        QueryId { hash: unsafe { u256::from_words(*ptr, *ptr.wrapping_add(1)) } }


Why do we need both a QueryHash and QueryId, why not reuse QueryHash = QueryId?

Also, why not use u256::from_le_bytes(self.data) or some such?

Centril · 2024-11-19T17:22:18Z

crates/core/src/subscription/execution_unit.rs

+impl Into<QueryHash> for QueryId {
+    fn into(self) -> QueryHash {
+        let ptr = &self.hash.0 as *const [u128; 2] as *const [u8; 32];
+        QueryHash { data: unsafe { *ptr } }
+    }
+}


Same here...

Centril · 2024-11-19T17:49:57Z

crates/core/src/client/messages.rs

+                    FormatSwitch::Bsatn(table_rows) => FormatSwitch::Bsatn(ws::SubscribeApplied {
+                        total_host_execution_duration_micros,
+                        request_id,
+                        query_id,
+                        rows: ws::SubscribeRows {
+                            table_id: result.table_id,
+                            table_name: result.table_name,
+                            table_rows
+                        },
+                    }.into()),
+                    FormatSwitch::Json(table_rows) => FormatSwitch::Json(ws::SubscribeApplied {
+                        total_host_execution_duration_micros,
+                        request_id,
+                        query_id,
+                        rows: ws::SubscribeRows {
+                            table_id: result.table_id,
+                            table_name: result.table_name,
+                            table_rows
+                        },
+                    }.into()),
+                }
+            },
+            SubscriptionResult::Unsubscribe(result) => {
+                protocol.assert_matches_format_switch(&result.table_rows);
+                match result.table_rows {
+                    FormatSwitch::Bsatn(table_rows) => FormatSwitch::Bsatn(ws::UnsubscribeApplied {
+                        total_host_execution_duration_micros,
+                        request_id,
+                        query_id,
+                        rows: ws::SubscribeRows {
+                            table_id: result.table_id,
+                            table_name: result.table_name,
+                            table_rows
+                        },
+                    }.into()),
+                    FormatSwitch::Json(table_rows) => FormatSwitch::Json(ws::UnsubscribeApplied {
+                        total_host_execution_duration_micros,
+                        request_id,
+                        query_id,
+                        rows: ws::SubscribeRows {
+                            table_id: result.table_id,
+                            table_name: result.table_name,
+                            table_rows
+                        },
+                    }.into()),


The code duplication here is rather unfortunate :(

Centril · 2024-11-19T17:56:01Z

crates/core/src/subscription/module_subscription_manager.rs

+        let mut remove_table_query = |table_id: TableId| {
+            if let Entry::Occupied(mut entry) = self.tables.entry(table_id) {
+                let hashes = entry.get_mut();
+                if hashes.remove(&query) && hashes.is_empty() {
+                    entry.remove();
+                }
+            }
+        };


This and the version below from which it was copied looks like it could be extracted to a method.

Centril · 2024-11-19T17:57:32Z

crates/core/src/subscription/module_subscription_manager.rs

+            }
+        }
+
+        if let Some(ids) = self.subscribers.get_mut(&query) {


Could use ?

Centril · 2024-11-19T18:02:14Z

crates/core/src/subscription/query.rs

+    let mut compiled = compile_sql(relational_db, auth, tx, &input)?;
+    match compiled.len() {
+        1 => {},
+        0 => return Err(SubscriptionError::Empty.into()),
+        _ => return Err(SubscriptionError::Multiple.into()),
+    }
+
+    Err(SubscriptionError::SideEffect(match compiled.pop().unwrap() {


Suggested change

let mut compiled = compile_sql(relational_db, auth, tx, &input)?;

match compiled.len() {

1 => {},

0 => return Err(SubscriptionError::Empty.into()),

_ => return Err(SubscriptionError::Multiple.into()),

}

Err(SubscriptionError::SideEffect(match compiled.pop().unwrap() {

let mut compiled = compile_sql(relational_db, auth, tx, &input)?;

let single = match (compiled.pop(), complied.pop()) {

(None, None) => return Err(SubscriptionError::Empty.into()),

(Some(_), Some(_)) => return Err(SubscriptionError::Multiple.into()),

(Some(x), None) => x,

};

Err(SubscriptionError::SideEffect(match single {

Centril · 2024-11-19T18:03:42Z

crates/core/src/subscription/execution_unit.rs

@@ -208,7 +223,7 @@ impl ExecutionUnit {
        sql: &str,
        slow_query_threshold: Option<Duration>,
        compression: Compression,
-    ) -> Option<TableUpdate<F>> {
+    ) -> TableUpdate<F> {


Why this change? I noticed the filter_map -> map

gefjon · 2024-11-20T21:02:47Z

@jsdt Note there's an edge case where republishing a module with a hot-swap can remove an index, which can break an indexed semijoin subscription query, since we refuse incremental semijoins which are not indexed on both of the joined columns. The intended semantics here is that, after a hotswap, we recompile all active subscribed queries. If any of them fails to compile, like in the removed-index case, the host sends an error message to the client and then ends the subscription, but the websocket connection is not closed. Any queries that successfully compile stay in place.

Please find out if this PR implements that behavior. If it does not, please create a ticket to fix it in a follow-up.

Subscriptions impl, server side

1d8e2ca

lcodes requested review from bfops, cloutiertyler, Centril and gefjon as code owners November 15, 2024 17:01

Centril requested changes Nov 19, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Subscriptions impl, server side #1997

Subscriptions impl, server side #1997

lcodes commented Nov 15, 2024

bfops commented Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

jsdt Nov 21, 2024

gefjon Nov 21, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

Centril Nov 19, 2024

gefjon commented Nov 20, 2024

		/// After connecting, to inform client of its identity.
		IdentityToken(IdentityToken),

		let ptr = &self.data as const [u8; 32] as const u128;
		QueryId { hash: unsafe { u256::from_words(ptr, ptr.wrapping_add(1)) } }

Subscriptions impl, server side #1997

Are you sure you want to change the base?

Subscriptions impl, server side #1997

Conversation

lcodes commented Nov 15, 2024

Description of Changes

API and ABI breaking changes

Expected complexity level and risk

Testing

bfops commented Nov 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gefjon commented Nov 20, 2024