ARROW-5950: [Rust][DataFusion] Add logger

It seems useful to me to do some extra (debug, info, etc) logging.

I added to the benchmark.
Executing with `RUST_LOG="debug"` gives (next to sqlparser logs):

```
[2020-12-21T23:04:59Z DEBUG datafusion::execution::context] Logical plan:
     Sort: #l_returnflag ASC NULLS FIRST, #l_linestatus ASC NULLS FIRST
      Projection: #l_returnflag, #l_linestatus, #SUM(l_quantity) AS sum_qty, #SUM(l_extendedprice) AS sum_base_price, #SUM(l_extendedprice Multiply Int64(1) Minus l_discount) AS sum_disc_price, #SUM(l_extendedprice Multiply Int64(1) Minus l_discount Multiply Int64(1) Plus l_tax) AS sum_charge, #AVG(l_quantity) AS avg_qty, #AVG(l_extendedprice) AS avg_price, #AVG(l_discount) AS avg_disc, #COUNT(UInt8(1)) AS count_order
        Aggregate: groupBy=[[#l_returnflag, #l_linestatus]], aggr=[[SUM(#l_quantity), SUM(#l_extendedprice), SUM(#l_extendedprice Multiply Int64(1) Minus #l_discount), SUM(#l_extendedprice Multiply Int64(1) Minus #l_discount Multiply Int64(1) Plus #l_tax), AVG(#l_quantity), AVG(#l_extendedprice), AVG(#l_discount), COUNT(UInt8(1))]]
          Filter: #l_shipdate LtEq CAST(Utf8("1998-09-02") AS Date32(Day))
            TableScan: lineitem projection=None
[2020-12-21T23:04:59Z DEBUG datafusion::execution::context] Optimized logical plan:
     Sort: #l_returnflag ASC NULLS FIRST, #l_linestatus ASC NULLS FIRST
      Projection: #l_returnflag, #l_linestatus, #SUM(l_quantity) AS sum_qty, #SUM(l_extendedprice) AS sum_base_price, #SUM(l_extendedprice Multiply Int64(1) Minus l_discount) AS sum_disc_price, #SUM(l_extendedprice Multiply Int64(1) Minus l_discount Multiply Int64(1) Plus l_tax) AS sum_charge, #AVG(l_quantity) AS avg_qty, #AVG(l_extendedprice) AS avg_price, #AVG(l_discount) AS avg_disc, #COUNT(UInt8(1)) AS count_order
        Aggregate: groupBy=[[#l_returnflag, #l_linestatus]], aggr=[[SUM(#l_quantity), SUM(#l_extendedprice), SUM(#l_extendedprice Multiply Int64(1) Minus #l_discount), SUM(#l_extendedprice Multiply Int64(1) Minus #l_discount Multiply Int64(1) Plus #l_tax), AVG(#l_quantity), AVG(#l_extendedprice), AVG(#l_discount), COUNT(UInt8(1))]]
          Filter: #l_shipdate LtEq CAST(Utf8("1998-09-02") AS Date32(Day))
            TableScan: lineitem projection=Some([4, 5, 6, 7, 8, 9, 10])
Query 1 iteration 9 took 802.7 ms
Query 1 avg time: 915.30 ms
```

Closes #8986 from Dandandan/add_logger

Lead-authored-by: Heres, Daniel <[email protected]>
Co-authored-by: Daniël Heres <[email protected]>
Signed-off-by: Andrew Lamb <[email protected]>

Author: Dandandan

rust/arrow-flight/src/arrow.flight.protocol.rs

@@ -498,8 +498,9 @@ pub mod flight_service_server {
     #[async_trait]
     pub trait FlightService: Send + Sync + 'static {
         #[doc = "Server streaming response type for the Handshake method."]
-        type HandshakeStream: Stream<Item = Result<super::HandshakeResponse, tonic::Status>>
-            + Send
+        type HandshakeStream: Stream<
+            Item = Result<super::HandshakeResponse, tonic::Status>,
+        > + Send
             + Sync
             + 'static;
         #[doc = ""]

rust/benchmarks/Cargo.toml

@@ -31,4 +31,5 @@ parquet = { path = "../parquet" }
 datafusion = { path = "../datafusion" }
 structopt = { version = "0.3", default-features = false }
 tokio = { version = "0.2", features = ["macros", "rt-core", "rt-threaded"] }
-futures = "0.3"
+futures = "0.3"
+env_logger = "^0.8"

rust/benchmarks/src/bin/tpch.rs

@@ -109,6 +109,8 @@ async fn main() -> Result<()> {
 }
 
 async fn benchmark(opt: BenchmarkOpt) -> Result<()> {
+    env_logger::init();
+
     println!("Running benchmarks with the following options: {:?}", opt);
     let config = ExecutionConfig::new()
         .with_concurrency(opt.concurrency)

rust/datafusion/Cargo.toml

@@ -60,6 +60,7 @@ async-trait = "0.1.41"
 futures = "0.3"
 pin-project-lite= "^0.2.0"
 tokio = { version = "0.2", features = ["macros", "rt-core", "rt-threaded", "sync"] }
+log = "^0.4"
 
 [dev-dependencies]
 rand = "0.7"

rust/datafusion/src/execution/context.rs

@@ -16,8 +16,8 @@
 // under the License.
 
 //! ExecutionContext contains methods for registering data sources and executing queries
-
 use crate::optimizer::hash_build_probe_order::HashBuildProbeOrder;
+use log::debug;
 use std::fs;
 use std::path::Path;
 use std::string::String;
@@ -316,9 +316,11 @@ impl ExecutionContext {
     /// Optimize the logical plan by applying optimizer rules
     pub fn optimize(&self, plan: &LogicalPlan) -> Result<LogicalPlan> {
         // Apply standard rewrites and optimizations
+        debug!("Logical plan:\n {:?}", plan);
         let mut plan = ProjectionPushDown::new().optimize(&plan)?;
         plan = FilterPushDown::new().optimize(&plan)?;
         plan = HashBuildProbeOrder::new().optimize(&plan)?;
+        debug!("Optimized logical plan:\n {:?}", plan);
 
         self.state
             .lock()