use chunk lengths instead of rows

Author: myrrc

vortex-duckdb/src/duckdb/table_function/mod.rs

@@ -30,8 +30,8 @@ use crate::duckdb::data_chunk::DataChunk;
 use crate::duckdb::expr::Expression;
 use crate::duckdb::table_function::cardinality::cardinality_callback;
 use crate::duckdb::table_function::partition::get_partition_data_callback;
-use crate::duckdb::table_function::table_scan_progress::table_scan_progress_callback;
 use crate::duckdb::table_function::pushdown_complex_filter::pushdown_complex_filter_callback;
+use crate::duckdb::table_function::table_scan_progress::table_scan_progress_callback;
 use crate::duckdb::table_function::virtual_columns::get_virtual_columns_callback;
 use crate::duckdb_try;
 

vortex-duckdb/src/scan.rs

@@ -114,8 +114,8 @@ pub struct VortexGlobalData {
     iterator: ThreadSafeIterator<VortexResult<(ArrayRef, Arc<ConversionCache>)>>,
     batch_id: AtomicU64,
     ctx: ExecutionCtx,
-    rows_total: AtomicU64,
-    rows_read: AtomicU64,
+    chunk_len_total: AtomicU64,
+    chunk_len_read: AtomicU64,
 }
 
 pub struct VortexLocalData {
@@ -348,6 +348,9 @@ impl TableFunction for VortexTableFunction {
                     return Ok(());
                 };
 
+                global_state
+                    .chunk_len_total
+                    .fetch_add(chunk.len(), Ordering::Relaxed);
                 let (array_result, conversion_cache) = result?;
 
                 let array_result = array_result.optimize_recursive()?;
@@ -383,6 +386,10 @@ impl TableFunction for VortexTableFunction {
                 .vortex_expect("error: exporter missing");
 
             let has_more_data = exporter.export(chunk)?;
+            // exporter.export updates chunk length to remaning bytes
+            global_state
+                .chunk_len_read
+                .fetch_add(chunk.len(), Ordering::Relaxed);
 
             if !has_more_data {
                 // This exporter is fully consumed.
@@ -482,8 +489,8 @@ impl TableFunction for VortexTableFunction {
             batch_id: AtomicU64::new(0),
             // TODO(joe): fetch this from somewhere??.
             ctx: ExecutionCtx::new(VortexSession::default()),
-            rows_read: AtomicU64::new(0),
-            rows_total: AtomicU64::new(0)
+            chunk_len_total: AtomicU64::new(0),
+            chunk_len_read: AtomicU64::new(0),
         })
     }
 
@@ -518,8 +525,8 @@ impl TableFunction for VortexTableFunction {
         _bind_data: &mut Self::BindData,
         global_state: &mut Self::GlobalState,
     ) -> f64 {
-        global_state.rows_read.load(Ordering::Relaxed) as f64
-            / global_state.rows_total.load(Ordering::Relaxed) as f64
+        global_state.chunk_len_read.load(Ordering::Relaxed) as f64
+            / global_state.chunk_len_total.load(Ordering::Relaxed) as f64
     }
 
     fn pushdown_complex_filter(