daqbuf/daqbuf-query
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add query parameter for workaround

Browse Source
This commit is contained in:
Dominik Werder
2025-07-23 23:56:48 +02:00
parent cddc33fcc1
commit 381208a200
2 changed files with 51 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/api4/binned.rs
View File

@@ -143,6 +143,8 @@ pub struct BinnedQuery {
pbd_rts_pbp_block: Option<Vec<Vec<u8>>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pbd_evs: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
use_scylla6_workarounds: Option<u32>,
}
impl BinnedQuery {
@@ -172,6 +174,7 @@ impl BinnedQuery {
pbd_enable: None,
pbd_rts_pbp_block: None,
pbd_evs: None,
use_scylla6_workarounds: None,
}
}
@@ -334,6 +337,10 @@ impl BinnedQuery {
pub fn pbd_evs(&self) -> Option<bool> {
self.pbd_evs.clone()
}
pub fn use_scylla6_workarounds(&self) -> Option<u32> {
self.use_scylla6_workarounds.clone()
}
}
impl HasBackend for BinnedQuery {
@@ -412,6 +419,9 @@ impl FromUrl for BinnedQuery {
.get("pbd_rts_pbp_block")
.and_then(|x| serde_json::from_str(x).ok()),
pbd_evs: pairs.get("pbd_evs").and_then(|x| x.parse().ok()),
use_scylla6_workarounds: pairs
.get("use_scylla6_workarounds")
.and_then(|x| x.parse().ok()),
};
debug!("BinnedQuery::from_url {:?}", ret);
Ok(ret)
@@ -501,6 +511,9 @@ impl AppendToUrl for BinnedQuery {
if let Some(x) = self.pbd_enable.as_ref() {
g.append_pair("pbd_enable", &x.to_string());
}
if let Some(x) = self.use_scylla6_workarounds.as_ref() {
g.append_pair("use_scylla6_workarounds", &x.to_string());
}
}
}
@@ -513,6 +526,8 @@ pub struct BinWriteIndexQuery {
#[serde(default, skip_serializing_if = "Option::is_none")]
rt: Option<RetentionTime>,
pbp: PrebinnedPartitioning,
#[serde(default, skip_serializing_if = "Option::is_none")]
use_scylla6_workarounds: Option<u32>,
}
impl BinWriteIndexQuery {
@@ -535,6 +550,10 @@ impl BinWriteIndexQuery {
pub fn log_level(&self) -> &str {
&self.log_level
}
pub fn use_scylla6_workarounds(&self) -> Option<u32> {
self.use_scylla6_workarounds.clone()
}
}
impl HasBackend for BinWriteIndexQuery {
@@ -568,6 +587,9 @@ impl FromUrl for BinWriteIndexQuery {
.and_then(|x| x.parse().ok())
.and_then(|x| PrebinnedPartitioning::from_db_ix(x).ok())
.unwrap_or(PrebinnedPartitioning::Day1),
use_scylla6_workarounds: pairs
.get("use_scylla6_workarounds")
.and_then(|x| x.parse().ok()),
};
let selfname = std::any::type_name::<Self>();
debug!("{}::from_pairs {:?}", selfname, ret);
@@ -587,5 +609,8 @@ impl AppendToUrl for BinWriteIndexQuery {
g.append_pair("rt", &x.to_index_db_u16().to_string());
}
g.append_pair("pbp", &self.pbp.db_ix().to_string());
if let Some(x) = self.use_scylla6_workarounds.as_ref() {
g.append_pair("use_scylla6_workarounds", &x.to_string());
}
}
}

26
src/api4/events.rs
View File

@@ -83,6 +83,8 @@ pub struct PlainEventsQuery {
#[serde(default)]
use_rt: Option<RetentionTime>,
querymarker: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
use_scylla6_workarounds: Option<u32>,
}
impl PlainEventsQuery {
@@ -112,6 +114,7 @@ impl PlainEventsQuery {
log_level: String::new(),
use_rt: None,
querymarker: String::new(),
use_scylla6_workarounds: None,
}
}
@@ -253,6 +256,10 @@ impl PlainEventsQuery {
pub fn use_rt(&self) -> Option<RetentionTime> {
self.use_rt.clone()
}
pub fn use_scylla6_workarounds(&self) -> Option<u32> {
self.use_scylla6_workarounds.clone()
}
}
impl HasBackend for PlainEventsQuery {
@@ -341,6 +348,9 @@ impl FromUrl for PlainEventsQuery {
querymarker: pairs
.get("querymarker")
.map_or(String::new(), |x| x.to_string()),
use_scylla6_workarounds: pairs
.get("use_scylla6_workarounds")
.and_then(|x| x.parse().ok()),
};
Ok(ret)
}
@@ -407,6 +417,9 @@ impl AppendToUrl for PlainEventsQuery {
if let Some(x) = self.use_rt.as_ref() {
g.append_pair("useRt", &x.to_string());
}
if let Some(x) = self.use_scylla6_workarounds.as_ref() {
g.append_pair("use_scylla6_workarounds", &x.to_string());
}
}
}
@@ -460,6 +473,7 @@ pub struct EventsSubQuerySettings {
use_rt: Option<RetentionTime>,
merger_out_len_max: Option<u32>,
scylla_read_queue_len: Option<u32>,
use_scylla6_workarounds: Option<u32>,
}
impl EventsSubQuerySettings {
@@ -486,6 +500,7 @@ impl Default for EventsSubQuerySettings {
use_rt: None,
merger_out_len_max: None,
scylla_read_queue_len: None,
use_scylla6_workarounds: None,
}
}
}
@@ -505,6 +520,7 @@ impl From<&PlainEventsQuery> for EventsSubQuerySettings {
use_rt: value.use_rt(),
merger_out_len_max: value.merger_out_len_max(),
scylla_read_queue_len: value.scylla_read_queue_len(),
use_scylla6_workarounds: value.use_scylla6_workarounds.clone(),
}
}
}
@@ -525,6 +541,7 @@ impl From<&BinnedQuery> for EventsSubQuerySettings {
use_rt: value.use_rt(),
merger_out_len_max: value.merger_out_len_max(),
scylla_read_queue_len: value.scylla_read_queue_len(),
use_scylla6_workarounds: value.use_scylla6_workarounds().clone(),
}
}
}
@@ -545,6 +562,7 @@ impl From<&Api1Query> for EventsSubQuerySettings {
use_rt: None,
merger_out_len_max: None,
scylla_read_queue_len: None,
use_scylla6_workarounds: None,
}
}
}
@@ -556,6 +574,8 @@ pub struct EventsSubQuery {
ty: String,
reqid: String,
log_level: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
use_scylla6_workarounds: Option<u32>,
}
impl EventsSubQuery {
@@ -565,12 +585,14 @@ impl EventsSubQuery {
reqid: String,
log_level: String,
) -> Self {
let use_scylla6_workarounds = settings.use_scylla6_workarounds.clone();
Self {
select,
settings,
ty: "EventsSubQuery".into(),
reqid,
log_level,
use_scylla6_workarounds,
}
}
@@ -666,6 +688,10 @@ impl EventsSubQuery {
pub fn settings(&self) -> &EventsSubQuerySettings {
&self.settings
}
pub fn use_scylla6_workarounds(&self) -> Option<u32> {
self.use_scylla6_workarounds.clone()
}
}
#[derive(Debug, Serialize, Deserialize)]