feat(sqlite): validate payload types at build time

PR #27 closed the reachable LargeUtf8/Utf8View/list read-back gaps, but
the underlying defect class remained: the write side accepted any Arrow
type (silent TEXT/NULL fallback) while the read side reconstructed only a
subset, so an unsupported payload column built a "successful" sidecar that
then failed on the first query (runtimedb#631).

Close the class structurally:

- Add `payload_type_supported` as the single source of truth for which
  Arrow types round-trip, and `validate_payload_schema` to gate every
  build entry point (`SqliteSidecarBuilder::begin`, `open_or_build`). An
  unsupported column now fails at index-build time with the column named,
  instead of at query time — which is where runtimedb's index-create step
  will surface it.
- Widen actual support to the common parquet/DuckDB scalar types that
  previously fell through: Boolean, Date32, Date64, and Timestamp (all
  units, time zone preserved on read-back).

Tests: reject a Decimal128 payload at build time; round-trip Boolean,
Date32, and a tz-carrying Timestamp.

Author: anoop-narang

src/sqlite_provider.rs

@@ -25,7 +25,7 @@ use arrow_array::{
     Array, ArrayRef, Float32Array, Float64Array, Int32Array, Int64Array, OffsetSizeTrait,
     RecordBatch, StringArray, UInt32Array, UInt64Array,
 };
-use arrow_schema::{DataType, FieldRef, SchemaRef};
+use arrow_schema::{DataType, FieldRef, SchemaRef, TimeUnit};
 use async_trait::async_trait;
 use datafusion::catalog::{Session, TableProvider};
 use datafusion::common::Result as DFResult;
@@ -213,6 +213,7 @@ impl SqliteLookupProvider {
                 "pool_size must be at least 1".into(),
             ));
         }
+        validate_payload_schema(&schema, 0)?;
         let conn = open_conn(db_path)?;
 
         let table_exists: bool = conn
@@ -336,6 +337,7 @@ impl SqliteSidecarBuilder {
                 value_col_indices.len()
             )));
         }
+        validate_payload_schema(&schema, key_col_index)?;
         let (create_sql, insert_sql) = ddl(table_name, &schema);
         let conn = open_conn(db_path)?;
         // Manual BEGIN/COMMIT rather than a borrowed `Transaction` so the
@@ -981,11 +983,85 @@ fn build_table(
 
 // ── Type conversion helpers ───────────────────────────────────────────────────
 
+/// Single source of truth for which Arrow types this provider can store in the
+/// sidecar **and** reconstruct on read-back. Every type that returns `true` here
+/// must have a matching arm in all of `arrow_cell_to_sql` (write), `arrow_type_to_sql`
+/// (column affinity), and `sql_values_to_arrow` (read). `validate_payload_schema`
+/// gates every build against this set so an unsupported column fails at index
+/// build time — not later at query time, half a sidecar in.
+fn payload_type_supported(dt: &DataType) -> bool {
+    match dt {
+        DataType::Boolean
+        | DataType::Int32
+        | DataType::Int64
+        | DataType::UInt32
+        | DataType::UInt64
+        | DataType::Float32
+        | DataType::Float64
+        | DataType::Utf8
+        | DataType::LargeUtf8
+        | DataType::Utf8View
+        | DataType::Date32
+        | DataType::Date64
+        | DataType::Timestamp(_, _) => true,
+        DataType::List(item) | DataType::LargeList(item) => {
+            list_item_type_supported(item.data_type())
+        }
+        _ => false,
+    }
+}
+
+/// Item types reconstructable by `json_text_to_list`. Lists are serialized to
+/// JSON TEXT, so only types with a JSON representation `serialize_list` writes
+/// and `json_text_to_list` reads are supported.
+fn list_item_type_supported(dt: &DataType) -> bool {
+    matches!(
+        dt,
+        DataType::Utf8
+            | DataType::LargeUtf8
+            | DataType::Utf8View
+            | DataType::Int32
+            | DataType::Int64
+    )
+}
+
+/// Reject a build whose payload columns include a type the read-back path cannot
+/// reconstruct. Called from every build entry point so the failure surfaces at
+/// index-creation time with the offending column named, rather than as a
+/// `unsupported Arrow type` error on the first query against a sidecar that
+/// "built successfully". `key_col_index` is skipped — keys are validated
+/// separately by `extract_key`.
+fn validate_payload_schema(schema: &SchemaRef, key_col_index: usize) -> DFResult<()> {
+    for (i, field) in schema.fields().iter().enumerate() {
+        if i == key_col_index {
+            continue;
+        }
+        if !payload_type_supported(field.data_type()) {
+            return Err(DataFusionError::Execution(format!(
+                "SqliteLookupProvider: payload column '{}' has unsupported type {:?}. \
+                 Supported types: Boolean, Int32/64, UInt32/64, Float32/64, \
+                 Utf8/LargeUtf8/Utf8View, Date32/64, Timestamp, and List/LargeList \
+                 of {{Utf8, LargeUtf8, Utf8View, Int32, Int64}}.",
+                field.name(),
+                field.data_type()
+            )));
+        }
+    }
+    Ok(())
+}
+
 fn arrow_type_to_sql(dt: &DataType) -> &'static str {
     match dt {
-        DataType::UInt64 | DataType::UInt32 | DataType::Int32 | DataType::Int64 => "INTEGER",
+        DataType::UInt64
+        | DataType::UInt32
+        | DataType::Int32
+        | DataType::Int64
+        | DataType::Boolean
+        | DataType::Date32
+        | DataType::Date64
+        | DataType::Timestamp(_, _) => "INTEGER",
         DataType::Float32 | DataType::Float64 => "REAL",
-        _ => "TEXT", // Utf8, LargeUtf8, List variants → TEXT (JSON for lists)
+        _ => "TEXT", // Utf8, LargeUtf8, Utf8View, List variants → TEXT (JSON for lists)
     }
 }
 
