This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-gateway.git
The following commit(s) were added to refs/heads/main by this push:
new 2d08fc0 [SPARK-59021] Add scg-store-memory and scg-pool-static crates
2d08fc0 is described below
commit 2d08fc0249a99f93cd2e36869f18f7188f064707
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Aug 26 23:50:19 2026 -0700
[SPARK-59021] Add scg-store-memory and scg-pool-static crates
### What changes were proposed in this pull request?
Part of importing the Spark Connect Gateway reference implementation into
`apache/spark-connect-gateway` as a series of small, bottom-up PRs (tracked
in
the do-not-merge reference PR #1).
This adds two small crates that implement the `scg-routing` traits, grouped
into
one PR:
- `scg-store-memory` — an in-process `AffinityStore` for single-replica
deployments and tests (for multi-replica HA, `scg-store-redis`, added in a
later PR, is used instead).
- `scg-pool-static` — a backend `Pool` whose membership is fixed at startup
(the
dynamic K8s-Endpoints-watch variant `scg-pool-k8s` comes in a later PR).
Both depend on the `scg-routing` crate whose traits they implement. They
compile
and test on their own (2 unit tests each) and are appended to the workspace
`members` list. They are consumed later by the `proxy` crate (both are proxy
dev-dependencies used by its tests).
### Why are the changes needed?
Implements the reference gateway from the approved SPIP (SPARK-58455),
imported
one reviewable piece at a time.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`cargo fmt --all --check`, `cargo clippy --workspace --all-targets -- -D
warnings`,
`cargo build --workspace`, and `cargo test --workspace` all pass locally
(store-memory 2 tests, pool-static 2 tests).
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Code.
Closes #11 from viirya/import-08-store-memory-pool-static.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
Cargo.lock | 18 ++++++++
Cargo.toml | 2 +
crates/pool-static/Cargo.toml | 31 ++++++++++++++
crates/pool-static/src/lib.rs | 78 ++++++++++++++++++++++++++++++++++
crates/store-memory/Cargo.toml | 35 +++++++++++++++
crates/store-memory/src/lib.rs | 96 ++++++++++++++++++++++++++++++++++++++++++
6 files changed, 260 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 8194fef..c59f45e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2329,6 +2329,14 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "scg-pool-static"
+version = "0.1.0"
+dependencies = [
+ "scg-routing",
+ "thiserror 2.0.20",
+]
+
[[package]]
name = "scg-ratelimit"
version = "0.1.0"
@@ -2352,6 +2360,16 @@ dependencies = [
"tonic",
]
+[[package]]
+name = "scg-store-memory"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "parking_lot",
+ "scg-routing",
+ "tokio",
+]
+
[[package]]
name = "scg-tenant"
version = "0.1.0"
diff --git a/Cargo.toml b/Cargo.toml
index 77d7202..c527304 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -24,6 +24,8 @@ members = [
"crates/audit",
"crates/ratelimit",
"crates/tenant",
+ "crates/store-memory",
+ "crates/pool-static",
]
[workspace.package]
diff --git a/crates/pool-static/Cargo.toml b/crates/pool-static/Cargo.toml
new file mode 100644
index 0000000..5c67017
--- /dev/null
+++ b/crates/pool-static/Cargo.toml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "scg-pool-static"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "Static round-robin backend pool for the Spark Connect Gateway."
+publish = false
+
+[dependencies]
+scg-routing = { path = "../routing" }
+thiserror = { workspace = true }
diff --git a/crates/pool-static/src/lib.rs b/crates/pool-static/src/lib.rs
new file mode 100644
index 0000000..8854f50
--- /dev/null
+++ b/crates/pool-static/src/lib.rs
@@ -0,0 +1,78 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Backend pool whose membership is fixed at startup. Pairs with
+//! `scg-pool-k8s` for the dynamic K8s-Endpoints-watch variant; both
+//! implement the same `Pool` trait and can be swapped via config.
+
+use scg_routing::{BackendMember, Pool};
+
+#[derive(Debug, thiserror::Error)]
+pub enum StaticPoolError {
+ #[error("static pool: at least one backend address required")]
+ Empty,
+}
+
+pub struct StaticPool {
+ backends: Vec<String>,
+}
+
+impl StaticPool {
+ pub fn new<I, S>(addresses: I) -> Result<Self, StaticPoolError>
+ where
+ I: IntoIterator<Item = S>,
+ S: Into<String>,
+ {
+ let backends: Vec<String> =
addresses.into_iter().map(Into::into).collect();
+ if backends.is_empty() {
+ return Err(StaticPoolError::Empty);
+ }
+ Ok(Self { backends })
+ }
+}
+
+impl Pool for StaticPool {
+ fn members(&self) -> Vec<BackendMember> {
+ // Static pools have no notion of health — every configured
+ // backend is presumed healthy. Wrap with `scg-healthcheck`'s
+ // `HealthAwarePool` to add active gRPC health probing; on a
+ // bare StaticPool a failed forward surfaces as an error to
+ // the client and the selection strategy moves onward for the
+ // next session. Labels/weights are defaults until the config
+ // grows per-address metadata.
+ self.backends.iter().map(BackendMember::new).collect()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn members_returns_configured_list_with_default_metadata() {
+ let p = StaticPool::new(["a", "b", "c"]).unwrap();
+ let m = p.members();
+ assert_eq!(
+ m.iter().map(|b| b.addr.as_str()).collect::<Vec<_>>(),
+ vec!["a", "b", "c"]
+ );
+ assert!(m.iter().all(|b| b.weight == 1 && b.labels.is_empty()));
+ }
+
+ #[test]
+ fn empty_rejected() {
+ assert!(StaticPool::new(Vec::<String>::new()).is_err());
+ }
+}
diff --git a/crates/store-memory/Cargo.toml b/crates/store-memory/Cargo.toml
new file mode 100644
index 0000000..6d1533a
--- /dev/null
+++ b/crates/store-memory/Cargo.toml
@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name = "scg-store-memory"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "In-process affinity store for single-replica Spark Connect
Gateway deployments."
+publish = false
+
+[dependencies]
+scg-routing = { path = "../routing" }
+parking_lot = { workspace = true }
+async-trait = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true }
diff --git a/crates/store-memory/src/lib.rs b/crates/store-memory/src/lib.rs
new file mode 100644
index 0000000..9207196
--- /dev/null
+++ b/crates/store-memory/src/lib.rs
@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! In-process `AffinityStore` for single-replica deployments and
+//! tests. For multi-replica HA use `scg-store-redis` — the
+//! `AffinityStore` trait is kept narrow specifically so swapping
+//! the backing store does not ripple into the proxy layer.
+
+use parking_lot::RwLock;
+use scg_routing::{AffinityStore, SessionKey};
+use std::collections::HashMap;
+
+#[derive(Default)]
+pub struct MemoryStore {
+ sessions: RwLock<HashMap<SessionKey, String>>,
+ ops: RwLock<HashMap<String, String>>,
+}
+
+impl MemoryStore {
+ pub fn new() -> Self {
+ Self::default()
+ }
+}
+
+#[async_trait::async_trait]
+impl AffinityStore for MemoryStore {
+ async fn lookup_session(&self, key: &SessionKey) -> Option<String> {
+ self.sessions.read().get(key).cloned()
+ }
+
+ async fn bind_session_if_absent(&self, key: SessionKey, backend: String)
-> String {
+ let mut g = self.sessions.write();
+ g.entry(key).or_insert(backend).clone()
+ }
+
+ async fn forget_session(&self, key: &SessionKey) {
+ self.sessions.write().remove(key);
+ }
+
+ async fn lookup_op(&self, op_id: &str) -> Option<String> {
+ self.ops.read().get(op_id).cloned()
+ }
+
+ async fn bind_op(&self, op_id: String, backend: String) {
+ self.ops.write().insert(op_id, backend);
+ }
+
+ async fn forget_op(&self, op_id: &str) {
+ self.ops.write().remove(op_id);
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn session_stickiness_invariant() {
+ let s = MemoryStore::new();
+ let k = SessionKey::new("alice", "sess-1");
+ assert!(s.lookup_session(&k).await.is_none());
+ let first = s
+ .bind_session_if_absent(k.clone(), "be-a:15002".into())
+ .await;
+ assert_eq!(first, "be-a:15002");
+ // Re-binding must not move an existing session — stickiness invariant.
+ let second = s
+ .bind_session_if_absent(k.clone(), "be-b:15002".into())
+ .await;
+ assert_eq!(second, "be-a:15002");
+ s.forget_session(&k).await;
+ assert!(s.lookup_session(&k).await.is_none());
+ }
+
+ #[tokio::test]
+ async fn op_reverse_index() {
+ let s = MemoryStore::new();
+ assert!(s.lookup_op("op-1").await.is_none());
+ s.bind_op("op-1".into(), "be-a:15002".into()).await;
+ assert_eq!(s.lookup_op("op-1").await.as_deref(), Some("be-a:15002"));
+ s.forget_op("op-1").await;
+ assert!(s.lookup_op("op-1").await.is_none());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
