Feat[OBE-11872] - Support decompression of data in SCol

Cargo.toml

@@ -178,6 +178,7 @@ rand = { version = "0.8.5", default-features = false, features = ["small_rng"] }
 serde_json = { version = "=1.0.140", default-features = false, features = ["raw_value", "std"] }
 serde = { version = "1.0.215", default-features = false, features = ["alloc", "derive", "rc"] }
 serde_with = { version = "3.12.0", default-features = false, features = ["macros", "std"] }
+serde_yaml = { version = "0.9.34", default-features = false }
 snafu = { version = "0.7.5", default-features = false, features = ["futures", "std"] }
 rusqlite = { version = "0.37.0", features = [ "bundled" ] }
 tokio = { version = "1.43.0", default-features = false, features = ["full"] }
@@ -205,7 +206,7 @@ regex = { version = "1.11.1", default-features = false, features = ["std", "perf
 tokio-util = { version = "0.7", default-features = false, features = ["io", "time"] }
 bytesize = { version = "1.3.0", default-features = false }
 bytes = { version = "1.9.0", default-features = false, features = ["serde"] }
-async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd"] }
+async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd", "zlib", "deflate"] }
 object_store = { version = "0.10.2", features = ["gcp"], default-features = false }
 once_cell = { version = "1.20.2", default-features = false }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }

lib/observo/scol/Cargo.toml

@@ -66,6 +66,8 @@ reqwest = { version = "0.12", default-features = false }
 # core). Pull hyper 1 under an alias so retry classification can downcast `object_store::Error`
 # sources to the *matching* `hyper::Error` type (see objstore::is_retriable).
 hyper1 = { package = "hyper", version = "1", default-features = false }
+async-compression = { workspace = true }
+snap.workspace = true
 
 # Optional dependencies for testing (below)
 testcontainers = { version = "0.19", optional = true }
@@ -92,10 +94,13 @@ regex = "1"
 serde_urlencoded = "0.7"
 tempfile = "3.15.0"
 tracing-test = { version = "0.2.5", features = [ "no-env-filter" ] }
+serde_yaml = { workspace = true }
 test-case = { version = "3.3" }
 tokio = { version = "1.43.0", features = ["test-util"] }
 scol = { path = ".", features = ["test-scenarios"] }
 chkpts = { path = "../chkpts" }
+flate2 = { workspace = true }
+zstd = { workspace = true }
 
 [features]
 test-scenarios = ["vector-lib/test", "dep:textwrap", "dep:rstest", "dep:oxide-auth",

lib/vector-common/src/chkpts.rs

@@ -39,6 +39,24 @@ pub enum ChkptErr {
     Unknown(crate::Error)
 }
 
+/// Precondition guarding a `compare_and_set`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SetCond<'a> {
+    /// The checkpoint must exist and its latest value must equal this.
+    Old(Cow<'a, str>),
+    /// The checkpoint must not exist.
+    Absent,
+}
+
+impl SetCond<'_> {
+    pub fn into_owned(self) -> SetCond<'static> {
+        match self {
+            SetCond::Old(v) => SetCond::Old(Cow::Owned(v.into_owned())),
+            SetCond::Absent => SetCond::Absent,
+        }
+    }
+}
+
 impl Display for ChkptErr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{self:?}")
@@ -58,7 +76,7 @@ impl Error for ChkptErr {
 pub trait Accessor: Send + dyn_clone::DynClone + Sync {
     async fn get(&self, id: Cow<'_, str>) -> Result<Value, ChkptErr>;
     async fn set(&self, id: Cow<'_, str>, value: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
-    async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, if_old: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
+    async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, cond: SetCond<'_>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn del(&self, id: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn compare_and_del(&self, id: Cow<'_, str>, if_old: Cow<'_, str>) -> Result<(), ChkptErr>;
     async fn del_range(&self, from: Cow<'_, str>, to: Cow<'_, str>) -> Result<u64, ChkptErr>;

src/sources/scol/mod.rs

@@ -23,7 +23,7 @@ impl SourceConfig for Config {
         let chkptr = cx.checkpoint_accessor().await;
         let src = self
             .clone()
-            .build_source(cx.out, cx.shutdown, chkptr, lns)
+            .build_source(cx.out, cx.shutdown, chkptr, lns, cx.globals.limits.compression)
             .map(|r| match r {
                 Ok(_) => Ok(()),
                 Err(e) => {