feat: update to datafusion 35 (#2586)

- [x] upgrade to datafusion 35
- [x] fixed #2582

src/service/metrics/otlp_grpc.rs

@@ -302,7 +302,11 @@ pub async fn handle_grpc_request(
                     let buf = metric_data_map
                         .entry(local_metric_name.to_owned())
                         .or_default();
-                    let schema = metric_schema_map.get(local_metric_name).unwrap().clone();
+                    let schema = metric_schema_map
+                        .get(local_metric_name)
+                        .unwrap()
+                        .clone()
+                        .with_metadata(HashMap::new());
                     let schema_key = schema.hash_key();
                     // get hour key
                     let hour_key = crate::service::ingestion::get_wal_time_key(

src/service/search/datafusion/date_format_udf.rs

@@ -51,9 +51,12 @@ pub(crate) static DATE_FORMAT_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
 pub fn date_format_expr_impl() -> ScalarFunctionImplementation {
     let func = move |args: &[ArrayRef]| -> datafusion::error::Result<ArrayRef> {
         if args.len() != 3 {
-            return Err(DataFusionError::SQL(ParserError::ParserError(
-                "UDF params should be: date_format(field, format, zone)".to_string(),
-            )));
+            return Err(DataFusionError::SQL(
+                ParserError::ParserError(
+                    "UDF params should be: date_format(field, format, zone)".to_string(),
+                ),
+                None,
+            ));
         }
 
         // 1. cast both arguments to Union. These casts MUST be aligned with the signature or this

src/service/search/datafusion/exec.rs

@@ -1002,6 +1002,10 @@ pub fn create_session_config(search_type: &SearchType) -> Result<SessionConfig>
     let mut config = SessionConfig::from_env()?
         .with_batch_size(PARQUET_BATCH_SIZE)
         .with_information_schema(true);
+    config = config.set_bool(
+        "datafusion.execution.listing_table_ignore_subdirectory",
+        false,
+    );
     if search_type == &SearchType::Normal {
         config = config.set_bool("datafusion.execution.parquet.pushdown_filters", true);
         config = config.set_bool("datafusion.execution.parquet.reorder_filters", true);

src/service/search/datafusion/match_udf.rs

@@ -60,9 +60,10 @@ pub(crate) static MATCH_IGNORE_CASE_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
 pub fn match_expr_impl(case_insensitive: bool) -> ScalarFunctionImplementation {
     let func = move |args: &[ArrayRef]| -> datafusion::error::Result<ArrayRef> {
         if args.len() != 2 {
-            return Err(DataFusionError::SQL(ParserError::ParserError(
-                "match UDF expects two string".to_string(),
-            )));
+            return Err(DataFusionError::SQL(
+                ParserError::ParserError("match UDF expects two string".to_string()),
+                None,
+            ));
         }
 
         // 1. cast both arguments to string. These casts MUST be aligned with the signature or this

src/service/search/datafusion/storage/memory.rs

@@ -24,6 +24,7 @@ use object_store::{
 };
 use tokio::io::AsyncWrite;
 
+use super::GetRangeExt;
 use crate::common::{
     infra::{cache::file_data, storage},
     utils::time::BASE_TIME,
@@ -105,7 +106,12 @@ impl ObjectStore for FS {
                     version: None,
                 };
                 let (range, data) = match options.range {
-                    Some(range) => (range.clone(), data.slice(range)),
+                    Some(range) => {
+                        let r = range
+                            .as_range(data.len())
+                            .map_err(|e| super::Error::BadRange(e.to_string()))?;
+                        (r.clone(), data.slice(r))
+                    }
                     None => (0..data.len(), data),
                 };
                 Ok(GetResult {

src/service/search/datafusion/storage/mod.rs

@@ -13,6 +13,10 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::ops::Range;
+
+use object_store::GetRange;
+use snafu::Snafu;
 use thiserror::Error as ThisError;
 
 pub mod file_list;
@@ -28,7 +32,7 @@ pub enum StorageType {
 /// A specialized `Error` for in-memory object store-related errors
 #[derive(ThisError, Debug)]
 #[allow(missing_docs)]
-enum Error {
+pub(crate) enum Error {
     #[error("Out of range")]
     OutOfRange(String),
     #[error("Bad range")]
@@ -43,3 +47,65 @@ impl From<Error> for object_store::Error {
         }
     }
 }
+
+#[derive(Debug, Snafu)]
+pub(crate) enum InvalidGetRange {
+    #[snafu(display(
+        "Wanted range starting at {requested}, but object was only {length} bytes long"
+    ))]
+    StartTooLarge { requested: usize, length: usize },
+
+    #[snafu(display("Range started at {start} and ended at {end}"))]
+    Inconsistent { start: usize, end: usize },
+}
+
+pub(crate) trait GetRangeExt {
+    fn is_valid(&self) -> Result<(), InvalidGetRange>;
+    /// Convert to a [`Range`] if valid.
+    fn as_range(&self, len: usize) -> Result<Range<usize>, InvalidGetRange>;
+}
+
+impl GetRangeExt for GetRange {
+    fn is_valid(&self) -> Result<(), InvalidGetRange> {
+        match self {
+            Self::Bounded(r) if r.end <= r.start => {
+                return Err(InvalidGetRange::Inconsistent {
+                    start: r.start,
+                    end: r.end,
+                });
+            }
+            _ => (),
+        };
+        Ok(())
+    }
+
+    /// Convert to a [`Range`] if valid.
+    fn as_range(&self, len: usize) -> Result<Range<usize>, InvalidGetRange> {
+        self.is_valid()?;
+        match self {
+            Self::Bounded(r) => {
+                if r.start >= len {
+                    Err(InvalidGetRange::StartTooLarge {
+                        requested: r.start,
+                        length: len,
+                    })
+                } else if r.end > len {
+                    Ok(r.start..len)
+                } else {
+                    Ok(r.clone())
+                }
+            }
+            Self::Offset(o) => {
+                if *o >= len {
+                    Err(InvalidGetRange::StartTooLarge {
+                        requested: *o,
+                        length: len,
+                    })
+                } else {
+                    Ok(*o..len)
+                }
+            }
+            Self::Suffix(n) => Ok(len.saturating_sub(*n)..len),
+        }
+    }
+}

src/service/search/datafusion/storage/tmpfs.rs

@@ -26,6 +26,7 @@ use object_store::{
 use thiserror::Error as ThisError;
 use tokio::io::AsyncWrite;
 
+use super::GetRangeExt;
 use crate::common::{infra::cache::tmpfs, utils::time::BASE_TIME};
 
 /// A specialized `Error` for in-memory object store-related errors
@@ -94,7 +95,12 @@ impl ObjectStore for Tmpfs {
             version: None,
         };
         let (range, data) = match options.range {
-            Some(range) => (range.clone(), data.slice(range)),
+            Some(range) => {
+                let r = range
+                    .as_range(data.len())
+                    .map_err(|e| super::Error::BadRange(e.to_string()))?;
+                (r.clone(), data.slice(r))
+            }
             None => (0..data.len(), data),
         };
         Ok(GetResult {

src/service/search/datafusion/time_range_udf.rs

@@ -50,9 +50,12 @@ pub(crate) static TIME_RANGE_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
 pub fn time_range_expr_impl() -> ScalarFunctionImplementation {
     let func = move |args: &[ArrayRef]| -> datafusion::error::Result<ArrayRef> {
         if args.len() != 3 {
-            return Err(DataFusionError::SQL(ParserError::ParserError(
-                "UDF params should be: time_range(field, start, end)".to_string(),
-            )));
+            return Err(DataFusionError::SQL(
+                ParserError::ParserError(
+                    "UDF params should be: time_range(field, start, end)".to_string(),
+                ),
+                None,
+            ));
         }
 
         // 1. cast both arguments to Union. These casts MUST be aligned with the signature or this

src/service/search/grpc/errors.rs

@@ -36,10 +36,13 @@ fn get_key_from_error(err: &str, pos: usize) -> Option<String> {
 
 impl From<DataFusionError> for Error {
     fn from(err: DataFusionError) -> Self {
-        if let DataFusionError::SchemaError(SchemaError::FieldNotFound {
-            field,
-            valid_fields: _,
-        }) = err
+        if let DataFusionError::SchemaError(
+            SchemaError::FieldNotFound {
+                field,
+                valid_fields: _,
+            },
+            _,
+        ) = err
         {
             return Error::ErrorCode(ErrorCodes::SearchFieldNotFound(field.name));
         }