Fix scan2 CI regressions

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

vortex-datafusion/src/convert/stats.rs

@@ -64,6 +64,30 @@ pub(crate) fn aggregate_stats_to_df(
     })
 }
 
+pub(crate) fn aggregate_stats_to_df_as(
+    stats: &[VortexPrecision<Scalar>],
+    target_dtype: &DType,
+) -> VortexResult<ColumnStatistics> {
+    if stats.len() != BYTE_SIZE_INDEX + 1 {
+        return Err(vortex_err!(
+            "expected {} aggregate statistics, got {}",
+            BYTE_SIZE_INDEX + 1,
+            stats.len()
+        ));
+    }
+
+    Ok(ColumnStatistics {
+        null_count: scalar_u64_to_df_usize(&stats[NULL_COUNT_INDEX])?,
+        min_value: scalar_to_df_as(&stats[MIN_INDEX], target_dtype)?,
+        max_value: scalar_to_df_as(&stats[MAX_INDEX], target_dtype)?,
+        // Match the V1 file-statistics path: min/max/null count/size are useful for pruning,
+        // while sum has narrower type expectations in DataFusion and is not used for pruning.
+        sum_value: Precision::Absent,
+        distinct_count: Precision::Absent,
+        byte_size: scalar_u64_to_df_usize(&stats[BYTE_SIZE_INDEX])?,
+    })
+}
+
 /// Convert a stats set for an array with the given dtype.
 #[allow(dead_code)]
 pub(crate) fn stats_set_to_df(
@@ -145,6 +169,22 @@ fn scalar_to_df(stat: &VortexPrecision<Scalar>) -> VortexResult<Precision<Scalar
     }
 }
 
