feat: upgrade datafusion to v50.0.0 and update related dependencies f… (#563)

* feat: upgrade datafusion to v50.0.0 and update related dependencies for compatibility

* fix

* fmt

crates/policy/src/policy/function.rs

@@ -173,6 +173,7 @@ impl PartialEq for Functions {
 }
 
 #[derive(Clone, Serialize, Deserialize)]
+#[allow(dead_code)]
 pub struct Value;
 
 #[cfg(test)]

crates/s3select-api/src/object_store.rs

@@ -19,19 +19,10 @@ use futures::pin_mut;
 use futures::{Stream, StreamExt};
 use futures_core::stream::BoxStream;
 use http::HeaderMap;
-use object_store::Attributes;
-use object_store::GetOptions;
-use object_store::GetResult;
-use object_store::ListResult;
-use object_store::MultipartUpload;
-use object_store::ObjectMeta;
-use object_store::ObjectStore;
-use object_store::PutMultipartOpts;
-use object_store::PutOptions;
-use object_store::PutPayload;
-use object_store::PutResult;
-use object_store::path::Path;
-use object_store::{Error as o_Error, Result};
+use object_store::{
+    Attributes, Error as o_Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
+    PutMultipartOptions, PutOptions, PutPayload, PutResult, Result, path::Path,
+};
 use pin_project_lite::pin_project;
 use rustfs_common::DEFAULT_DELIMITER;
 use rustfs_ecstore::StorageAPI;
@@ -102,7 +93,7 @@ impl ObjectStore for EcObjectStore {
         unimplemented!()
     }
 
-    async fn put_multipart_opts(&self, _location: &Path, _opts: PutMultipartOpts) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(&self, _location: &Path, _opts: PutMultipartOptions) -> Result<Box<dyn MultipartUpload>> {
         unimplemented!()
     }
 
@@ -122,7 +113,7 @@ impl ObjectStore for EcObjectStore {
         let meta = ObjectMeta {
             location: location.clone(),
             last_modified: Utc::now(),
-            size: reader.object_info.size as usize,
+            size: reader.object_info.size as u64,
             e_tag: reader.object_info.etag,
             version: None,
         };
@@ -151,12 +142,12 @@ impl ObjectStore for EcObjectStore {
         Ok(GetResult {
             payload,
             meta,
-            range: 0..reader.object_info.size as usize,
+            range: 0..reader.object_info.size as u64,
             attributes,
         })
     }
 
-    async fn get_ranges(&self, _location: &Path, _ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, _location: &Path, _ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         unimplemented!()
     }
 
@@ -175,7 +166,7 @@ impl ObjectStore for EcObjectStore {
         Ok(ObjectMeta {
             location: location.clone(),
             last_modified: Utc::now(),
-            size: info.size as usize,
+            size: info.size as u64,
             e_tag: info.etag,
             version: None,
         })
@@ -185,7 +176,7 @@ impl ObjectStore for EcObjectStore {
         unimplemented!()
     }
 
-    fn list(&self, _prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
+    fn list(&self, _prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
         unimplemented!()
     }
 

crates/s3select-query/src/metadata/mod.rs

@@ -61,7 +61,7 @@ impl MetadataProvider {
         Self {
             provider,
             current_session_table_provider,
-            config_options: session.inner().config_options().clone(),
+            config_options: session.inner().config_options().as_ref().clone(),
             session,
             func_manager,
         }

crates/s3select-query/src/sql/logical/optimizer.rs

@@ -26,7 +26,6 @@ use datafusion::{
         propagate_empty_relation::PropagateEmptyRelation, push_down_filter::PushDownFilter, push_down_limit::PushDownLimit,
         replace_distinct_aggregate::ReplaceDistinctWithAggregate, scalar_subquery_to_join::ScalarSubqueryToJoin,
         simplify_expressions::SimplifyExpressions, single_distinct_to_groupby::SingleDistinctToGroupBy,
-        unwrap_cast_in_comparison::UnwrapCastInComparison,
     },
 };
 use rustfs_s3select_api::{
@@ -66,7 +65,6 @@ impl Default for DefaultLogicalOptimizer {
         let rules: Vec<Arc<dyn OptimizerRule + Send + Sync>> = vec![
             // df default rules start
             Arc::new(SimplifyExpressions::new()),
-            Arc::new(UnwrapCastInComparison::new()),
             Arc::new(ReplaceDistinctWithAggregate::new()),
             Arc::new(EliminateJoin::new()),
             Arc::new(DecorrelatePredicateSubquery::new()),
@@ -91,7 +89,6 @@ impl Default for DefaultLogicalOptimizer {
             // The previous optimizations added expressions and projections,
             // that might benefit from the following rules
             Arc::new(SimplifyExpressions::new()),
-            Arc::new(UnwrapCastInComparison::new()),
             Arc::new(CommonSubexprEliminate::new()),
             // PushDownProjection can pushdown Projections through Limits, do PushDownLimit again.
             Arc::new(PushDownLimit::new()),