EXPERIMENTAL COMMIT: always canonicalize dict values eagerly

Signed-off-by: Andrew Duffy <[email protected]>

Author: a10y

vortex-datafusion/src/persistent/opener.rs

@@ -251,6 +251,7 @@ impl FileOpener for VortexOpener {
                     FieldName::from(field.name().as_str())
                 })
                 .collect::<Vec<_>>();
+            // println!("FIELDS: {fields:?}");
             let projection_expr = select(fields, root());
 
             // We share our layout readers with others partitions in the scan, so we can only need to read each layout in each file once.

vortex-datafusion/src/persistent/source.rs

@@ -133,6 +133,7 @@ impl FileSource for VortexSource {
         };
 
         let projection = base_config.file_column_projection_indices().map(Arc::from);
+        // println!("CREATE FILE OPENER: projection {projection:?}");
 
         let opener = VortexOpener {
             object_store,

vortex-layout/src/layouts/dict/reader.rs

@@ -8,7 +8,7 @@ use std::sync::{Arc, OnceLock};
 use futures::future::BoxFuture;
 use futures::{FutureExt, TryFutureExt, try_join};
 use vortex_array::compute::{MinMaxResult, min_max, take};
-use vortex_array::{ArrayRef, MaskFuture};
+use vortex_array::{Array, ArrayRef, IntoArray, MaskFuture};
 use vortex_dict::DictArray;
 use vortex_dtype::{DType, FieldMask};
 use vortex_error::{VortexError, VortexExpect, VortexResult};
@@ -76,6 +76,7 @@ impl DictReader {
                     )
                     .vortex_expect("must construct dict values array evaluation")
                     .map_err(Arc::new)
+                    .map_ok(|arr| arr.to_canonical().into_array())
                     .boxed()
                     .shared()
             })

vortex-scan/src/split_by.rs

@@ -39,6 +39,10 @@ impl SplitBy {
                 // Register all splits in the row range for all layouts that are needed
                 // to read the field mask.
                 layout_reader.register_splits(field_mask, row_range, &mut row_splits)?;
+                // Iterate the splits, show how large they are on average
+                //for (&start, &end) in row_splits.iter().tuples() {
+                //    println!("split size: {}", end - start);
+                //}
                 row_splits
             }
             SplitBy::RowCount(n) => row_range

vortex-scan/src/tasks.rs

@@ -45,6 +45,8 @@ pub(super) fn split_exec<A: 'static + Send>(
         return Ok(ok(None).boxed());
     }
 
+    // println!("row_range: {}", row_range.end - row_range.start);
+
     let filter_mask = match ctx.filter.as_ref() {
         // No filter == immediate mask
         None => {