Various fixes to support updated Taxi Dashboard example

vegafusion-common/src/data/table.rs

@@ -100,7 +100,7 @@ impl VegaFusionTable {
         let empty_record_batch = RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new(
             ORDER_COL,
             DataType::UInt64,
-            false,
+            true,
         )])));
         VegaFusionTable::from(empty_record_batch)
     }
@@ -125,7 +125,7 @@ impl VegaFusionTable {
             new_fields.remove(order_col_index);
         }
 
-        new_fields.insert(0, Field::new(ORDER_COL, ORDER_COL_DTYPE, false));
+        new_fields.insert(0, Field::new(ORDER_COL, ORDER_COL_DTYPE, true));
 
         let new_schema = Arc::new(Schema::new(new_fields)) as SchemaRef;
 

vegafusion-core/src/spec/chart.rs

@@ -50,25 +50,8 @@ pub fn default_schema() -> String {
 
 impl ChartSpec {
     pub fn walk(&self, visitor: &mut dyn ChartVisitor) -> Result<()> {
-        // Visit top-level chart
-        visitor.visit_chart(self)?;
-
-        // Top-level with empty scope
-        let scope: Vec<u32> = Vec::new();
-        for data in &self.data {
-            visitor.visit_data(data, &scope)?;
-        }
-        for scale in &self.scales {
-            visitor.visit_scale(scale, &scope)?;
-        }
-        for axis in &self.axes {
-            visitor.visit_axis(axis, &scope)?;
-        }
-        for signal in &self.signals {
-            visitor.visit_signal(signal, &scope)?;
-        }
-
         // Child groups
+        let scope: Vec<u32> = Vec::new();
         let mut group_index = 0;
         for mark in &self.marks {
             if mark.type_ == "group" {
@@ -85,6 +68,23 @@ impl ChartSpec {
             }
         }
 
+        // Top-level with empty scope
+        for data in &self.data {
+            visitor.visit_data(data, &scope)?;
+        }
+        for scale in &self.scales {
+            visitor.visit_scale(scale, &scope)?;
+        }
+        for axis in &self.axes {
+            visitor.visit_axis(axis, &scope)?;
+        }
+        for signal in &self.signals {
+            visitor.visit_signal(signal, &scope)?;
+        }
+
+        // Visit top-level chart
+        visitor.visit_chart(self)?;
+
         Ok(())
     }
 

vegafusion-core/src/spec/visitors.rs

@@ -2,7 +2,7 @@ use crate::expression::parser::parse;
 use crate::proto::gen::tasks::data_url_task::Url;
 use crate::proto::gen::tasks::{
     scan_url_format, DataSourceTask, DataUrlTask, DataValuesTask, ParseFieldSpec, ParseFieldSpecs,
-    ScanUrlFormat, Task, TzConfig, Variable,
+    ScanUrlFormat, Task, TzConfig, Variable, VariableNamespace,
 };
 use crate::proto::gen::transforms::TransformPipeline;
 use crate::spec::chart::{ChartSpec, ChartVisitor};
@@ -24,7 +24,7 @@ use std::convert::TryFrom;
 use std::ops::Deref;
 use vegafusion_common::data::scalar::ScalarValueHelpers;
 use vegafusion_common::data::table::VegaFusionTable;
-use vegafusion_common::error::{Result, VegaFusionError};
+use vegafusion_common::error::Result;
 
 #[derive(Clone, Debug, Default)]
 pub struct MakeTaskScopeVisitor {
@@ -241,9 +241,8 @@ impl<'a> ChartVisitor for MakeTasksVisitor<'a> {
             let expression = parse(update)?;
             Task::new_signal(signal_var, scope, expression, &self.tz_config)
         } else {
-            return Err(VegaFusionError::internal(format!(
-                "Signal must have an initial value or an update expression: {signal:#?}"
-            )));
+            let value = TaskValue::Scalar(ScalarValue::Null);
+            Task::new_value(signal_var, scope, value)
         };
 
         self.tasks.push(task);
@@ -342,6 +341,17 @@ impl<'a> ChartVisitor for UpdateVarsChartVisitor<'a> {
         {
             self.update_vars
                 .insert((Variable::new_signal(&signal.name), Vec::from(scope)));
+        } else {
+            // Check for empty signal that has an update in a child signal of the same name
+            // Note: We rely on the fact that visit_signal is called after child groups have been
+            // visited
+            for v in &self.update_vars {
+                if v.0.namespace == VariableNamespace::Signal as i32 && v.0.name == signal.name {
+                    self.update_vars
+                        .insert((Variable::new_signal(&signal.name), Vec::from(scope)));
+                    break;
+                }
+            }
         }
 
         // Check for signal expressions that have update dependencies

vegafusion-datafusion-udfs/src/udfs/array/span.rs

@@ -1,4 +1,3 @@
-use std::convert::TryFrom;
 use std::sync::Arc;
 use vegafusion_common::arrow::datatypes::{DataType, Field};
 use vegafusion_common::data::scalar::ScalarValueHelpers;
@@ -21,17 +20,17 @@ fn make_span_udf() -> ScalarUDF {
         Ok(match arg {
             ColumnarValue::Scalar(value) => {
                 match value {
+                    ScalarValue::Null => ColumnarValue::Scalar(ScalarValue::from(0.0)),
                     ScalarValue::Float64(_) => {
-                        ColumnarValue::Scalar(ScalarValue::try_from(&DataType::Float64).unwrap())
+                        // Span of scalar (including null) is 0
+                        ColumnarValue::Scalar(ScalarValue::from(0.0))
                     }
                     ScalarValue::List(Some(arr), element_type) => {
                         match element_type.data_type() {
                             DataType::Float64 => {
                                 if arr.is_empty() {
-                                    // Span of empty array is null
-                                    ColumnarValue::Scalar(
-                                        ScalarValue::try_from(&DataType::Float64).unwrap(),
-                                    )
+                                    // Span of empty array is 0
+                                    ColumnarValue::Scalar(ScalarValue::from(0.0))
                                 } else {
                                     let first = arr.first().unwrap().to_f64().unwrap();
                                     let last = arr.last().unwrap().to_f64().unwrap();
@@ -61,6 +60,7 @@ fn make_span_udf() -> ScalarUDF {
             1,
             vec![
                 DataType::Float64, // For null
+                DataType::Null,    // For null
                 DataType::List(Box::new(Field::new("item", DataType::Float64, true))),
             ],
             Volatility::Immutable,

vegafusion-datafusion-udfs/src/udfs/object/mod.rs

@@ -9,7 +9,7 @@ pub fn make_object_constructor_udf(keys: &[String], value_types: &[DataType]) ->
     let fields: Vec<_> = keys
         .iter()
         .zip(value_types.iter())
-        .map(|(k, dtype)| Field::new(k, dtype.clone(), false))
+        .map(|(k, dtype)| Field::new(k, dtype.clone(), true))
         .collect();
 
     let struct_dtype = DataType::Struct(fields.clone());

vegafusion-runtime/src/expression/compiler/mod.rs

@@ -499,7 +499,7 @@ mod test_compile {
                 &["a".to_string(), "two".to_string()],
                 &[
                     DataType::Float64,
-                    DataType::Struct(vec![Field::new("three", DataType::Float64, false)]),
+                    DataType::Struct(vec![Field::new("three", DataType::Float64, true)]),
                 ],
             )),
             args: vec![
@@ -528,7 +528,10 @@ mod test_compile {
         ]);
 
         println!("value: {result_value:?}");
-        assert_eq!(result_value, expected_value);
+
+        // ScalarValue::from(...) creates a Field with nullable=false. We always use nullable=true,
+        // so compare string repr (which doesn't include nullable info) instead of value
+        assert_eq!(format!("{result_value:?}"), format!("{expected_value:?}"));
     }
 
     #[test]
@@ -570,7 +573,7 @@ mod test_compile {
     fn test_compile_datum_nested_member() {
         let expr = parse("datum['two'].foo * 3").unwrap();
         // let expr = parse("[datum['two'].foo * 3, datum['two'].foo]").unwrap();
-        let foo_field = Field::new("foo", DataType::Float64, false);
+        let foo_field = Field::new("foo", DataType::Float64, true);
 
         let two_type = DataType::Struct(vec![foo_field]);
         let two_field = Field::new("two", two_type, true);

vegafusion-runtime/src/transform/bin.rs

@@ -175,7 +175,9 @@ pub fn calculate_bin_params(
         let span_expr = compile(span_expression, config, Some(schema))?;
         let span_scalar = span_expr.eval_to_scalar()?;
         if let Ok(span_f64) = span_scalar.to_f64() {
-            span = span_f64;
+            if span_f64 > 0.0 {
+                span = span_f64;
+            }
         }
     }
 

vegafusion-runtime/src/transform/sequence.rs

@@ -63,7 +63,7 @@ impl TransformTrait for Sequence {
         let data_schema = Arc::new(Schema::new(vec![Field::new(
             &col_name,
             DataType::Float64,
-            false,
+            true,
         )])) as SchemaRef;
         let data_batch = RecordBatch::try_new(data_schema, vec![data_array])?;
         let data_table = VegaFusionTable::from(data_batch);