starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 00:09:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: broadcast event use file id (#6891)

Browse Source 
- [x] fix: broadcast event use file id
- [x] feat: add metrics for consistent hashing
This commit is contained in:
Hengfei Yang
2025-05-23 16:56:06 +08:00
committed by GitHub
parent 9c585d73a4
commit b058106166
26 changed files with 517 additions and 454 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli/basic/cli.rs
+16
View File
@@ -231,6 +231,14 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
.required(true)
.help("snowflake id"),
]),
clap::Command::new("consistent-hash").about("consistent hash").args([
clap::Arg::new("file")
.short('f')
.long("file")
.required(true)
.num_args(1..)
.help("file"),
]),
clap::Command::new("upgrade-db")
.about("upgrade db table schemas").args(dataArgs()),
])
@@ -495,6 +503,14 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
let td = t.format("%Y-%m-%dT%H:%M:%SZ").to_string();
println!("datetimes: {}", td);
}
"consistent-hash" => {
let files = command
.get_many::<String>("file")
.unwrap()
.collect::<Vec<_>>();
let files = files.iter().map(|f| f.to_string()).collect::<Vec<_>>();
super::http::consistent_hash(files).await?;
}
"upgrade-db" => {
crate::migration::init_db().await?;
}
src/cli/basic/http.rs
+34 -3
View File
@@ -96,6 +96,7 @@ pub async fn node_offline() -> Result<(), anyhow::Error> {
}
Ok(())
}
pub async fn node_online() -> Result<(), anyhow::Error> {
let url = "/node/enable?value=true";
let response = request(url, None, reqwest::Method::PUT).await?;
@@ -262,6 +263,35 @@ pub async fn local_node_metrics() -> Result<(), anyhow::Error> {
Ok(())
}
pub async fn consistent_hash(files: Vec<String>) -> Result<(), anyhow::Error> {
let url = "/node/consistent_hash";
let req = config::meta::search::HashFileRequest { files };
let body = serde_json::to_vec(&req)?;
let response = request(url, Some(body), reqwest::Method::POST).await?;
let Some(body) = response else {
return Err(anyhow::anyhow!("consistent hash failed"));
};
let response: config::meta::search::HashFileResponse = serde_json::from_str(&body)?;
// Create header row with all column names
let mut table = Table::new();
table.add_row(Row::new(vec![
Cell::new("File"),
Cell::new("Querier"),
Cell::new("Alert Querier"),
]));
for (file, nodes) in response.files.iter() {
table.add_row(Row::new(vec![
Cell::new(file),
Cell::new(nodes.get("querier_interactive").unwrap_or(&"".to_string())),
Cell::new(nodes.get("querier_background").unwrap_or(&"".to_string())),
]));
}
table.printstd();
Ok(())
}
async fn request(
url: &str,
body: Option<Vec<u8>>,
@@ -273,11 +303,12 @@ async fn request(
let url = format!("{}{}", local.http_addr, url);
let user = cfg.auth.root_user_email.clone();
let password = cfg.auth.root_user_password.clone();
let mut response = client.request(method, url).basic_auth(user, Some(password));
let mut req = client.request(method, url).basic_auth(user, Some(password));
if let Some(body) = body {
response = response.body(body);
req = req.header("Content-Type", "application/json");
req = req.body(body);
}
let response = response.send().await?;
let response = req.send().await?;
let status = response.status();
let body = response.text().await?;
if status.is_success() {
src/common/infra/cluster/etcd.rs
+1 -1
View File
@@ -250,7 +250,7 @@ pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result
};
// update node status metrics
node.metrics = super::update_node_status_metrics();
node.metrics = super::update_node_status_metrics().await;
let val = json::to_string(&node).unwrap();
src/common/infra/cluster/mod.rs
+25 -1
View File
@@ -163,6 +163,22 @@ pub async fn print_consistent_hash() -> HashMap<String, HashMap<String, Vec<u64>
map
}
pub async fn count_consistent_hash() -> HashMap<String, usize> {
let mut map = HashMap::new();
let r = QUERIER_INTERACTIVE_CONSISTENT_HASH.read().await;
map.insert("querier_interactive".to_string(), r.len());
drop(r);
let r = QUERIER_BACKGROUND_CONSISTENT_HASH.read().await;
map.insert("querier_background".to_string(), r.len());
drop(r);
let r = COMPACTOR_CONSISTENT_HASH.read().await;
map.insert("compactor".to_string(), r.len());
drop(r);
let r = FLATTEN_COMPACTOR_CONSISTENT_HASH.read().await;
map.insert("flatten_compactor".to_string(), r.len());
map
}
/// Register and keep alive the node to cluster
pub async fn register_and_keep_alive() -> Result<()> {
let cfg = get_config();
@@ -616,7 +632,7 @@ async fn set_node_status_metrics(node: &Node) {
}
}
fn update_node_status_metrics() -> NodeMetrics {
async fn update_node_status_metrics() -> NodeMetrics {
let node_status = get_node_metrics();
config::metrics::NODE_UP
@@ -650,6 +666,14 @@ fn update_node_status_metrics() -> NodeMetrics {
.with_label_values(&["resets"])
.set(node_status.tcp_conns_resets as i64);
// update node consistent hash metrics
let consistent_hash = count_consistent_hash().await;
for (k, v) in consistent_hash {
config::metrics::NODE_CONSISTENT_HASH
.with_label_values(&[k.as_str()])
.set(v as i64);
}
node_status
}
src/common/infra/cluster/nats.rs
+1 -1
View File
@@ -260,7 +260,7 @@ pub(crate) async fn set_status(status: NodeStatus) -> Result<()> {
};
// update node status metrics
node.metrics = super::update_node_status_metrics();
node.metrics = super::update_node_status_metrics().await;
let val = json::to_string(&node).unwrap();
src/config/src/meta/search.rs
+11
View File
@@ -16,6 +16,7 @@
use std::collections::VecDeque;
use bytes::Bytes as BytesImpl;
use hashbrown::HashMap;
use proto::cluster_rpc;
use serde::{Deserialize, Deserializer, Serialize};
use utoipa::ToSchema;
@@ -1203,6 +1204,16 @@ pub struct ValuesRequest {
pub sql: String,
}
#[derive(Debug, Deserialize, Clone, Serialize)]
pub struct HashFileRequest {
pub files: Vec<String>,
}
#[derive(Debug, Default, Deserialize, Clone, Serialize)]
pub struct HashFileResponse {
pub files: HashMap<String, HashMap<String, String>>,
}
#[cfg(test)]
mod tests {
use super::*;
src/config/src/meta/stream.rs
+6 -1
View File
@@ -168,6 +168,7 @@ pub struct ListStreamParams {
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct FileKey {
pub id: i64,
pub account: String,
pub key: String,
pub meta: FileMeta,
@@ -176,8 +177,9 @@ pub struct FileKey {
}
impl FileKey {
pub fn new(account: String, key: String, meta: FileMeta, deleted: bool) -> Self {
pub fn new(id: i64, account: String, key: String, meta: FileMeta, deleted: bool) -> Self {
Self {
id,
account,
key,
meta,
@@ -188,6 +190,7 @@ impl FileKey {
pub fn from_file_name(file: &str) -> Self {
Self {
id: 0,
account: String::default(),
key: file.to_string(),
meta: FileMeta::default(),
@@ -474,6 +477,7 @@ impl From<&cluster_rpc::FileMeta> for FileMeta {
impl From<&FileKey> for cluster_rpc::FileKey {
fn from(req: &FileKey) -> Self {
cluster_rpc::FileKey {
id: req.id,
account: req.account.clone(),
key: req.key.clone(),
meta: Some(cluster_rpc::FileMeta::from(&req.meta)),
@@ -486,6 +490,7 @@ impl From<&FileKey> for cluster_rpc::FileKey {
impl From<&cluster_rpc::FileKey> for FileKey {
fn from(req: &cluster_rpc::FileKey) -> Self {
FileKey {
id: req.id,
account: req.account.clone(),
key: req.key.clone(),
meta: FileMeta::from(req.meta.as_ref().unwrap()),
src/config/src/metrics.rs
+13
View File
@@ -789,6 +789,16 @@ pub static NODE_TCP_CONNECTIONS: Lazy<IntGaugeVec> = Lazy::new(|| {
.expect("Metric created")
});
pub static NODE_CONSISTENT_HASH: Lazy<IntGaugeVec> = Lazy::new(|| {
IntGaugeVec::new(
Opts::new("node_consistent_hash", "Consistent hash")
.namespace(NAMESPACE)
.const_labels(create_const_labels()),
&["type"],
)
.expect("Metric created")
});
// query disk cache metrics
pub static QUERY_DISK_CACHE_HIT_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
IntCounterVec::new(
@@ -1052,6 +1062,9 @@ fn register_metrics(registry: &Registry) {
registry
.register(Box::new(NODE_TCP_CONNECTIONS.clone()))
.expect("Metric registered");
registry
.register(Box::new(NODE_CONSISTENT_HASH.clone()))
.expect("Metric registered");
// query disk cache metrics
registry
src/handler/http/request/status/mod.rs
+43 -8
View File
@@ -20,20 +20,23 @@ use actix_web::{
cookie::{Cookie, SameSite},
get, head,
http::header,
put, web,
post, put, web,
};
use arrow_schema::Schema;
use config::{
Config, META_ORG_ID, QUICK_MODEL_FIELDS, SQL_FULL_TEXT_SEARCH_FIELDS, TIMESTAMP_COL_NAME,
cluster::LOCAL_NODE,
get_config, get_instance_id,
meta::{cluster::NodeStatus, function::ZoFunction},
meta::{
cluster::{NodeStatus, Role, RoleGroup},
function::ZoFunction,
search::{HashFileRequest, HashFileResponse},
},
utils::{base64, json, schema_ext::SchemaExt},
};
use hashbrown::HashMap;
use infra::{
cache::{self, file_data::disk::FileType},
file_list,
cache, file_list,
schema::{STREAM_SCHEMAS, STREAM_SCHEMAS_LATEST},
};
use serde::Serialize;
@@ -349,11 +352,13 @@ pub async fn cache_status() -> Result<HttpResponse, Error> {
let mem_file_num = cache::file_data::memory::len().await;
let (mem_max_size, mem_cur_size) = cache::file_data::memory::stats().await;
let disk_file_num = cache::file_data::disk::len(FileType::DATA).await;
let (disk_max_size, disk_cur_size) = cache::file_data::disk::stats(FileType::DATA).await;
let disk_result_file_num = cache::file_data::disk::len(FileType::RESULT).await;
let disk_file_num = cache::file_data::disk::len(cache::file_data::disk::FileType::DATA).await;
let (disk_max_size, disk_cur_size) =
cache::file_data::disk::stats(cache::file_data::disk::FileType::DATA).await;
let disk_result_file_num =
cache::file_data::disk::len(cache::file_data::disk::FileType::RESULT).await;
let (disk_result_max_size, disk_result_cur_size) =
cache::file_data::disk::stats(FileType::RESULT).await;
cache::file_data::disk::stats(cache::file_data::disk::FileType::RESULT).await;
stats.insert(
"FILE_DATA",
json::json!({
@@ -820,3 +825,33 @@ async fn node_metrics() -> Result<HttpResponse, Error> {
let metrics = config::utils::sysinfo::get_node_metrics();
Ok(MetaHttpResponse::json(metrics))
}
#[post("/consistent_hash")]
async fn consistent_hash(body: web::Json<HashFileRequest>) -> Result<HttpResponse, Error> {
let mut ret = HashFileResponse::default();
for file in body.files.iter() {
let mut nodes = HashMap::new();
nodes.insert(
"querier_interactive".to_string(),
cluster::get_node_from_consistent_hash(
file,
&Role::Querier,
Some(RoleGroup::Interactive),
)
.await
.unwrap_or_default(),
);
nodes.insert(
"querier_background".to_string(),
cluster::get_node_from_consistent_hash(
file,
&Role::Querier,
Some(RoleGroup::Background),
)
.await
.unwrap_or_default(),
);
ret.files.insert(file.clone(), nodes);
}
Ok(MetaHttpResponse::json(ret))
}
src/handler/http/router/mod.rs
+2 -1
View File
@@ -261,7 +261,8 @@ pub fn get_basic_routes(svc: &mut web::ServiceConfig) {
.service(status::enable_node)
.service(status::flush_node)
.service(status::list_node)
.service(status::node_metrics),
.service(status::node_metrics)
.service(status::consistent_hash),
);
if get_config().common.swagger_enabled {
src/infra/src/file_list/mod.rs
+48 -22
View File
@@ -52,11 +52,11 @@ pub fn connect_local_cache() -> Box<dyn FileList> {
pub trait FileList: Sync + Send + 'static {
async fn create_table(&self) -> Result<()>;
async fn create_table_index(&self) -> Result<()>;
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()>;
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()>;
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64>;
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64>;
async fn remove(&self, file: &str) -> Result<()>;
async fn batch_add(&self, files: &[FileKey]) -> Result<()>;
async fn batch_add_with_id(&self, files: &[(i64, &FileKey)]) -> Result<()>;
async fn batch_add_with_id(&self, files: &[FileKey]) -> Result<()>;
async fn batch_add_history(&self, files: &[FileKey]) -> Result<()>;
async fn update_dump_records(&self, dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()>;
async fn batch_process(&self, files: &[FileKey]) -> Result<()>;
@@ -71,7 +71,7 @@ pub trait FileList: Sync + Send + 'static {
async fn contains(&self, file: &str) -> Result<bool>;
async fn update_flattened(&self, file: &str, flattened: bool) -> Result<()>;
async fn update_compressed_size(&self, file: &str, size: i64) -> Result<()>;
async fn list(&self) -> Result<Vec<(String, String, FileMeta)>>;
async fn list(&self) -> Result<Vec<FileKey>>;
async fn query(
&self,
org_id: &str,
@@ -80,15 +80,15 @@ pub trait FileList: Sync + Send + 'static {
time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, String, FileMeta)>>;
) -> Result<Vec<FileKey>>;
async fn query_for_merge(
&self,
org_id: &str,
stream_type: StreamType,
stream_name: &str,
date_range: Option<(String, String)>,
) -> Result<Vec<(String, String, FileMeta)>>;
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<(i64, String, String, FileMeta)>>;
) -> Result<Vec<FileKey>>;
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<FileKey>>;
async fn query_ids(
&self,
org_id: &str,
@@ -96,6 +96,7 @@ pub trait FileList: Sync + Send + 'static {
stream_name: &str,
time_range: Option<(i64, i64)>,
) -> Result<Vec<FileId>>;
async fn query_ids_by_files(&self, files: &[FileKey]) -> Result<stdHashMap<String, i64>>;
async fn query_old_data_hours(
&self,
org_id: &str,
@@ -192,12 +193,12 @@ pub async fn create_table_index() -> Result<()> {
}
#[inline]
pub async fn add(account: &str, file: &str, meta: &FileMeta) -> Result<()> {
pub async fn add(account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
CLIENT.add(account, file, meta).await
}
#[inline]
pub async fn add_history(account: &str, file: &str, meta: &FileMeta) -> Result<()> {
pub async fn add_history(account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
CLIENT.add_history(account, file, meta).await
}
@@ -261,7 +262,7 @@ pub async fn update_compressed_size(file: &str, size: i64) -> Result<()> {
}
#[inline]
pub async fn list() -> Result<Vec<(String, String, FileMeta)>> {
pub async fn list() -> Result<Vec<FileKey>> {
CLIENT.list().await
}
@@ -274,7 +275,7 @@ pub async fn query(
time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
validate_time_range(time_range)?;
CLIENT
.query(
@@ -295,7 +296,7 @@ pub async fn query_for_merge(
stream_type: StreamType,
stream_name: &str,
date_range: Option<(String, String)>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
CLIENT
.query_for_merge(org_id, stream_type, stream_name, date_range)
.await
@@ -303,7 +304,7 @@ pub async fn query_for_merge(
#[inline]
#[tracing::instrument(name = "infra:file_list:query_db_by_ids", skip_all)]
pub async fn query_by_ids(ids: &[i64]) -> Result<Vec<(i64, String, String, FileMeta)>> {
pub async fn query_by_ids(ids: &[i64]) -> Result<Vec<FileKey>> {
CLIENT.query_by_ids(ids).await
}
@@ -321,6 +322,12 @@ pub async fn query_ids(
.await
}
#[inline]
#[tracing::instrument(name = "infra:file_list:db:query_ids_by_files")]
pub async fn query_ids_by_files(files: &[FileKey]) -> Result<stdHashMap<String, i64>> {
CLIENT.query_ids_by_files(files).await
}
#[inline]
#[tracing::instrument(name = "infra:file_list:db:query_old_data_hours")]
pub async fn query_old_data_hours(
@@ -572,16 +579,29 @@ pub struct FileRecord {
pub flattened: bool,
}
impl From<&FileRecord> for FileMeta {
fn from(record: &FileRecord) -> Self {
impl From<&FileRecord> for FileKey {
fn from(r: &FileRecord) -> Self {
Self {
min_ts: record.min_ts,
max_ts: record.max_ts,
records: record.records,
original_size: record.original_size,
compressed_size: record.compressed_size,
index_size: record.index_size,
flattened: record.flattened,
id: r.id,
account: r.account.to_string(),
key: "files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
meta: r.into(),
deleted: r.deleted,
segment_ids: None,
}
}
}
impl From<&FileRecord> for FileMeta {
fn from(r: &FileRecord) -> Self {
Self {
min_ts: r.min_ts,
max_ts: r.max_ts,
records: r.records,
original_size: r.original_size,
compressed_size: r.compressed_size,
index_size: r.index_size,
flattened: r.flattened,
}
}
}
@@ -655,3 +675,9 @@ pub struct FileId {
#[sqlx(default)]
pub deleted: bool,
}
#[derive(Clone, Debug, Default, sqlx::FromRow)]
pub struct FileIdWithFile {
pub id: i64,
pub file: String,
}
src/infra/src/file_list/mysql.rs
+60 -47
View File
@@ -63,11 +63,11 @@ impl super::FileList for MysqlFileList {
create_table_index().await
}
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list", account, file, meta).await
}
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list_history", account, file, meta)
.await
}
@@ -94,7 +94,7 @@ impl super::FileList for MysqlFileList {
self.inner_batch_process("file_list", files).await
}
async fn batch_add_with_id(&self, _files: &[(i64, &FileKey)]) -> Result<()> {
async fn batch_add_with_id(&self, _files: &[FileKey]) -> Result<()> {
unimplemented!("Unsupported")
}
@@ -361,7 +361,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
Ok(())
}
async fn list(&self) -> Result<Vec<(String, String, FileMeta)>> {
async fn list(&self) -> Result<Vec<FileKey>> {
return Ok(vec![]); // disallow list all data
}
@@ -373,7 +373,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
_time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = time_range {
if start == 0 && end == 0 {
return Ok(Vec::new());
@@ -390,7 +390,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
let ret = if flattened.is_some() {
sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = ? AND flattened = ? LIMIT 1000;
"#,
@@ -404,7 +404,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let max_ts_upper_bound = super::calculate_max_ts_upper_bound(time_end, stream_type);
sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = ? AND max_ts >= ? AND max_ts <= ? AND min_ts <= ?;
"#,
@@ -422,17 +422,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.observe(time);
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
@@ -442,7 +432,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
stream_type: StreamType,
stream_name: &str,
date_range: Option<(String, String)>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = date_range.as_ref() {
if start.is_empty() && end.is_empty() {
return Ok(Vec::new());
@@ -459,7 +449,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let (date_start, date_end) = date_range.unwrap_or(("".to_string(), "".to_string()));
let ret = sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = ? AND date >= ? AND date <= ?;
"#,
@@ -475,21 +465,11 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.observe(time);
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<(i64, String, String, FileMeta)>> {
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<FileKey>> {
if ids.is_empty() {
return Ok(Vec::default());
}
@@ -522,17 +502,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
ret.extend_from_slice(&res?);
}
Ok(ret
.into_iter()
.map(|r| {
(
r.id,
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
FileMeta::from(&r),
)
})
.collect())
Ok(ret.iter().map(|r| r.into()).collect())
}
async fn query_ids(
@@ -587,7 +557,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.bind(time_end)
.fetch_all(&pool)
.await
}));
}));
}
let mut rets = Vec::new();
@@ -609,6 +579,45 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
Ok(rets)
}
async fn query_ids_by_files(&self, files: &[FileKey]) -> Result<stdHashMap<String, i64>> {
let mut ret = stdHashMap::with_capacity(files.len());
// group by date
let mut stream_files = HashMap::new();
let mut files_map = HashMap::with_capacity(files.len());
for file in files {
if file.id > 0 {
ret.insert(file.key.clone(), file.id);
continue;
}
let (stream_key, date_key, file_name) =
parse_file_key_columns(&file.key).map_err(|e| Error::Message(e.to_string()))?;
let stream_entry = stream_files.entry(stream_key).or_insert(HashMap::new());
let date_entry = stream_entry.entry(date_key).or_insert(Vec::new());
date_entry.push(file_name.clone());
files_map.insert(file_name, &file.key);
}
for (stream_key, stream_files) in stream_files {
let pool = CLIENT_RO.clone();
for (date_key, files) in stream_files {
let sql = format!(
"SELECT id, file FROM file_list WHERE stream = ? AND date = ? AND file IN ({});",
files.join(",")
);
let query_res = sqlx::query_as::<_, super::FileIdWithFile>(&sql)
.bind(&stream_key)
.bind(&date_key)
.fetch_all(&pool)
.await?;
for file in query_res {
if let Some(file_name) = files_map.get(&file.file) {
ret.insert(file_name.to_string(), file.id);
}
}
}
}
Ok(ret)
}
async fn query_old_data_hours(
&self,
org_id: &str,
@@ -1521,7 +1530,7 @@ impl MysqlFileList {
account: &str,
file: &str,
meta: &FileMeta,
) -> Result<()> {
) -> Result<i64> {
let pool = CLIENT.clone();
let (stream_key, date_key, file_name) =
parse_file_key_columns(file).map_err(|e| Error::Message(e.to_string()))?;
@@ -1551,12 +1560,12 @@ INSERT IGNORE INTO {table} (account, org, stream, date, file, deleted, min_ts, m
.execute(&pool)
.await {
Err(sqlx::Error::Database(e)) => if e.is_unique_violation() {
Ok(())
Ok(0)
} else {
Err(Error::Message(e.to_string()))
},
Err(e) => Err(e.into()),
Ok(_) => Ok(()),
Ok(v) => Ok(v.last_insert_id() as i64),
}
}
@@ -1615,6 +1624,10 @@ INSERT IGNORE INTO {table} (account, org, stream, date, file, deleted, min_ts, m
// get ids of the files
let mut ids = Vec::with_capacity(files.len());
for file in files {
if file.id > 0 {
ids.push(file.id.to_string());
continue;
}
let (stream_key, date_key, file_name) = parse_file_key_columns(&file.key)
.map_err(|e| Error::Message(e.to_string()))?;
DB_QUERY_NUMS
src/infra/src/file_list/postgres.rs
+87 -73
View File
@@ -63,11 +63,11 @@ impl super::FileList for PostgresFileList {
create_table_index().await
}
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list", account, file, meta).await
}
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list_history", account, file, meta)
.await
}
@@ -95,7 +95,7 @@ impl super::FileList for PostgresFileList {
self.inner_batch_process("file_list", files).await
}
async fn batch_add_with_id(&self, _files: &[(i64, &FileKey)]) -> Result<()> {
async fn batch_add_with_id(&self, _files: &[FileKey]) -> Result<()> {
unimplemented!("Unsupported")
}
@@ -374,7 +374,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
Ok(())
}
async fn list(&self) -> Result<Vec<(String, String, FileMeta)>> {
async fn list(&self) -> Result<Vec<FileKey>> {
return Ok(vec![]); // disallow list all data
}
@@ -386,7 +386,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
_time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = time_range {
if start == 0 && end == 0 {
return Ok(Vec::new());
@@ -403,7 +403,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
let ret = if flattened.is_some() {
sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND flattened = $2 LIMIT 1000;
"#
@@ -415,7 +415,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let (time_start, time_end) = time_range.unwrap_or((0, 0));
let max_ts_upper_bound = super::calculate_max_ts_upper_bound(time_end, stream_type);
let sql = r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4;
"#;
@@ -434,17 +434,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.observe(time);
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
@@ -454,7 +444,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
stream_type: StreamType,
stream_name: &str,
date_range: Option<(String, String)>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = date_range.as_ref() {
if start.is_empty() && end.is_empty() {
return Ok(Vec::new());
@@ -471,7 +461,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let (date_start, date_end) = date_range.unwrap_or(("".to_string(), "".to_string()));
let sql = r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND date >= $2 AND date <= $3;
"#;
@@ -488,21 +478,11 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.observe(time);
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<(i64, String, String, FileMeta)>> {
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<FileKey>> {
if ids.is_empty() {
return Ok(Vec::default());
}
@@ -535,17 +515,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
ret.extend_from_slice(&res?);
}
Ok(ret
.into_iter()
.map(|r| {
(
r.id,
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
FileMeta::from(&r),
)
})
.collect())
Ok(ret.iter().map(|r| r.into()).collect())
}
async fn query_ids(
@@ -622,6 +592,45 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
Ok(rets)
}
async fn query_ids_by_files(&self, files: &[FileKey]) -> Result<stdHashMap<String, i64>> {
let mut ret = stdHashMap::with_capacity(files.len());
// group by date
let mut stream_files = HashMap::new();
let mut files_map = HashMap::with_capacity(files.len());
for file in files {
if file.id > 0 {
ret.insert(file.key.clone(), file.id);
continue;
}
let (stream_key, date_key, file_name) =
parse_file_key_columns(&file.key).map_err(|e| Error::Message(e.to_string()))?;
let stream_entry = stream_files.entry(stream_key).or_insert(HashMap::new());
let date_entry = stream_entry.entry(date_key).or_insert(Vec::new());
date_entry.push(file_name.clone());
files_map.insert(file_name, &file.key);
}
for (stream_key, stream_files) in stream_files {
let pool = CLIENT_RO.clone();
for (date_key, files) in stream_files {
let sql = format!(
"SELECT id, file FROM file_list WHERE stream = $1 AND date = $2 AND file IN ({});",
files.join(",")
);
let query_res = sqlx::query_as::<_, super::FileIdWithFile>(&sql)
.bind(&stream_key)
.bind(&date_key)
.fetch_all(&pool)
.await?;
for file in query_res {
if let Some(file_name) = files_map.get(&file.file) {
ret.insert(file_name.to_string(), file.id);
}
}
}
}
Ok(ret)
}
async fn query_old_data_hours(
&self,
org_id: &str,
@@ -1469,7 +1478,7 @@ impl PostgresFileList {
account: &str,
file: &str,
meta: &FileMeta,
) -> Result<()> {
) -> Result<i64> {
let pool = CLIENT.clone();
let (stream_key, date_key, file_name) =
parse_file_key_columns(file).map_err(|e| Error::Message(e.to_string()))?;
@@ -1477,39 +1486,40 @@ impl PostgresFileList {
DB_QUERY_NUMS
.with_label_values(&["insert", table, ""])
.inc();
match
sqlx
::query(
format!(
r#"
let ret: std::result::Result<Option<i64>, sea_orm::SqlxError> = sqlx::query_scalar(
format!(
r#"
INSERT INTO {table} (account, org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
ON CONFLICT DO NOTHING;
"#
).as_str()
)
.bind(account)
.bind(org_id)
.bind(stream_key)
.bind(date_key)
.bind(file_name)
.bind(false)
.bind(meta.min_ts)
.bind(meta.max_ts)
.bind(meta.records)
.bind(meta.original_size)
.bind(meta.compressed_size)
.bind(meta.index_size)
.bind(meta.flattened)
.execute(&pool).await
{
Err(sqlx::Error::Database(e)) => if e.is_unique_violation() {
Ok(())
} else {
Err(Error::Message(e.to_string()))
ON CONFLICT DO NOTHING
RETURNING id;
"#
).as_str()
)
.bind(account)
.bind(org_id)
.bind(stream_key)
.bind(date_key)
.bind(file_name)
.bind(false)
.bind(meta.min_ts)
.bind(meta.max_ts)
.bind(meta.records)
.bind(meta.original_size)
.bind(meta.compressed_size)
.bind(meta.index_size)
.bind(meta.flattened)
.fetch_one(&pool).await;
match ret {
Err(sqlx::Error::Database(e)) => {
if e.is_unique_violation() {
Ok(0)
} else {
Err(Error::Message(e.to_string()))
}
}
Err(e) => Err(e.into()),
Ok(_) => Ok(()),
Ok(v) => Ok(v.unwrap_or(0)),
}
}
@@ -1568,6 +1578,10 @@ INSERT INTO {table} (account, org, stream, date, file, deleted, min_ts, max_ts,
// get ids of the files
let mut ids = Vec::with_capacity(files.len());
for file in files {
if file.id > 0 {
ids.push(file.id.to_string());
continue;
}
let (stream_key, date_key, file_name) = parse_file_key_columns(&file.key)
.map_err(|e| Error::Message(e.to_string()))?;
DB_QUERY_NUMS
src/infra/src/file_list/sqlite.rs
+68 -76
View File
@@ -61,11 +61,11 @@ impl super::FileList for SqliteFileList {
create_table_index().await
}
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list", account, file, meta).await
}
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<()> {
async fn add_history(&self, account: &str, file: &str, meta: &FileMeta) -> Result<i64> {
self.inner_add("file_list_history", account, file, meta)
.await
}
@@ -92,12 +92,8 @@ impl super::FileList for SqliteFileList {
self.inner_batch_process("file_list", files).await
}
async fn batch_add_with_id(&self, files: &[(i64, &FileKey)]) -> Result<()> {
let files = files
.iter()
.map(|(id, f)| (Some(*id), *f))
.collect::<Vec<_>>();
self.inner_batch_process_with_id("file_list", &files).await
async fn batch_add_with_id(&self, files: &[FileKey]) -> Result<()> {
self.inner_batch_process("file_list", files).await
}
async fn batch_add_history(&self, files: &[FileKey]) -> Result<()> {
@@ -329,23 +325,14 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
Ok(())
}
async fn list(&self) -> Result<Vec<(String, String, FileMeta)>> {
async fn list(&self) -> Result<Vec<FileKey>> {
let pool = CLIENT_RO.clone();
let ret = sqlx::query_as::<_, super::FileRecord>(
r#"SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened FROM file_list;"#,
r#"SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened FROM file_list;"#,
)
.fetch_all(&pool)
.await?;
Ok(ret
.iter()
.map(|r| {
(
r.account.to_string(),
format!("files/{}/{}/{}", r.stream, r.date, r.file),
r.into(),
)
})
.collect())
Ok(ret.iter().map(|r| r.into()).collect())
}
async fn query(
@@ -356,7 +343,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
_time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = time_range {
if start == 0 && end == 0 {
return Ok(Vec::new());
@@ -369,7 +356,7 @@ SELECT min_ts, max_ts, records, original_size, compressed_size, index_size, flat
let ret = if flattened.is_some() {
sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND flattened = $2 LIMIT 1000;
"#,
@@ -383,7 +370,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let max_ts_upper_bound = super::calculate_max_ts_upper_bound(time_end, stream_type);
sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4;
"#,
@@ -397,17 +384,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
};
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
@@ -417,7 +394,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
stream_type: StreamType,
stream_name: &str,
date_range: Option<(String, String)>,
) -> Result<Vec<(String, String, FileMeta)>> {
) -> Result<Vec<FileKey>> {
if let Some((start, end)) = date_range.as_ref() {
if start.is_empty() && end.is_empty() {
return Ok(Vec::new());
@@ -430,7 +407,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
let (date_start, date_end) = date_range.unwrap_or(("".to_string(), "".to_string()));
let ret = sqlx::query_as::<_, super::FileRecord>(
r#"
SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
SELECT id, account, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened
FROM file_list
WHERE stream = $1 AND date >= $2 AND date <= $3;
"#,
@@ -442,21 +419,11 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
.await;
Ok(ret?
.iter()
.filter_map(|r| {
if r.deleted {
None
} else {
Some((
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
))
}
})
.filter_map(|r| if r.deleted { None } else { Some(r.into()) })
.collect())
}
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<(i64, String, String, FileMeta)>> {
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<FileKey>> {
if ids.is_empty() {
return Ok(Vec::default());
}
@@ -482,17 +449,7 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
ret.extend_from_slice(&res);
}
Ok(ret
.iter()
.map(|r| {
(
r.id,
r.account.to_string(),
"files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
r.into(),
)
})
.collect())
Ok(ret.iter().map(|r| r.into()).collect())
}
async fn query_ids(
@@ -561,6 +518,45 @@ SELECT account, stream, date, file, deleted, min_ts, max_ts, records, original_s
Ok(rets)
}
async fn query_ids_by_files(&self, files: &[FileKey]) -> Result<stdHashMap<String, i64>> {
let mut ret = stdHashMap::with_capacity(files.len());
// group by date
let mut stream_files = HashMap::new();
let mut files_map = HashMap::with_capacity(files.len());
for file in files {
if file.id > 0 {
ret.insert(file.key.clone(), file.id);
continue;
}
let (stream_key, date_key, file_name) =
parse_file_key_columns(&file.key).map_err(|e| Error::Message(e.to_string()))?;
let stream_entry = stream_files.entry(stream_key).or_insert(HashMap::new());
let date_entry = stream_entry.entry(date_key).or_insert(Vec::new());
date_entry.push(file_name.clone());
files_map.insert(file_name, &file.key);
}
for (stream_key, stream_files) in stream_files {
let pool = CLIENT_RO.clone();
for (date_key, files) in stream_files {
let sql = format!(
"SELECT id, file FROM file_list WHERE stream = $1 AND date = $2 AND file IN ({});",
files.join(",")
);
let query_res = sqlx::query_as::<_, super::FileIdWithFile>(&sql)
.bind(&stream_key)
.bind(&date_key)
.fetch_all(&pool)
.await?;
for file in query_res {
if let Some(file_name) = files_map.get(&file.file) {
ret.insert(file_name.to_string(), file.id);
}
}
}
}
Ok(ret)
}
async fn query_old_data_hours(
&self,
org_id: &str,
@@ -1299,7 +1295,7 @@ impl SqliteFileList {
account: &str,
file: &str,
meta: &FileMeta,
) -> Result<()> {
) -> Result<i64> {
self.inner_add_with_id(table, None, account, file, meta)
.await
}
@@ -1311,7 +1307,7 @@ impl SqliteFileList {
account: &str,
file: &str,
meta: &FileMeta,
) -> Result<()> {
) -> Result<i64> {
let (stream_key, date_key, file_name) =
parse_file_key_columns(file).map_err(|e| Error::Message(e.to_string()))?;
let org_id = stream_key[..stream_key.find('/').unwrap()].to_string();
@@ -1340,25 +1336,16 @@ INSERT INTO {table} (id, account, org, stream, date, file, deleted, min_ts, max_
.execute(&*client)
.await {
Err(sqlx::Error::Database(e)) => if e.is_unique_violation() {
Ok(())
Ok(0)
} else {
Err(Error::Message(e.to_string()))
},
Err(e) => Err(e.into()),
Ok(_) => Ok(()),
Ok(v) => Ok(v.last_insert_rowid()),
}
}
async fn inner_batch_process(&self, table: &str, files: &[FileKey]) -> Result<()> {
let files: Vec<(Option<i64>, _)> = files.iter().map(|f| (None, f)).collect::<Vec<_>>();
self.inner_batch_process_with_id(table, &files).await
}
async fn inner_batch_process_with_id(
&self,
table: &str,
files: &[(Option<i64>, &FileKey)],
) -> Result<()> {
if files.is_empty() {
return Ok(());
}
@@ -1367,14 +1354,15 @@ INSERT INTO {table} (id, account, org, stream, date, file, deleted, min_ts, max_
let client = client.lock().await;
let mut tx = client.begin().await?;
let add_items = files.iter().filter(|v| !v.1.deleted).collect::<Vec<_>>();
let add_items = files.iter().filter(|f| !f.deleted).collect::<Vec<_>>();
if !add_items.is_empty() {
let chunks = add_items.chunks(100);
for files in chunks {
let mut query_builder: QueryBuilder<Sqlite> = QueryBuilder::new(
format!("INSERT INTO {table} (id, account, org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened)").as_str(),
);
query_builder.push_values(files, |mut b, (id, item)| {
query_builder.push_values(files, |mut b, item| {
let id = if item.id > 0 { Some(item.id) } else { None };
let (stream_key, date_key, file_name) =
parse_file_key_columns(&item.key).expect("parse file key failed");
let org_id = stream_key[..stream_key.find('/').unwrap()].to_string();
@@ -1405,14 +1393,18 @@ INSERT INTO {table} (id, account, org, stream, date, file, deleted, min_ts, max_
}
}
let del_items = files.iter().filter(|v| v.1.deleted).collect::<Vec<_>>();
let del_items = files.iter().filter(|f| f.deleted).collect::<Vec<_>>();
if !del_items.is_empty() {
let chunks = del_items.chunks(1000);
for files in chunks {
// get ids of the files
let mut ids = Vec::with_capacity(files.len());
for file in files {
let (stream_key, date_key, file_name) = parse_file_key_columns(&file.1.key)
if file.id > 0 {
ids.push(file.id.to_string());
continue;
}
let (stream_key, date_key, file_name) = parse_file_key_columns(&file.key)
.map_err(|e| Error::Message(e.to_string()))?;
let query_res: std::result::Result<Option<i64>, sea_orm::SqlxError> = sqlx::query_scalar(
r#"SELECT id FROM file_list WHERE stream = $1 AND date = $2 AND file = $3;"#,
src/job/file_list_dump.rs
+1
View File
@@ -436,6 +436,7 @@ async fn generate_dump(
infra::storage::put(&account, &file_key, buf.into()).await?;
let dump_file = FileKey {
id: 0,
account: account.clone(),
key: file_key.clone(),
meta: meta.clone(),
src/job/files/broadcast.rs
+1 -1
View File
@@ -30,7 +30,7 @@ pub async fn run() -> Result<(), anyhow::Error> {
}
q.drain(..).collect::<Vec<_>>()
};
if let Err(e) = broadcast::send(&files, None).await {
if let Err(e) = broadcast::send(&files).await {
log::error!("[broadcast] local queue to nodes error: {}", e);
}
}
src/job/files/parquet.rs
+4 -2
View File
@@ -328,6 +328,7 @@ async fn prepare_files(
let prefix = columns.join("/");
let partition = partition_files_with_size.entry(prefix).or_default();
partition.push(FileKey::new(
0,
"".to_string(), // here we don't need it
file_key.clone(),
parquet_meta,
@@ -536,8 +537,9 @@ async fn move_files(
}
// write file list to storage
let ret = db::file_list::set(&account, &new_file_name, Some(new_file_meta), false).await;
if let Err(e) = ret {
if let Err(e) =
db::file_list::set(&account, &new_file_name, Some(new_file_meta), false).await
{
log::error!(
"[INGESTER:JOB] Failed write parquet file meta: {}, error: {}",
new_file_name,
src/migration/file_list.rs
+2 -8
View File
@@ -15,7 +15,7 @@
use config::{
get_config,
meta::stream::{ALL_STREAM_TYPES, FileKey, PartitionTimeLevel},
meta::stream::{ALL_STREAM_TYPES, PartitionTimeLevel},
utils::time::{BASE_TIME, now_micros},
};
use infra::file_list as infra_file_list;
@@ -63,13 +63,7 @@ pub async fn run(from: &str, to: &str) -> Result<(), anyhow::Error> {
)
.await
.expect("load file_list failed");
let put_items = files
.into_iter()
.map(|(account, file_key, file_meta)| {
FileKey::new(account, file_key, file_meta, false)
})
.collect::<Vec<_>>();
dest.batch_add(&put_items)
dest.batch_add(&files)
.await
.expect("load list_list into db failed");
}
src/proto/proto/cluster/common.proto
+6 -5
View File
@@ -45,11 +45,12 @@ message FileList {
}
message FileKey {
string key = 1;
FileMeta meta = 2;
bool deleted = 3;
optional bytes segment_ids = 4;
string account = 5;
int64 id = 1;
string account = 2;
string key = 3;
FileMeta meta = 4;
bool deleted = 5;
optional bytes segment_ids = 6;
}
message IdxFileName {
src/proto/src/generated/cluster.rs
+11 -9
View File
@@ -86,16 +86,18 @@ pub struct FileList {
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct FileKey {
#[prost(string, tag = "1")]
pub key: ::prost::alloc::string::String,
#[prost(message, optional, tag = "2")]
pub meta: ::core::option::Option<FileMeta>,
#[prost(bool, tag = "3")]
pub deleted: bool,
#[prost(bytes = "vec", optional, tag = "4")]
pub segment_ids: ::core::option::Option<::prost::alloc::vec::Vec<u8>>,
#[prost(string, tag = "5")]
#[prost(int64, tag = "1")]
pub id: i64,
#[prost(string, tag = "2")]
pub account: ::prost::alloc::string::String,
#[prost(string, tag = "3")]
pub key: ::prost::alloc::string::String,
#[prost(message, optional, tag = "4")]
pub meta: ::core::option::Option<FileMeta>,
#[prost(bool, tag = "5")]
pub deleted: bool,
#[prost(bytes = "vec", optional, tag = "6")]
pub segment_ids: ::core::option::Option<::prost::alloc::vec::Vec<u8>>,
}
#[derive(Eq)]
#[allow(clippy::derive_partial_eq_without_eq)]
src/service/compact/flatten.rs
+2 -9
View File
@@ -139,17 +139,10 @@ pub async fn generate_by_stream(
start.elapsed().as_millis()
);
for (account, key, meta) in files {
if PROCESSING_FILES.read().contains(&key) {
for file in files {
if PROCESSING_FILES.read().contains(&file.key) {
continue;
}
let file = FileKey {
account,
key,
meta,
deleted: false,
segment_ids: None,
};
// add into queue
PROCESSING_FILES.write().insert(file.key.clone());
worker_tx.send(file).await?;
src/service/compact/merge.rs
+13 -6
View File
@@ -644,11 +644,9 @@ pub async fn merge_by_stream(
for file in delete_file_list {
events.push(FileKey {
account: file.account.clone(),
key: file.key.clone(),
meta: file.meta.clone(),
deleted: true,
segment_ids: None,
..file.clone()
});
}
events.sort_by(|a, b| a.key.cmp(&b.key));
@@ -999,7 +997,7 @@ pub async fn merge_files(
)
.await?;
}
new_files.push(FileKey::new(account, new_file_key, new_file_meta, false));
new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false));
}
MergeParquetResult::Multiple { bufs, file_metas } => {
for (buf, file_meta) in bufs.into_iter().zip(file_metas.into_iter()) {
@@ -1041,7 +1039,7 @@ pub async fn merge_files(
.await?;
}
new_files.push(FileKey::new(account, new_file_key, new_file_meta, false));
new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false));
}
log::info!(
"[COMPACTOR:WORKER:{thread_id}] merged {} files into a new file: {:?}, original_size: {}, compressed_size: {}, took: {} ms",
@@ -1114,6 +1112,7 @@ async fn generate_inverted_index(
if let Err(e) = write_file_list(
org_id,
&[FileKey {
id: 0,
account,
key: file_name.clone(),
meta: filemeta,
@@ -1203,7 +1202,15 @@ async fn write_file_list(org_id: &str, events: &[FileKey]) -> Result<(), anyhow:
if success {
// send broadcast to other nodes
if get_config().cache_latest_files.enabled {
if let Err(e) = db::file_list::broadcast::send(events, None).await {
// get id for all the new files
let file_ids = infra_file_list::query_ids_by_files(events).await?;
let mut events = events.to_vec();
for event in events.iter_mut() {
if let Some(id) = file_ids.get(&event.key) {
event.id = *id;
}
}
if let Err(e) = db::file_list::broadcast::send(&events).await {
log::error!("[COMPACTOR] send broadcast for file_list failed: {}", e);
}
}
src/service/db/file_list/broadcast.rs
+8 -17
View File
@@ -41,25 +41,16 @@ static EVENTS: Lazy<RwLock<HashMap<String, EventChannel>>> =
type EventChannel = Arc<mpsc::UnboundedSender<Vec<FileKey>>>;
/// send an event to broadcast, will create a new channel for each nodes
pub async fn send(items: &[FileKey], node_uuid: Option<String>) -> Result<(), anyhow::Error> {
pub async fn send(items: &[FileKey]) -> Result<(), anyhow::Error> {
let cfg = get_config();
if cfg.common.local_mode || items.is_empty() {
return Ok(());
}
let nodes = if let Some(node_uuid) = node_uuid {
cluster::get_node_by_uuid(&node_uuid)
.await
.map(|node| vec![node])
.unwrap_or_default()
} else {
cluster::get_cached_nodes(|node| {
node.scheduled
&& (node.is_querier())
&& (node.status == NodeStatus::Prepare || node.status == NodeStatus::Online)
})
.await
.unwrap_or_default()
};
let nodes = cluster::get_cached_nodes(|node| {
node.scheduled && node.is_querier() && node.status == NodeStatus::Online
})
.await
.unwrap_or_default();
let mut events = EVENTS.write().await;
for node in nodes {
if node.uuid.eq(&LOCAL_NODE.uuid) {
@@ -74,7 +65,7 @@ pub async fn send(items: &[FileKey], node_uuid: Option<String>) -> Result<(), an
}
// check if the item is for interactive node
if let Some(node_name) = cluster::get_node_from_consistent_hash(
&item.key,
&item.id.to_string(),
&Role::Querier,
Some(RoleGroup::Interactive),
)
@@ -87,7 +78,7 @@ pub async fn send(items: &[FileKey], node_uuid: Option<String>) -> Result<(), an
}
// check if the item is for background node
if let Some(node_name) = cluster::get_node_from_consistent_hash(
&item.key,
&item.id.to_string(),
&Role::Querier,
Some(RoleGroup::Background),
)
src/service/db/file_list/mod.rs
+24 -21
View File
@@ -46,7 +46,8 @@ pub static BLOCKED_ORGS: Lazy<HashSet<String>> = Lazy::new(|| {
});
pub async fn set(account: &str, key: &str, meta: Option<FileMeta>, deleted: bool) -> Result<()> {
let file_data = FileKey::new(
let mut file_data = FileKey::new(
0,
account.to_string(),
key.to_string(),
meta.clone().unwrap_or_default(),
@@ -56,11 +57,15 @@ pub async fn set(account: &str, key: &str, meta: Option<FileMeta>, deleted: bool
// write into file_list storage
// retry 5 times
for _ in 0..5 {
if let Err(e) = progress(account, key, meta.as_ref(), deleted).await {
log::error!("[FILE_LIST] Error saving file to storage, retrying: {}", e);
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
} else {
break;
match progress(account, key, meta.as_ref(), deleted).await {
Ok(id) => {
file_data.id = id;
break;
}
Err(e) => {
log::error!("[FILE_LIST] Error saving file to storage, retrying: {}", e);
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
}
}
@@ -75,26 +80,24 @@ pub async fn set(account: &str, key: &str, meta: Option<FileMeta>, deleted: bool
Ok(())
}
async fn progress(account: &str, key: &str, data: Option<&FileMeta>, delete: bool) -> Result<()> {
async fn progress(account: &str, key: &str, data: Option<&FileMeta>, delete: bool) -> Result<i64> {
let mut id = 0;
if delete {
if let Err(e) = infra::file_list::remove(key).await {
log::error!(
"service:db:file_list: delete {}, set_file_to_cache error: {}",
key,
e
);
log::error!("service:db:file_list: delete {}, remove error: {}", key, e);
}
} else {
if let Err(e) = infra::file_list::add(account, key, data.unwrap()).await {
log::error!(
"service:db:file_list: add {}, set_file_to_cache error: {}",
key,
e
);
} else if let Some(data) = data {
match infra::file_list::add(account, key, data).await {
Ok(v) => {
id = v;
}
Err(e) => {
log::error!("service:db:file_list: add {}, add error: {}", key, e);
}
}
// update stream stats realtime
if config::get_config().common.local_mode {
if let Err(e) = infra::cache::stats::incr_stream_stats(key, data.unwrap()) {
if let Err(e) = infra::cache::stats::incr_stream_stats(key, data) {
log::error!(
"service:db:file_list: add {}, incr_stream_stats error: {}",
key,
@@ -104,7 +107,7 @@ async fn progress(account: &str, key: &str, data: Option<&FileMeta>, delete: boo
}
}
Ok(())
Ok(id)
}
pub async fn cache_stats() -> Result<()> {
src/service/file_list.rs
+22 -106
View File
@@ -19,7 +19,7 @@ use config::{
get_config,
meta::{
search::ScanStats,
stream::{FileKey, FileMeta, PartitionTimeLevel, StreamType},
stream::{FileKey, PartitionTimeLevel, StreamType},
},
metrics::{FILE_LIST_CACHE_HIT_COUNT, FILE_LIST_ID_SELECT_COUNT},
utils::file::get_file_meta as util_get_file_meta,
@@ -48,7 +48,7 @@ pub async fn query(
time_max: i64,
) -> Result<Vec<FileKey>> {
let cfg = get_config();
let files = file_list::query(
let mut files = file_list::query(
org_id,
stream_type,
stream_name,
@@ -73,8 +73,8 @@ pub async fn query(
.collect::<HashSet<_>>();
let missing_files: usize = files
.iter()
.map(|(_account, name, _)| {
if dumped_file_names.contains(name) {
.map(|f| {
if dumped_file_names.contains(&f.key) {
0
} else {
1
@@ -87,42 +87,19 @@ pub async fn query(
);
}
}
let mut file_keys = Vec::with_capacity(files.len());
for (account, key, meta) in files {
file_keys.push(FileKey {
account,
key,
meta,
deleted: false,
segment_ids: None,
});
}
if !cfg.common.file_list_dump_dual_write {
// we only consider these files in case of dual write disabled,
// because with dual write there are some edge cases which cannot be sovled even with id
// de-dup so the data gets counted twice.
for file in dumped_files {
file_keys.push(FileKey {
account: file.account,
key: "files/".to_string() + &file.stream + "/" + &file.date + "/" + &file.file,
meta: FileMeta {
min_ts: file.min_ts,
max_ts: file.max_ts,
records: file.records,
original_size: file.original_size,
compressed_size: file.compressed_size,
index_size: file.index_size,
flattened: file.flattened,
},
deleted: file.deleted,
segment_ids: None,
})
for file in dumped_files.iter() {
files.push(file.into())
}
}
file_keys.par_sort_unstable_by(|a, b| a.key.cmp(&b.key));
file_keys.dedup_by(|a, b| a.key == b.key);
Ok(file_keys)
files.par_sort_unstable_by(|a, b| a.key.cmp(&b.key));
files.dedup_by(|a, b| a.key == b.key);
Ok(files)
}
/// NOTE: This will not query the files from file_dump. If you also want files from the dump, use
@@ -148,20 +125,10 @@ pub async fn query_for_merge(
Some((date_start.to_string(), date_end.to_string())),
)
.await?;
let mut file_keys = Vec::with_capacity(files.len());
for (account, key, meta) in files {
file_keys.push(FileKey {
account,
key,
meta,
deleted: false,
segment_ids: None,
});
}
// we don't need to query from dump here, because
// we expected all dumped files to be already compacted,
// and the compaction marks old files as deleted, which is not possible with dump
Ok(file_keys)
Ok(files)
}
#[tracing::instrument(name = "service::file_list::query_by_ids", skip_all)]
@@ -191,10 +158,7 @@ pub async fn query_by_ids(
Vec::new()
}
};
let cached_ids = cached_files
.iter()
.map(|(id, ..)| *id)
.collect::<HashSet<_>>();
let cached_ids = cached_files.iter().map(|f| f.id).collect::<HashSet<_>>();
log::info!(
"{}",
search_inspector_fields(
@@ -221,18 +185,8 @@ pub async fn query_by_ids(
.with_label_values(&[])
.set(cached_ids.len() as i64);
let mut file_keys = Vec::with_capacity(ids.len());
for (_, account, key, meta) in cached_files {
file_keys.push(FileKey {
account,
key,
meta,
deleted: false,
segment_ids: None,
});
}
(
file_keys,
cached_files,
ids_set.difference(&cached_ids).cloned().collect::<Vec<_>>(),
)
} else {
@@ -242,21 +196,6 @@ pub async fn query_by_ids(
// 2. query from remote db
let start = std::time::Instant::now();
let db_files = file_list::query_by_ids(&ids).await?;
let db_files = db_files
.into_iter()
.map(|(id, account, key, meta)| {
(
id,
FileKey {
account,
key,
meta,
deleted: false,
segment_ids: None,
},
)
})
.collect::<Vec<_>>();
log::info!(
"{}",
search_inspector_fields(
@@ -288,37 +227,15 @@ pub async fn query_by_ids(
)
.await?;
let dumped_files: Vec<_> = dumped_files
.into_iter()
let dumped_files: Vec<FileKey> = dumped_files
.iter()
.filter(|r| ids.contains(&r.id))
.map(|r| {
(
r.id,
FileKey {
account: r.account,
key: "files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file,
meta: FileMeta {
min_ts: r.min_ts,
max_ts: r.max_ts,
records: r.records,
original_size: r.original_size,
compressed_size: r.compressed_size,
index_size: r.index_size,
flattened: r.flattened,
},
deleted: r.deleted,
segment_ids: None,
},
)
})
.map(|r| r.into())
.collect();
if cfg.common.file_list_dump_enabled && cfg.common.file_list_dump_debug_check {
let dump_ids = dumped_files
.iter()
.map(|entry| entry.0)
.collect::<HashSet<_>>();
let db_ids = db_files.iter().map(|entry| entry.0).collect::<Vec<_>>();
let dump_ids = dumped_files.iter().map(|f| f.id).collect::<HashSet<_>>();
let db_ids = db_files.iter().map(|f| f.id).collect::<Vec<_>>();
let missing_files: usize = db_ids
.iter()
.map(|id| if dump_ids.contains(id) { 0 } else { 1 })
@@ -333,7 +250,6 @@ pub async fn query_by_ids(
// 3. set the local cache
if !cfg.common.local_mode {
let start = std::time::Instant::now();
let db_files: Vec<_> = db_files.iter().map(|(id, f)| (*id, f)).collect();
if let Err(e) = file_list::LOCAL_CACHE.batch_add_with_id(&db_files).await {
log::error!(
"[trace_id {trace_id}] file_list set cache failed for db files: {:?}",
@@ -342,7 +258,6 @@ pub async fn query_by_ids(
}
if !cfg.common.file_list_dump_dual_write {
let dumped_files: Vec<_> = dumped_files.iter().map(|(id, f)| (*id, f)).collect();
if let Err(e) = file_list::LOCAL_CACHE
.batch_add_with_id(&dumped_files)
.await
@@ -374,9 +289,9 @@ pub async fn query_by_ids(
}
// 4. merge the results
files.extend(db_files.into_iter().map(|(_, f)| f));
files.extend(db_files);
if !cfg.common.file_list_dump_dual_write {
files.extend(dumped_files.into_iter().map(|(_, f)| f));
files.extend(dumped_files);
}
files.par_sort_unstable_by(|a, b| a.key.cmp(&b.key));
files.dedup_by(|a, b| a.key == b.key);
@@ -421,7 +336,7 @@ pub async fn query_ids(
}
}
if !cfg.common.file_list_dump_dual_write {
files.extend(dumped_files.into_iter());
files.extend(dumped_files);
}
files.par_sort_unstable_by(|a, b| a.id.cmp(&b.id));
files.dedup_by(|a, b| a.id == b.id);
@@ -458,6 +373,7 @@ pub fn calculate_local_files_size(files: &[String]) -> Result<u64> {
pub async fn delete_parquet_file(account: &str, key: &str, file_list_only: bool) -> Result<()> {
// delete from file list in metastore
file_list::batch_process(&[FileKey::new(
0,
account.to_string(),
key.to_string(),
Default::default(),
src/service/file_list_dump.rs
+8 -36
View File
@@ -230,16 +230,7 @@ pub async fn query(
let db_time = db_start.elapsed().as_millis();
let process_start = std::time::Instant::now();
let dump_files: Vec<_> = dump_files
.into_iter()
.map(|f| FileKey {
account: f.account.to_string(),
key: format!("files/{}/{}/{}", stream_key, f.date, f.file),
meta: (&f).into(),
deleted: false,
segment_ids: None,
})
.collect();
let dump_files: Vec<_> = dump_files.iter().map(|f| f.into()).collect();
let max_ts_upper_bound = calculate_max_ts_upper_bound(range.1, stream_type);
let stream_key = format!("{org}/{stream_type}/{stream}");
@@ -283,16 +274,7 @@ pub async fn get_ids_in_range(
let db_time = db_start.elapsed().as_millis();
let process_start = std::time::Instant::now();
let dump_files: Vec<_> = dump_files
.into_iter()
.map(|f| FileKey {
account: f.account.to_string(),
key: format!("files/{}/{}/{}", stream_key, f.date, f.file),
meta: (&f).into(),
deleted: false,
segment_ids: None,
})
.collect();
let dump_files: Vec<_> = dump_files.iter().map(|f| f.into()).collect();
let max_ts_upper_bound = calculate_max_ts_upper_bound(range.1, stream_type);
let stream_key = format!("{org}/{stream_type}/{stream}");
@@ -367,12 +349,11 @@ async fn move_and_delete(
let items: Vec<_> = list
.iter()
.chain(dump_files.iter())
.map(|f| FileKey {
account: f.account.to_string(),
key: format!("files/{}/{}/{}", f.stream, f.date, f.file),
meta: f.into(),
deleted: true,
segment_ids: None,
.map(|f| {
let mut f = FileKey::from(f);
f.deleted = true;
f.segment_ids = None;
f
})
.collect();
if let Err(e) = infra::file_list::batch_process(&items).await {
@@ -461,16 +442,7 @@ pub async fn stats(
return Ok(vec![]);
}
let dump_files: Vec<_> = dump_files
.into_iter()
.map(|f| FileKey {
account: f.account.to_string(),
key: format!("files/{}/{}/{}", f.stream, f.date, f.file),
meta: (&f).into(),
deleted: false,
segment_ids: None,
})
.collect();
let dump_files: Vec<_> = dump_files.iter().map(|f| f.into()).collect();
let sql = format!(
r#"