@@ -1057,7 +1133,54 @@ fn arrow_cell_to_sql(col: &ArrayRef, row: usize) -> SqlValue {
                 .unwrap()
                 .value(row),
         ),
+        DataType::Boolean => SqlValue::Integer(
+            col.as_any()
+                .downcast_ref::<arrow_array::BooleanArray>()
+                .unwrap()
+                .value(row) as i64,
+        ),
+        DataType::Date32 => SqlValue::Integer(
+            col.as_any()
+                .downcast_ref::<arrow_array::Date32Array>()
+                .unwrap()
+                .value(row) as i64,
+        ),
+        DataType::Date64 => SqlValue::Integer(
+            col.as_any()
+                .downcast_ref::<arrow_array::Date64Array>()
+                .unwrap()
+                .value(row),
+        ),
+        // Timestamps are stored as their raw i64 tick count; the unit and time
+        // zone live in the schema and are restored on read-back.
+        DataType::Timestamp(unit, _) => {
+            let v = match unit {
+                TimeUnit::Second => col
+                    .as_any()
+                    .downcast_ref::<arrow_array::TimestampSecondArray>()
+                    .unwrap()
+                    .value(row),
+                TimeUnit::Millisecond => col
+                    .as_any()
+                    .downcast_ref::<arrow_array::TimestampMillisecondArray>()
+                    .unwrap()
+                    .value(row),
+                TimeUnit::Microsecond => col
+                    .as_any()
+                    .downcast_ref::<arrow_array::TimestampMicrosecondArray>()
+                    .unwrap()
+                    .value(row),
+                TimeUnit::Nanosecond => col
+                    .as_any()
+                    .downcast_ref::<arrow_array::TimestampNanosecondArray>()
+                    .unwrap()
+                    .value(row),
+            };
+            SqlValue::Integer(v)
+        }
         DataType::List(_) | DataType::LargeList(_) => SqlValue::Text(serialize_list(col, row)),
+        // Unreachable for any schema that passed `validate_payload_schema`; kept
+        // as a defensive fallback rather than a panic.
         _ => SqlValue::Null,
     }
 }
@@ -1239,6 +1362,61 @@ fn sql_values_to_arrow(dt: &DataType, values: Vec<SqlValue>) -> DFResult<ArrayRe
             }
             Arc::new(b.finish())
         }
