chore(cubestore): Upgrade DF: Address low-hanging warnings

Author: srh

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -8329,7 +8329,7 @@ async fn limit_pushdown_group(service: Box<dyn SqlClient>) {
         .await
         .unwrap();
 
-    let mut res = assert_limit_pushdown(
+    let res = assert_limit_pushdown(
         &service,
         "SELECT id, SUM(n) FROM (
                 SELECT * FROM foo.pushdown1

rust/cubestore/cubestore/src/cluster/mod.rs

@@ -45,7 +45,6 @@ use crate::telemetry::tracing::{TraceIdAndSpanId, TracingHelper};
 use crate::CubeError;
 use async_trait::async_trait;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::error::ArrowError;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::cube_ext;
 use datafusion::error::DataFusionError;

rust/cubestore/cubestore/src/cluster/worker_pool.rs

@@ -460,15 +460,12 @@ mod tests {
     use std::time::Duration;
 
     use async_trait::async_trait;
-    use datafusion::arrow::datatypes::{DataType, Field, Schema};
-    use datafusion::dfschema::ToDFSchema;
     use futures_timer::Delay;
     use serde::{Deserialize, Serialize};
     use tokio::runtime::{Builder, Runtime};
 
     use crate::cluster::worker_pool::{worker_main, WorkerPool};
     use crate::config::Config;
-    use crate::queryplanner::serialized_plan::SerializedLogicalPlan;
     use crate::util::respawn;
     use crate::CubeError;
     use datafusion::cube_ext;
@@ -654,22 +651,6 @@ mod tests {
         });
     }
 
-    // TODO upgrade DF
-    // #[tokio::test]
-    // async fn serialize_plan() -> Result<(), CubeError> {
-    //     let schema = Schema::new(vec![
-    //         Field::new("c1", DataType::Int64, false),
-    //         Field::new("c2", DataType::Utf8, false),
-    //     ]);
-    //     let plan = SerializedLogicalPlan::EmptyRelation {
-    //         produce_one_row: false,
-    //         schema: schema.to_dfschema_ref()?,
-    //     };
-    //     let bytes = bincode::serialize(&plan)?;
-    //     bincode::deserialize::<SerializedLogicalPlan>(bytes.as_slice())?;
-    //     Ok(())
-    // }
-
     type TestServicePool = WorkerPool<ServConfigurator, ServProcessor, ServTransport>;
 
     #[derive(Debug)]

rust/cubestore/cubestore/src/queryplanner/check_memory.rs

@@ -1,12 +1,11 @@
 use crate::util::memory::MemoryHandler;
 use async_trait::async_trait;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::error::Result as ArrowResult;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::error::DataFusionError;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties, RecordBatchStream,
+    DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream,
     SendableRecordBatchStream,
 };
 use flatbuffers::bitflags::_core::any::Any;
@@ -33,7 +32,7 @@ impl CheckMemoryExec {
 }
 
 impl DisplayAs for CheckMemoryExec {
-    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "CheckMemoryExec")
     }
 }

rust/cubestore/cubestore/src/queryplanner/filter_by_key_range.rs

@@ -4,12 +4,11 @@ use crate::table::data::cmp_partition_key;
 use async_trait::async_trait;
 use datafusion::arrow::array::ArrayRef;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::error::ArrowError;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::error::DataFusionError;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, Partitioning, PlanProperties,
+    DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
     SendableRecordBatchStream,
 };
 use futures::StreamExt;
@@ -45,7 +44,7 @@ impl FilterByKeyRangeExec {
 }
 
 impl DisplayAs for FilterByKeyRangeExec {
-    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "FilterByKeyRangeExec")
     }
 }

rust/cubestore/cubestore/src/queryplanner/flatten_union.rs

@@ -1,11 +1,10 @@
 use datafusion::common::tree_node::Transformed;
 use datafusion::common::DFSchema;
 use datafusion::error::DataFusionError;
-use datafusion::execution::context::ExecutionProps;
 use datafusion::logical_expr::{LogicalPlan, Union};
 use datafusion::optimizer::optimizer::OptimizerRule;
-use datafusion::optimizer::{utils, OptimizerConfig};
-use std::fmt::{Debug, Formatter};
+use datafusion::optimizer::OptimizerConfig;
+use std::fmt::Debug;
 use std::sync::Arc;
 
 #[derive(Debug)]

rust/cubestore/cubestore/src/queryplanner/merge_sort.rs

@@ -1,10 +1,9 @@
 use async_trait::async_trait;
 use datafusion::arrow::array::{
-    build_compare, make_comparator, ArrayRef, BooleanArray, DynComparator, RecordBatch,
+    make_comparator, ArrayRef, BooleanArray, DynComparator, RecordBatch,
 };
 use datafusion::arrow::compute::{filter_record_batch, SortOptions};
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::error::ArrowError;
 use datafusion::error::DataFusionError;
 use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::expressions::Column;
@@ -56,7 +55,7 @@ impl LastRowByUniqueKeyExec {
 }
 
 impl DisplayAs for LastRowByUniqueKeyExec {
-    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "LastRowByUniqueKeyExec")
     }
 }

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -19,7 +19,7 @@ pub mod trace_data_loaded;
 use rewrite_inlist_literals::RewriteInListLiterals;
 use serialized_plan::PreSerializedPlan;
 pub use topk::MIN_TOPK_STREAM_ROWS;
