fix(datafusion): coerce numeric literals to and from Float16

rust/lance-datafusion/Cargo.toml

@@ -24,6 +24,7 @@ datafusion-physical-expr.workspace = true
 datafusion-substrait = {workspace = true, optional = true}
 datafusion.workspace = true
 futures.workspace = true
+half.workspace = true
 jsonb = {workspace = true}
 lance-arrow.workspace = true
 lance-core = {workspace = true, features = ["datafusion"]}

rust/lance-index/src/scalar/expression.rs

@@ -2673,6 +2673,10 @@ mod tests {
         parser: Box<MultiQueryParser>,
     }
 
+    fn f16_scalar(value: f32) -> ScalarValue {
+        ScalarValue::Float16(Some(half::f16::from_f32(value)))
+    }
+
     impl ColInfo {
         fn new(data_type: DataType, parser: Box<dyn ScalarQueryParser>) -> Self {
             Self {
@@ -2934,6 +2938,68 @@ mod tests {
         assert!(plan.refine_expr.is_none());
     }
 
+    /// A `Float16` column must reach its scalar index like any other numeric
+    /// column. This is the second, quieter face of the coercion gap: `maybe_scalar`
+    /// runs `safe_coerce_scalar` on a literal the planner has *already* coerced,
+    /// so without a `ScalarValue::Float16` source arm the whole predicate silently
+    /// becomes a refine filter. The rows stay correct, which is why only the plan
+    /// catches it.
+    ///
+    /// This drives `Planner::parse_filter` rather than `check_with_schema` on
+    /// purpose. `check_with_schema` builds the expression with
+    /// `create_logical_expr`, which does no type coercion, so the literal would
+    /// still be `Float64` here and would exercise the `Float64` to `Float16`
+    /// target arm instead. Production coerces first, in `resolve_value`, and only
+    /// this order needs the source arm.
+    #[rstest]
+    #[case("temp = 1.0", SargableQuery::Equals(f16_scalar(1.0)))]
+    #[case("temp = 1", SargableQuery::Equals(f16_scalar(1.0)))]
+    #[case(
+        "temp < 1.0",
+        SargableQuery::Range(Bound::Unbounded, Bound::Excluded(f16_scalar(1.0)))
+    )]
+    #[case(
+        // Four elements so DataFusion's `ShortenInListSimplifier` leaves the list
+        // alone; at three or fewer over a bare column it becomes an `OR` chain and
+        // stops exercising `maybe_scalar_list`.
+        "temp IN (1.0, 2.5, 3.0, 4.0)",
+        SargableQuery::IsIn(vec![
+            f16_scalar(1.0),
+            f16_scalar(2.5),
+            f16_scalar(3.0),
+            f16_scalar(4.0),
+        ])
+    )]
+    fn test_float16_column_reaches_its_index(#[case] expr: &str, #[case] expected: SargableQuery) {
+        let index_info = MockIndexInfoProvider::new(vec![(
+            "temp",
+            ColInfo::new(
+                DataType::Float16,
+                Box::new(SargableQueryParser::new(
+                    "temp_idx".to_string(),
+                    "BTree".to_string(),
+                    false,
+                )),
+            ),
+        )]);
+        let schema = Schema::new(vec![Field::new("temp", DataType::Float16, true)]);
+
+        let planner = Planner::new(Arc::new(schema));
+        let filter = planner.parse_filter(expr).unwrap();
+        let plan = planner
+            .create_filter_plan(filter, &index_info, true)
+            .unwrap();
+        let wanted = IndexedExpression::index_query(
+            "temp".to_string(),
+            "temp_idx".to_string(),
+            "BTree".to_string(),
+            Arc::new(expected),
+        );
+
+        assert_eq!(plan.index_query, wanted.scalar_query, "predicate: {expr}");
+        assert!(plan.refine_expr.is_none(), "predicate: {expr}");
+    }
+
     #[test]
     fn test_expressions() {
         let index_info = MockIndexInfoProvider::new(vec![

rust/lance/tests/query/primitives.rs

@@ -6,7 +6,8 @@ use std::sync::Arc;
 use arrow::datatypes::*;
 use arrow_array::{
     ArrayRef, BinaryArray, BinaryViewArray, Float32Array, Float64Array, Int32Array,
-    LargeBinaryArray, LargeStringArray, RecordBatch, StringArray, StringViewArray,
+    LargeBinaryArray, LargeStringArray, RecordBatch, RecordBatchIterator, StringArray,
+    StringViewArray,
 };
 use arrow_schema::DataType;
 use lance::Dataset;
@@ -16,6 +17,7 @@ use lance::dataset::{InsertBuilder, WriteParams};
 use lance::index::DatasetIndexExt;
 use lance_datagen::{ArrayGeneratorExt, RowCount, array, gen_batch};
 use lance_index::IndexType;
+use lance_index::scalar::ScalarIndexParams;
 
 use super::{test_filter, test_scan, test_take};
 use crate::utils::DatasetTestCases;
@@ -138,6 +140,7 @@ async fn test_btree_nullable_or_with_absent_value() {
 
 #[tokio::test]
 #[rstest::rstest]
+#[case::float16(DataType::Float16)]
 #[case::float32(DataType::Float32)]
 #[case::float64(DataType::Float64)]
 async fn test_query_float(#[case] data_type: DataType) {
@@ -146,17 +149,20 @@ async fn test_query_float(#[case] data_type: DataType) {
         .col("value", array::rand_type(&data_type).with_random_nulls(0.1))
         .into_batch_rows(RowCount::from(60))
         .unwrap();
+    // BloomFilter is left out for Float16 because that index rejects the type
+    // outright. `test_bloom_filter_rejects_float16` pins the rejection so this
+    // skip cannot outlive it.
+    let mut index_types = vec![
+        None,
+        Some(IndexType::BTree),
+        Some(IndexType::Bitmap),
+        Some(IndexType::ZoneMap),
+    ];
+    if data_type != DataType::Float16 {
+        index_types.push(Some(IndexType::BloomFilter));
+    }
     DatasetTestCases::from_data(batch)
-        .with_index_types(
-            "value",
-            [
-                None,
-                Some(IndexType::BTree),
-                Some(IndexType::Bitmap),
-                Some(IndexType::BloomFilter),
-                Some(IndexType::ZoneMap),
-            ],
-        )
+        .with_index_types("value", index_types)
         .run(|ds: Dataset, original: RecordBatch| async move {
             test_scan(&original, &ds).await;
             test_take(&original, &ds).await;
@@ -170,6 +176,44 @@ async fn test_query_float(#[case] data_type: DataType) {
         .await
 }
 
+/// `test_query_float` runs its Float16 case without `IndexType::BloomFilter`
+/// because that index refuses the type. Pin the refusal here so the skip cannot
+/// outlive it: once bloom filters accept Float16 this test fails, and whoever
+/// makes it pass should drop the skip too.
+#[tokio::test]
+async fn test_bloom_filter_rejects_float16() {
+    let batch = gen_batch()
+        .col("value", array::rand_type(&DataType::Float16))
+        .into_batch_rows(RowCount::from(16))
+        .unwrap();
+    let mut ds = Dataset::write(
+        RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema()),
+        "memory://bloom_f16",
+        None,
+    )
+    .await
+    .unwrap();
+
+    let err = ds
+        .create_index(
+            &["value"],
+            IndexType::BloomFilter,
+            None,
+            &ScalarIndexParams::default(),
+            false,
+        )
+        .await
+        .expect_err("bloom filter should still refuse Float16");
+    assert!(
+        matches!(err, lance::Error::InvalidInput { .. }),
+        "unexpected error variant: {err:?}"
+    );
+    assert!(
+        err.to_string().contains("Float16"),
+        "error should name the rejected type: {err}"
+    );
+}
+
 #[tokio::test]
 #[rstest::rstest]
 #[case::float32(DataType::Float32)]