feat: log the execution plan in a more compact fashion (#2906)

Currently we log the debug representation of the plan (twice!) every
time a plan is executed. This is way too much information (e.g. the
object store's debug information is repeated many times). In one
egregious case I was dealing with that had 8k fragments this was
generating 35MB of log data!

This PR changes to an indented display representation of the plan.

rust/lance-datafusion/src/exec.rs

@@ -18,8 +18,9 @@ use datafusion::{
         TaskContext,
     },
     physical_plan::{
-        stream::RecordBatchStreamAdapter, streaming::PartitionStream, DisplayAs, DisplayFormatType,
-        ExecutionPlan, PlanProperties, SendableRecordBatchStream,
+        display::DisplayableExecutionPlan, stream::RecordBatchStreamAdapter,
+        streaming::PartitionStream, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+        SendableRecordBatchStream,
     },
 };
 use datafusion_common::{DataFusionError, Statistics};
@@ -29,7 +30,7 @@ use lazy_static::lazy_static;
 use futures::stream;
 use lance_arrow::SchemaExt;
 use lance_core::Result;
-use log::{info, warn};
+use log::{debug, info, warn};
 
 /// An source execution node created from an existing stream
 ///
@@ -241,6 +242,11 @@ pub fn execute_plan(
     plan: Arc<dyn ExecutionPlan>,
     options: LanceExecutionOptions,
 ) -> Result<SendableRecordBatchStream> {
+    debug!(
+        "Executing plan:\n{}",
+        DisplayableExecutionPlan::new(plan.as_ref()).indent(true)
+    );
+
     let session_ctx = get_session_context(options);
 
     // NOTE: we are only executing the first partition here. Therefore, if

rust/lance/src/dataset/scanner.rs

@@ -46,7 +46,6 @@ use lance_index::{scalar::expression::ScalarIndexExpr, DatasetIndexExt};
 use lance_io::stream::RecordBatchStream;
 use lance_linalg::distance::MetricType;
 use lance_table::format::{Fragment, Index};
-use log::debug;
 use roaring::RoaringBitmap;
 use tracing::{info_span, instrument, Span};
 
@@ -1112,8 +1111,6 @@ impl Scanner {
             plan = rule.optimize(plan, &options)?;
         }
 
-        debug!("Execution plan:\n{:?}", plan);
-
         Ok(plan)
     }