-use udfs::{aggregate_udf_by_kind, registerable_aggregate_udfs, registerable_scalar_udfs};
+use udfs::{registerable_aggregate_udfs, registerable_scalar_udfs};
 mod filter_by_key_range;
 mod flatten_union;
 pub mod info_schema;
@@ -38,7 +38,6 @@ use crate::config::ConfigObj;
 use crate::metastore::multi_index::MultiPartition;
 use crate::metastore::table::{Table, TablePath};
 use crate::metastore::{IdRow, MetaStore};
-use crate::queryplanner::flatten_union::FlattenUnion;
 use crate::queryplanner::info_schema::{
     ColumnsInfoSchemaTableDef, RocksDBPropertiesTableDef, SchemataInfoSchemaTableDef,
     SystemCacheTableDef, SystemChunksTableDef, SystemIndexesTableDef, SystemJobsTableDef,
@@ -53,13 +52,11 @@ use crate::queryplanner::query_executor::{
     batches_to_dataframe, ClusterSendExec, InlineTableProvider,
 };
 use crate::queryplanner::serialized_plan::SerializedPlan;
-use crate::queryplanner::topk::{ClusterAggregateTopKUpper, ClusterAggregateTopKLower};
-// use crate::queryplanner::udfs::aggregate_udf_by_kind;
-use crate::queryplanner::udfs::{scalar_udf_by_kind, CubeAggregateUDFKind, CubeScalarUDFKind};
+use crate::queryplanner::topk::ClusterAggregateTopKLower;
 
 use crate::queryplanner::metadata_cache::MetadataCacheFactory;
 use crate::queryplanner::optimizations::rolling_optimizer::RollingOptimizerRule;
-use crate::queryplanner::pretty_printers::{pp_plan, pp_plan_ext, PPOptions};
+use crate::queryplanner::pretty_printers::{pp_plan_ext, PPOptions};
 use crate::sql::cache::SqlResultCache;
 use crate::sql::InlineTables;
 use crate::store::DataFrame;
@@ -74,17 +71,14 @@ use datafusion::catalog::Session;
 use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
 use datafusion::common::{plan_datafusion_err, TableReference};
 use datafusion::config::ConfigOptions;
-use datafusion::datasource::physical_plan::ParquetFileReaderFactory;
-use datafusion::datasource::{provider_as_source, DefaultTableSource, TableType};
+use datafusion::datasource::{provider_as_source, TableType};
 use datafusion::error::DataFusionError;
 use datafusion::execution::{SessionState, TaskContext};
 use datafusion::logical_expr::{
     AggregateUDF, Expr, Extension, LogicalPlan, ScalarUDF, TableProviderFilterPushDown,
     TableSource, WindowUDF,
 };
 use datafusion::physical_expr::EquivalenceProperties;
-// TODO upgrade DF
-// use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{
     collect, DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning,
@@ -94,8 +88,6 @@ use datafusion::prelude::{SessionConfig, SessionContext};
 use datafusion::sql::parser::Statement;
 use datafusion::sql::planner::{ContextProvider, SqlToRel};
 use datafusion::{cube_ext, datasource::TableProvider};
-use futures::TryStreamExt;
-use futures_util::TryFutureExt;
 use log::{debug, trace};
 use mockall::automock;
 use serde_derive::{Deserialize, Serialize};
@@ -808,7 +800,7 @@ impl fmt::Debug for InfoSchemaTableExec {
 }
 
 impl DisplayAs for InfoSchemaTableExec {
-    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "InfoSchemaTableExec")
     }
 }

rust/cubestore/cubestore/src/queryplanner/optimizations/mod.rs

@@ -12,11 +12,8 @@ use crate::queryplanner::optimizations::distributed_partial_aggregate::{
 use std::fmt::{Debug, Formatter};
 // use crate::queryplanner::optimizations::prefer_inplace_aggregates::try_switch_to_inplace_aggregates;
 use super::serialized_plan::PreSerializedPlan;
-use crate::queryplanner::optimizations::prefer_inplace_aggregates::try_regroup_columns;
 use crate::queryplanner::planning::CubeExtensionPlanner;
-use crate::queryplanner::pretty_printers::{pp_phys_plan, pp_plan};
 use crate::queryplanner::rolling::RollingWindowPlanner;
-use crate::queryplanner::serialized_plan::SerializedPlan;
 use crate::queryplanner::trace_data_loaded::DataLoadedSize;
 use crate::util::memory::MemoryHandler;
 use async_trait::async_trait;
@@ -129,7 +126,7 @@ impl PhysicalOptimizerRule for PreOptimizeRule {
     fn optimize(
         &self,
         plan: Arc<dyn ExecutionPlan>,
-        config: &ConfigOptions,
+        _config: &ConfigOptions,
     ) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
         pre_optimize_physical_plan(
             plan,

rust/cubestore/cubestore/src/queryplanner/optimizations/prefer_inplace_aggregates.rs

@@ -1,10 +1,8 @@
 use crate::queryplanner::planning::WorkerExec;
 use crate::queryplanner::query_executor::ClusterSendExec;
-use datafusion::arrow::compute::SortOptions;
 use datafusion::error::DataFusionError;
-use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion::physical_expr::LexOrdering;
 use datafusion::physical_plan::aggregates::AggregateExec;
-use datafusion::physical_plan::expressions::Column;
 use datafusion::physical_plan::filter::FilterExec;
 use datafusion::physical_plan::projection::ProjectionExec;
 use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;