+        DataType::Boolean => {
+            let arr: arrow_array::BooleanArray = values
+                .iter()
+                .map(|v| match v {
+                    SqlValue::Integer(i) => Some(*i != 0),
+                    _ => None,
+                })
+                .collect();
+            Arc::new(arr)
+        }
+        DataType::Date32 => {
+            let arr: arrow_array::Date32Array = values
+                .iter()
+                .map(|v| match v {
+                    SqlValue::Integer(i) => Some(*i as i32),
+                    _ => None,
+                })
+                .collect();
+            Arc::new(arr)
+        }
+        DataType::Date64 => {
+            let arr: arrow_array::Date64Array = values
+                .iter()
+                .map(|v| match v {
+                    SqlValue::Integer(i) => Some(*i),
+                    _ => None,
+                })
+                .collect();
+            Arc::new(arr)
+        }
+        // Rebuild the unit-specific array, then restore the schema's time zone so
+        // the reconstructed column's DataType matches the original exactly.
+        DataType::Timestamp(unit, tz) => {
+            let it = values.iter().map(|v| match v {
+                SqlValue::Integer(i) => Some(*i),
+                _ => None,
+            });
+            match unit {
+                TimeUnit::Second => Arc::new(
+                    arrow_array::TimestampSecondArray::from_iter(it).with_timezone_opt(tz.clone()),
+                ),
+                TimeUnit::Millisecond => Arc::new(
+                    arrow_array::TimestampMillisecondArray::from_iter(it)
+                        .with_timezone_opt(tz.clone()),
+                ),
+                TimeUnit::Microsecond => Arc::new(
+                    arrow_array::TimestampMicrosecondArray::from_iter(it)
+                        .with_timezone_opt(tz.clone()),
+                ),
+                TimeUnit::Nanosecond => Arc::new(
+                    arrow_array::TimestampNanosecondArray::from_iter(it)
+                        .with_timezone_opt(tz.clone()),
+                ),
+            }
+        }
         DataType::List(item_field) => json_text_to_list::<i32>(item_field, &values)?,
         DataType::LargeList(item_field) => json_text_to_list::<i64>(item_field, &values)?,
         DataType::Float64 => {

tests/sqlite_provider_test.rs

@@ -781,3 +781,137 @@ async fn test_custom_key_column_name() {
         .unwrap();
     assert_eq!(key_col.value(0), 1);
 }
+
+/// The defect class behind hotdata-dev/runtimedb#631 was that the write side
+/// accepted any Arrow type (silent TEXT/NULL fallback) while the read side only
+/// reconstructed a subset — so an unsupported payload column built a "successful"
+/// sidecar that then exploded on the first query. This asserts the failure now
+/// surfaces at *build* time (`begin`) with the offending column named, which is
+/// where runtimedb's index-create step will catch it.
+#[tokio::test]
+async fn test_unsupported_payload_type_rejected_at_build() {
+    let dir = tempdir().unwrap();
+
+    // Decimal128 is a common parquet type the read-back path cannot reconstruct.
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("rowid", DataType::Int64, false),
+        Field::new("price", DataType::Decimal128(10, 2), true),
+    ]));
+
+    let db_path = dir.path().join("bad.db");
+    let err =
+        SqliteSidecarBuilder::begin(db_path.to_str().unwrap(), "models", 2, schema, 0, vec![1])
+            .map(|_| ())
+            .expect_err("a Decimal128 payload column must be rejected at build time");
+
+    let msg = err.to_string();
+    assert!(
+        msg.contains("price") && msg.contains("unsupported type"),
+        "error should name the offending column and type, got: {msg}"
+    );
+}
+
+/// Round-trips the basic scalar types added alongside the validation gate
+/// (Boolean, Date32, and a timezone-carrying Timestamp). These are exactly the
+/// kinds of columns DuckDB/parquet emit that previously fell through to the
+/// "unsupported Arrow type" error on read-back.
+#[tokio::test]
+async fn test_basic_scalar_types_roundtrip() {
+    use arrow_array::{BooleanArray, Date32Array, TimestampMicrosecondArray};
+    use arrow_schema::TimeUnit;
+
+    let dir = tempdir().unwrap();
+
+    let ts_type = DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()));
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("rowid", DataType::Int64, false),
+        Field::new("flag", DataType::Boolean, true),
+        Field::new("day", DataType::Date32, true),
+        Field::new("ts", ts_type.clone(), true),
+    ]));
+
+    let ts_array =
+        TimestampMicrosecondArray::from(vec![Some(1_000_000_i64), None, Some(3_000_000)])
+            .with_timezone_opt(Some("UTC".to_string()));
+
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(Int64Array::from(vec![1_i64, 2, 3])),
+            Arc::new(BooleanArray::from(vec![Some(true), Some(false), None])),
+            Arc::new(Date32Array::from(vec![
+                Some(19_000_i32),
+                None,
+                Some(19_002),
+            ])),
+            Arc::new(ts_array),
+        ],
+    )
+    .unwrap();
+
+    let db_path = dir.path().join("basics.db");
+    let mut builder = SqliteSidecarBuilder::begin(
+        db_path.to_str().unwrap(),
+        "models",
+        2,
+        schema,
+        0,
+        vec![1, 2, 3],
+    )
+    .unwrap();
+    builder.push_batch(&batch).unwrap();
+    let provider = builder.finish().unwrap();
+
+    let batches = provider
+        .fetch_by_keys(&[1, 3], "rowid", None)
+        .await
+        .unwrap();
+    assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 2);
+
+    // The timestamp column must reconstruct with its original unit *and* time
+    // zone, otherwise the column's DataType won't match the declared schema.
+    for b in &batches {
+        assert_eq!(b.column_by_name("ts").unwrap().data_type(), &ts_type);
+        b.column_by_name("flag")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .expect("flag should reconstruct as BooleanArray");
+        b.column_by_name("day")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Date32Array>()
+            .expect("day should reconstruct as Date32Array");
+    }
+
+    // Spot-check rowid 1's values survived the round-trip.
+    let rowid_one = batches
+        .iter()
+        .flat_map(|b| {
+            let rid = b
+                .column_by_name("rowid")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<Int64Array>()
+                .unwrap();
+            let flag = b
+                .column_by_name("flag")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<BooleanArray>()
+                .unwrap();
+            let day = b
+                .column_by_name("day")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<Date32Array>()
+                .unwrap();
+            (0..b.num_rows())
+                .filter(|&i| rid.value(i) == 1)
+                .map(|i| (flag.value(i), day.value(i)))
+                .collect::<Vec<_>>()
+        })
+        .next()
+        .expect("rowid 1 should be present");
+    assert_eq!(rowid_one, (true, 19_000));
+}