+fn scalar_to_df_as(
+    stat: &VortexPrecision<Scalar>,
+    target_dtype: &DType,
+) -> VortexResult<Precision<ScalarValue>> {
+    let cast = |scalar: &Scalar| scalar.cast(target_dtype).and_then(|s| s.try_to_df()).ok();
+    Ok(match stat {
+        VortexPrecision::Exact(scalar) => cast(scalar)
+            .map(Precision::Exact)
+            .unwrap_or(Precision::Absent),
+        VortexPrecision::Inexact(scalar) => cast(scalar)
+            .map(Precision::Inexact)
+            .unwrap_or(Precision::Absent),
+        VortexPrecision::Absent => Precision::Absent,
+    })
+}
+
 fn scalar_u64_to_df_usize(stat: &VortexPrecision<Scalar>) -> VortexResult<Precision<usize>> {
     match stat {
         VortexPrecision::Exact(scalar) => Ok(Precision::Exact(scalar_u64_to_usize(scalar)?)),

vortex-datafusion/src/persistent/format.rs

@@ -71,7 +71,7 @@ use super::sink::VortexSink;
 use super::source::VortexSource;
 use crate::PrecisionExt as _;
 use crate::convert::TryToDataFusion;
-use crate::convert::stats::aggregate_stats_to_df;
+use crate::convert::stats::aggregate_stats_to_df_as;
 use crate::convert::stats::column_statistics_aggregate_fns;
 use crate::convert::stats::is_constant_to_distinct_count;
 
@@ -755,10 +755,22 @@ async fn infer_scan_plan_stats(table_schema: &SchemaRef, vxf: &VortexFile) -> DF
     let mut column_statistics = vec![ColumnStatistics::default(); table_schema.fields().len()];
     let mut requested_columns = Vec::new();
     let mut requested_exprs = Vec::new();
+    let mut requested_dtypes = Vec::new();
 
     for (idx, field) in table_schema.fields().iter().enumerate() {
         if struct_dtype.find(field.name()).is_some() {
+            let target_dtype = vxf
+                .session()
+                .arrow()
+                .from_arrow_field(field.as_ref())
+                .map_err(|e| {
+                    DataFusionError::Execution(format!(
+                        "Failed to derive Vortex DType for field {}: {e}",
+                        field.name()
+                    ))
+                })?;
             requested_columns.push(idx);
+            requested_dtypes.push(target_dtype);
             requested_exprs.push(vortex::expr::get_item(
                 field.name().as_str(),
                 vortex::expr::root(),
@@ -770,10 +782,15 @@ async fn infer_scan_plan_stats(table_schema: &SchemaRef, vxf: &VortexFile) -> DF
         .scan_plan_statistics_many(&requested_exprs, &funcs)
         .await
         .map_err(|e| DataFusionError::Execution(format!("Failed to infer scan2 stats: {e}")))?;
-    for (column_idx, stats) in requested_columns.into_iter().zip(stats) {
-        column_statistics[column_idx] = aggregate_stats_to_df(&stats).map_err(|e| {
-            DataFusionError::Execution(format!("Failed to convert scan2 stats: {e}"))
-        })?;
+    for ((column_idx, target_dtype), stats) in requested_columns
+        .into_iter()
+        .zip(requested_dtypes)
+        .zip(stats)
+    {
+        column_statistics[column_idx] =
+            aggregate_stats_to_df_as(&stats, &target_dtype).map_err(|e| {
+                DataFusionError::Execution(format!("Failed to convert scan2 stats: {e}"))
+            })?;
     }
 
     let total_byte_size = column_statistics

vortex-duckdb/src/exporter/sequence.rs

@@ -24,7 +24,9 @@ pub(crate) fn new_exporter_with_flatten(
     ctx: &mut ExecutionCtx,
     flatten: bool,
 ) -> VortexResult<Box<dyn ColumnExporter>> {
-    if flatten {
+    if flatten || array.ptype().is_unsigned_int() {
+        // DuckDB's native SequenceVector helper does not support unsigned vectors such as the
+        // `file_row_number` UBIGINT virtual column, so export those through the primitive path.
         return canonical::new_exporter(array.clone().into_array(), cache, ctx);
     }
     Ok(Box::new(SequenceExporter {

vortex-scan/src/plan/data_source.rs

@@ -29,6 +29,7 @@ use vortex_array::expr::Expression;
 use vortex_array::expr::forms::conjuncts;
 use vortex_array::expr::stats::Precision;
 use vortex_array::expr::stats::Stat;
+use vortex_array::expr::transform::coerce_expression;
 use vortex_array::extension::datetime::AnyTemporal;
 use vortex_array::scalar::Scalar;
 use vortex_array::scalar::ScalarValue;
@@ -360,7 +361,8 @@ impl ScanPlanDataSource {
         scan_request: DataSourceScanRequest,
         target_partitions: usize,
     ) -> VortexResult<DataSourceScanRef> {
-        let dtype = scan_request.projection.return_dtype(&self.dtype)?;
+        let dtype = normalize_scan_expr(scan_request.projection.clone(), &self.dtype)?
+            .return_dtype(&self.dtype)?;
 
         let meta = ScanMeta {
             label: Some("scan2".to_string()),
@@ -483,7 +485,8 @@ impl DataSource for ScanPlanDataSource {
         let provider = self.session.scan_scheduler_provider();
         let scheduler = provider.scheduler_for_scan(&meta);
 
-        let dtype = scan_request.projection.return_dtype(&self.dtype)?;
+        let dtype = normalize_scan_expr(scan_request.projection.clone(), &self.dtype)?
+            .return_dtype(&self.dtype)?;
         let limit_remaining = scan_request.limit.map(AtomicU64::new).map(Arc::new);
 
         Ok(Arc::new(ScanPlanDataSourceScan {
@@ -706,7 +709,8 @@ fn scan_plan_binding_stream(
     session: VortexSession,
     request: DataSourceScanRequest,
 ) -> VortexResult<SendableArrayStream> {
-    let output_dtype = request.projection.return_dtype(binding.root().dtype())?;
+    let output_dtype = normalize_scan_expr(request.projection.clone(), binding.root().dtype())?
+        .return_dtype(binding.root().dtype())?;
     let meta = ScanMeta {
         label: Some("scan2".to_string()),
     };
@@ -2245,12 +2249,12 @@ impl PreparedScanPlan {
         request: &DataSourceScanRequest,
     ) -> VortexResult<Self> {
         let dtype = binding.root().dtype();
-        let return_dtype = request.projection.return_dtype(dtype)?;
-        let projection = request.projection.optimize_recursive(dtype)?;
+        let projection = normalize_scan_expr(request.projection.clone(), dtype)?;
+        let return_dtype = projection.return_dtype(dtype)?;
         let filter = request
             .filter
             .clone()
-            .map(|filter| filter.optimize_recursive(dtype))
+            .map(|filter| normalize_scan_expr(filter, dtype))
             .transpose()?;
 
         let root = binding.root();
@@ -3002,6 +3006,10 @@ fn push_expr(
         .ok_or_else(|| vortex_err!("scan2 could not push expression {expr}"))
 }
 
+fn normalize_scan_expr(expr: Expression, dtype: &DType) -> VortexResult<Expression> {
+    coerce_expression(expr, dtype)?.optimize_recursive(dtype)
+}
+
 fn validate_temporal_comparisons(expr: &Expression, scope: &DType) -> VortexResult<()> {
     for child in expr.children().iter() {
         validate_temporal_comparisons(child, scope)?;