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

chore(deps): update to datafusion 39 (#3734)

Browse Source
This commit is contained in:
Hengfei Yang
2024-06-14 11:43:43 +08:00
committed by GitHub
parent 8303edcfd4
commit 53e45b6b90
10 changed files with 562 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+423 -116
View File
File diff suppressed because it is too large Load Diff
Cargo.toml
+8 -8
View File
@@ -215,13 +215,13 @@ chromiumoxide = { git = "https://github.com/mattsse/chromiumoxide", features = [
chrono = { version = "0.4", default-features = false, features = ["clock"] }
cityhasher = { version = "0.1", default-features = false }
dashmap = { version = "5.5", features = ["serde"] }
datafusion = "38"
datafusion-expr = "38"
arrow = { version = "51.0.0", features = ["ipc_compression", "prettyprint"] }
arrow-json = "51.0.0"
arrow-schema = { version = "51.0.0", features = ["serde"] }
parquet = { version = "51.0.0", features = ["arrow", "async", "object_store"] }
object_store = { version = "0.9.1", features = ["aws", "azure", "gcp"] }
datafusion = "39"
datafusion-expr = "39"
arrow = { version = "52.0.0", features = ["ipc_compression", "prettyprint"] }
arrow-json = "52.0.0"
arrow-schema = { version = "52.0.0", features = ["serde"] }
parquet = { version = "52.0.0", features = ["arrow", "async", "object_store"] }
object_store = { version = "0.10.1", features = ["aws", "azure", "gcp"] }
dotenv_config = "0.1"
dotenvy = "0.15"
env_logger = "0.10"
@@ -274,7 +274,7 @@ sqlx = { version = "0.7", features = [
"sqlite",
"chrono",
] }
sqlparser = { version = "0.45", features = ["serde", "visitor"] }
sqlparser = { version = "0.47", features = ["serde", "visitor"] }
svix-ksuid = { version = "0.8", features = ["serde"] }
sysinfo = "0.29"
tempfile = "3"
src/config/src/meta/sql.rs
+30 -16
View File
@@ -18,9 +18,9 @@ use regex::Regex;
use serde::Serialize;
use sqlparser::{
ast::{
BinaryOperator, Expr as SqlExpr, Function, FunctionArg, FunctionArgExpr, GroupByExpr,
Offset as SqlOffset, OrderByExpr, Select, SelectItem, SetExpr, Statement, TableFactor,
TableWithJoins, Value,
BinaryOperator, Expr as SqlExpr, Function, FunctionArg, FunctionArgExpr, FunctionArguments,
GroupByExpr, Offset as SqlOffset, OrderByExpr, Select, SelectItem, SetExpr, Statement,
TableFactor, TableWithJoins, Value,
},
parser::Parser,
};
@@ -557,7 +557,7 @@ fn parse_expr_like(
_negated: &bool,
expr: &SqlExpr,
pattern: &SqlExpr,
_escape_char: &Option<char>,
_escape_char: &Option<String>,
next_op: &SqlOperator,
field: &str,
fields: &mut Vec<(String, SqlValue, SqlOperator, SqlOperator)>,
@@ -660,16 +660,21 @@ fn parse_expr_function(
return parse_expr_fun_time_range(f, field, fields);
}
if f.args.len() < 2 {
let args = match &f.args {
FunctionArguments::None => return Ok(()),
FunctionArguments::Subquery(_) => return Ok(()),
FunctionArguments::List(args) => &args.args,
};
if args.len() < 2 {
return Ok(());
}
let nop = SqlOperator::And;
let next_op = SqlOperator::And;
let field_name = f.args.first().unwrap().to_string();
let field_name = args.first().unwrap().to_string();
let field_name = field_name.trim_matches(|c: char| c == '\'' || c == '"');
if parse_expr_check_field_name(field_name, field) {
match f.args.get(1).unwrap() {
match args.get(1).unwrap() {
FunctionArg::Named {
name: _name,
arg,
@@ -709,18 +714,23 @@ fn parse_expr_fun_time_range(
field: &str,
fields: &mut Vec<(String, SqlValue, SqlOperator, SqlOperator)>,
) -> Result<(), anyhow::Error> {
if f.args.len() != 3 {
let args = match &f.args {
FunctionArguments::None => return Ok(()),
FunctionArguments::Subquery(_) => return Ok(()),
FunctionArguments::List(args) => &args.args,
};
if args.len() != 3 {
return Err(anyhow::anyhow!(
"SqlExpr::Function: time_range function must have 3 arguments"
));
}
let next_op = SqlOperator::And;
let field_name = f.args.first().unwrap().to_string();
let field_name = args.first().unwrap().to_string();
let field_name = field_name.trim_matches(|c: char| c == '\'' || c == '"');
if parse_expr_check_field_name(field_name, field) {
let mut vals = Vec::new();
for arg in f.args.iter() {
for arg in args.iter() {
let val = match arg {
FunctionArg::Named {
name: _name,
@@ -799,8 +809,13 @@ fn get_field_name_from_expr(expr: &SqlExpr) -> Option<Vec<String>> {
(!fields.is_empty()).then_some(fields)
}
SqlExpr::Function(f) => {
let mut fields = Vec::with_capacity(f.args.len());
for arg in f.args.iter() {
let args = match &f.args {
FunctionArguments::None => return None,
FunctionArguments::Subquery(_) => return None,
FunctionArguments::List(args) => &args.args,
};
let mut fields = Vec::with_capacity(args.len());
for arg in args.iter() {
match arg {
FunctionArg::Named {
name: _name,
@@ -842,9 +857,7 @@ fn get_field_name_from_expr(expr: &SqlExpr) -> Option<Vec<String>> {
}
(!fields.is_empty()).then_some(fields)
}
SqlExpr::Cast { expr, .. } | SqlExpr::TryCast { expr, .. } => {
get_field_name_from_expr(expr)
}
SqlExpr::Cast { expr, .. } => get_field_name_from_expr(expr),
SqlExpr::Case {
operand: _,
conditions,
@@ -862,7 +875,8 @@ fn get_field_name_from_expr(expr: &SqlExpr) -> Option<Vec<String>> {
SqlExpr::AtTimeZone { timestamp, .. } => get_field_name_from_expr(timestamp),
SqlExpr::Extract { expr, .. } => get_field_name_from_expr(expr),
SqlExpr::MapAccess { column, .. } => get_field_name_from_expr(column),
SqlExpr::ArrayIndex { obj, .. } => get_field_name_from_expr(obj),
SqlExpr::CompositeAccess { expr, .. } => get_field_name_from_expr(expr),
SqlExpr::Subscript { expr, .. } => get_field_name_from_expr(expr),
_ => None,
}
}
src/infra/src/storage/local.rs
+16 -10
View File
@@ -21,9 +21,9 @@ use config::metrics;
use futures::stream::BoxStream;
use object_store::{
limit::LimitStore, local::LocalFileSystem, path::Path, Error, GetOptions, GetResult,
ListResult, MultipartId, ObjectMeta, ObjectStore, PutOptions, PutResult, Result,
ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload,
PutResult, Result,
};
use tokio::io::AsyncWrite;
use crate::storage::{format_key, CONCURRENT_REQUESTS};
@@ -61,13 +61,18 @@ impl std::fmt::Display for Local {
#[async_trait]
impl ObjectStore for Local {
async fn put_opts(&self, location: &Path, bytes: Bytes, opts: PutOptions) -> Result<PutResult> {
async fn put_opts(
&self,
location: &Path,
payload: PutPayload,
opts: PutOptions,
) -> Result<PutResult> {
let start = std::time::Instant::now();
let file = location.to_string();
let data_size = bytes.len();
let data_size = payload.content_length();
match self
.client
.put_opts(&(format_key(&file, self.with_prefix).into()), bytes, opts)
.put_opts(&(format_key(&file, self.with_prefix).into()), payload, opts)
.await
{
Ok(_output) => {
@@ -97,14 +102,15 @@ impl ObjectStore for Local {
}
}
async fn put_multipart(
&self,
_location: &Path,
) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
Err(Error::NotImplemented)
}
async fn abort_multipart(&self, _location: &Path, _multipart_id: &MultipartId) -> Result<()> {
async fn put_multipart_opts(
&self,
_location: &Path,
_opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
Err(Error::NotImplemented)
}
src/infra/src/storage/mod.rs
+1 -1
View File
@@ -77,7 +77,7 @@ pub async fn get(file: &str) -> Result<bytes::Bytes, anyhow::Error> {
}
pub async fn put(file: &str, data: bytes::Bytes) -> Result<(), anyhow::Error> {
DEFAULT.put(&file.into(), data).await?;
DEFAULT.put(&file.into(), data.into()).await?;
Ok(())
}
src/infra/src/storage/remote.rs
+16 -11
View File
@@ -20,10 +20,9 @@ use bytes::Bytes;
use config::{get_config, metrics};
use futures::stream::BoxStream;
use object_store::{
limit::LimitStore, path::Path, Error, GetOptions, GetResult, ListResult, MultipartId,
ObjectMeta, ObjectStore, PutOptions, PutResult, Result,
limit::LimitStore, path::Path, Error, GetOptions, GetResult, ListResult, MultipartUpload,
ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
};
use tokio::io::AsyncWrite;
use crate::storage::{format_key, CONCURRENT_REQUESTS};
@@ -53,13 +52,18 @@ impl std::fmt::Display for Remote {
#[async_trait]
impl ObjectStore for Remote {
async fn put_opts(&self, location: &Path, bytes: Bytes, opts: PutOptions) -> Result<PutResult> {
async fn put_opts(
&self,
location: &Path,
payload: PutPayload,
opts: PutOptions,
) -> Result<PutResult> {
let start = std::time::Instant::now();
let file = location.to_string();
let data_size = bytes.len();
let data_size = payload.content_length();
match self
.client
.put_opts(&(format_key(&file, true).into()), bytes, opts)
.put_opts(&(format_key(&file, true).into()), payload, opts)
.await
{
Ok(_) => {
@@ -89,14 +93,15 @@ impl ObjectStore for Remote {
}
}
async fn put_multipart(
&self,
_location: &Path,
) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
Err(Error::NotImplemented)
}
async fn abort_multipart(&self, _location: &Path, _multipart_id: &MultipartId) -> Result<()> {
async fn put_multipart_opts(
&self,
_location: &Path,
_opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
Err(Error::NotImplemented)
}
src/service/search/datafusion/rewrite.rs
+29 -18
View File
@@ -21,7 +21,7 @@ use config::FxIndexSet;
use datafusion::error::Result;
use itertools::Itertools;
use sqlparser::{
ast::{Expr, Function, GroupByExpr, Ident, Query, VisitMut, VisitorMut},
ast::{Expr, Function, FunctionArguments, GroupByExpr, Ident, Query, VisitMut, VisitorMut},
dialect::GenericDialect,
parser::Parser,
};
@@ -86,23 +86,29 @@ impl VisitorMut for Rewrite {
}
let mut field_names = FxIndexSet::default();
let mut delete_idx = Vec::new();
let empty_vec = vec![];
for (idx, select_item) in &mut select.projection.iter_mut().enumerate() {
match select_item {
sqlparser::ast::SelectItem::ExprWithAlias { expr, alias } => {
match expr {
Expr::Function(Function {
name,
args,
distinct,
..
}) => {
Expr::Function(Function { name, args, .. }) => {
// extract field name from the aggregate function
if AGGREGATE_UDF_LIST
.contains(&name.to_string().to_lowercase().as_str())
{
// check if aggregate function is count and distinct is
// false
if name.to_string().to_lowercase() == "count" && !(*distinct) {
let distinct = match &args {
FunctionArguments::List(args) => matches!(
args.duplicate_treatment,
Some(sqlparser::ast::DuplicateTreatment::Distinct,)
),
_ => false,
};
let args = match &args {
FunctionArguments::List(args) => &args.args,
_ => &empty_vec,
};
// check if aggregate function is count and distinct is false
if name.to_string().to_lowercase() == "count" && !(distinct) {
select.distinct = None;
}
// check if the field name is already in the projection list
@@ -133,13 +139,7 @@ impl VisitorMut for Rewrite {
}
// extrace function paraments from the projection list
sqlparser::ast::SelectItem::UnnamedExpr(expr) => {
if let Expr::Function(Function {
name,
args,
distinct,
..
}) = expr
{
if let Expr::Function(Function { name, args, .. }) = expr {
if !AGGREGATE_UDF_LIST
.contains(&name.to_string().to_lowercase().as_str())
{
@@ -154,7 +154,18 @@ impl VisitorMut for Rewrite {
}
}
} else {
if name.to_string().to_lowercase() == "count" && !(*distinct) {
let distinct = match &args {
FunctionArguments::List(args) => matches!(
args.duplicate_treatment,
Some(sqlparser::ast::DuplicateTreatment::Distinct)
),
_ => false,
};
let args = match &args {
FunctionArguments::List(args) => &args.args,
_ => &empty_vec,
};
if name.to_string().to_lowercase() == "count" && !(distinct) {
select.distinct = None;
}
let field_name = remove_brackets(format!("{}", args[0]).as_str());
src/service/search/datafusion/storage/memory.rs
+13 -13
View File
@@ -21,10 +21,9 @@ use config::utils::time::BASE_TIME;
use futures::{stream::BoxStream, StreamExt};
use infra::{cache::file_data, storage};
use object_store::{
path::Path, GetOptions, GetResult, GetResultPayload, ListResult, MultipartId, ObjectMeta,
ObjectStore, PutOptions, PutResult, Result,
path::Path, Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
};
use tokio::io::AsyncWrite;
use super::GetRangeExt;
@@ -81,6 +80,7 @@ impl ObjectStore for FS {
payload: GetResultPayload::Stream(
futures::stream::once(async move { Ok(data) }).boxed(),
),
attributes: Attributes::default(),
meta,
range,
})
@@ -116,6 +116,7 @@ impl ObjectStore for FS {
payload: GetResultPayload::Stream(
futures::stream::once(async move { Ok(data) }).boxed(),
),
attributes: Attributes::default(),
meta,
range,
})
@@ -224,24 +225,23 @@ impl ObjectStore for FS {
async fn put_opts(
&self,
location: &Path,
_bytes: Bytes,
_payload: PutPayload,
_opts: PutOptions,
) -> Result<PutResult> {
log::error!("NotImplemented put_opts: {}", location);
Err(object_store::Error::NotImplemented {})
}
async fn put_multipart(
&self,
location: &Path,
) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
log::error!("NotImplemented put_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
Err(object_store::Error::NotImplemented)
}
async fn abort_multipart(&self, location: &Path, _multipart_id: &MultipartId) -> Result<()> {
log::error!("NotImplemented abort_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
async fn put_multipart_opts(
&self,
_location: &Path,
_opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
Err(object_store::Error::NotImplemented)
}
async fn delete(&self, location: &Path) -> Result<()> {
src/service/search/datafusion/storage/tmpfs.rs
+14 -12
View File
@@ -22,11 +22,10 @@ use config::utils::time::BASE_TIME;
use futures::{stream::BoxStream, StreamExt};
use infra::cache::tmpfs;
use object_store::{
path::Path, GetOptions, GetResult, GetResultPayload, ListResult, MultipartId, ObjectMeta,
ObjectStore, PutOptions, PutResult, Result,
path::Path, Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
};
use thiserror::Error as ThisError;
use tokio::io::AsyncWrite;
use super::GetRangeExt;
@@ -80,6 +79,7 @@ impl ObjectStore for Tmpfs {
payload: GetResultPayload::Stream(
futures::stream::once(async move { Ok(data) }).boxed(),
),
attributes: Attributes::default(),
meta,
range,
})
@@ -108,6 +108,7 @@ impl ObjectStore for Tmpfs {
payload: GetResultPayload::Stream(
futures::stream::once(async move { Ok(data) }).boxed(),
),
attributes: Attributes::default(),
meta,
range,
})
@@ -198,24 +199,25 @@ impl ObjectStore for Tmpfs {
async fn put_opts(
&self,
location: &Path,
_bytes: Bytes,
_payload: PutPayload,
_opts: PutOptions,
) -> Result<PutResult> {
log::error!("NotImplemented put_opts: {}", location);
Err(object_store::Error::NotImplemented {})
}
async fn put_multipart(
&self,
location: &Path,
) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
log::error!("NotImplemented put_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
Err(object_store::Error::NotImplemented)
}
async fn abort_multipart(&self, location: &Path, _multipart_id: &MultipartId) -> Result<()> {
log::error!("NotImplemented abort_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
async fn put_multipart_opts(
&self,
location: &Path,
_opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
log::error!("NotImplemented put_multipart_opts: {}", location);
Err(object_store::Error::NotImplemented)
}
async fn delete(&self, location: &Path) -> Result<()> {
src/service/search/datafusion/storage/wal.rs
+12 -12
View File
@@ -20,10 +20,9 @@ use bytes::Bytes;
use futures::{stream::BoxStream, StreamExt};
use infra::storage;
use object_store::{
path::Path, GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore,
PutOptions, PutResult, Result,
path::Path, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
};
use tokio::io::AsyncWrite;
/// File system for local wal
#[derive(Debug, Default)]
@@ -99,24 +98,25 @@ impl ObjectStore for FS {
async fn put_opts(
&self,
location: &Path,
_bytes: Bytes,
_payload: PutPayload,
_opts: PutOptions,
) -> Result<PutResult> {
log::error!("NotImplemented put_opts: {}", location);
Err(object_store::Error::NotImplemented {})
}
async fn put_multipart(
&self,
location: &Path,
) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
log::error!("NotImplemented put_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
Err(object_store::Error::NotImplemented)
}
async fn abort_multipart(&self, location: &Path, _multipart_id: &MultipartId) -> Result<()> {
log::error!("NotImplemented abort_multipart: {}", location);
Err(object_store::Error::NotImplemented {})
async fn put_multipart_opts(
&self,
location: &Path,
_opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
log::error!("NotImplemented put_multipart_opts: {}", location);
Err(object_store::Error::NotImplemented)
}
async fn delete(&self, location: &Path) -> Result<()> {