fix(index): align scalar index field type support

python/python/lance/dataset.py

@@ -3245,12 +3245,17 @@ def _prepare_scalar_index_request(
                     and not pa.types.is_boolean(field_type)
                     and not pa.types.is_string(field_type)
                     and not pa.types.is_large_string(field_type)
+                    and not pa.types.is_binary(field_type)
+                    and not pa.types.is_large_binary(field_type)
                     and not pa.types.is_temporal(field_type)
+                    and not pa.types.is_decimal128(field_type)
+                    and not pa.types.is_decimal256(field_type)
                     and not pa.types.is_fixed_size_binary(field_type)
                 ):
                     raise TypeError(
-                        f"BTREE/BITMAP/ZONEMAP index column {column} must be int",
-                        ", float, bool, str, large_str, fixed-size-binary, or temporal",
+                        "BTREE/BITMAP/ZONEMAP index column "
+                        f"{column!r} must have a supported scalar type; "
+                        f"got {field_type!r}"
                     )
             elif index_type == "LABEL_LIST":
                 if not pa.types.is_list(field_type):
@@ -3277,10 +3282,6 @@ def _prepare_scalar_index_request(
                         f" or list of strings, or json, but got {value_type}"
                     )
 
-            if pa.types.is_duration(field_type):
-                raise TypeError(
-                    f"Scalar index column {column} cannot currently be a duration"
-                )
             return column, index_type, index_type
         elif isinstance(index_type, IndexConfig):
             logical_index_type = index_type.index_type.upper()
@@ -3425,7 +3426,8 @@ def create_scalar_index(
         ----------
         column : str
             The column to be indexed.  Must be a boolean, integer, float,
-            or string column.
+            string, binary, decimal128 or decimal256, fixed-size-binary, or
+            supported temporal column.
         index_type : str
             The type of the index.  One of ``"BTREE"``, ``"BITMAP"``,
             ``"LABEL_LIST"``, ``"NGRAM"``, ``"ZONEMAP"``, ``"INVERTED"``,

python/python/tests/test_scalar_index.py

@@ -12,6 +12,7 @@
 import uuid
 import zipfile
 from datetime import date, datetime, timedelta
+from decimal import Decimal
 from pathlib import Path
 
 import lance
@@ -695,24 +696,87 @@ def test_indexed_vector_scan_postfilter(
     assert scanner.to_table().num_rows == 0
 
 
-def test_fixed_size_binary(tmp_path):
-    arr = pa.array([b"0123012301230123", b"2345234523452345"], pa.uuid())
+@pytest.mark.parametrize(
+    "index_type, data_type, values, filter_expr",
+    [
+        pytest.param(
+            "BTREE",
+            pa.uuid(),
+            [b"0123012301230123", b"2345234523452345"],
+            (
+                "value = arrow_cast(0x32333435323334353233343532333435, "
+                "'FixedSizeBinary(16)')"
+            ),
+            id="btree-fixed-size-binary",
+        ),
+        *[
+            pytest.param(
+                index_type,
+                data_type,
+                values,
+                filter_expr,
+                id=f"{index_type.lower()}-{type_name}",
+            )
+            for type_name, data_type, values, filter_expr in [
+                (
+                    "large-string",
+                    pa.large_string(),
+                    ["alpha", "beta", "gamma"],
+                    "value = 'beta'",
+                ),
+                (
+                    "binary",
+                    pa.binary(),
+                    [b"alpha", b"beta", b"gamma"],
+                    "value = arrow_cast(0x62657461, 'Binary')",
+                ),
+                (
+                    "large-binary",
+                    pa.large_binary(),
+                    [b"alpha", b"beta", b"gamma"],
+                    "value = arrow_cast(0x62657461, 'LargeBinary')",
+                ),
+                (
+                    "decimal128",
+                    pa.decimal128(10, 2),
+                    [Decimal("1.00"), Decimal("2.00"), Decimal("3.00")],
+                    "value = arrow_cast(2.00, 'Decimal128(10, 2)')",
+                ),
+                (
+                    "decimal256",
+                    pa.decimal256(76, 2),
+                    [Decimal("1.00"), Decimal("2.00"), Decimal("3.00")],
+                    "value = arrow_cast(2.00, 'Decimal256(76, 2)')",
+                ),
+                (
+                    "duration",
+                    pa.duration("ms"),
+                    [1, 2, 3],
+                    "value = arrow_cast(2, 'Duration(Millisecond)')",
+                ),
+            ]
+            for index_type in ["BTREE", "BITMAP", "ZONEMAP"]
+        ],
+    ],
+)
+def test_scalar_index_types(tmp_path, index_type, data_type, values, filter_expr):
+    values = pa.array(values, type=data_type)
+    ds = lance.write_dataset(pa.table({"value": values}), tmp_path)
 
-    ds = lance.write_dataset(pa.table({"uuid": arr}), tmp_path)
+    ds.create_scalar_index("value", index_type)
 
-    ds.create_scalar_index("uuid", "BTREE")
+    scanner = ds.scanner(filter=filter_expr)
+    assert "ScalarIndexQuery" in scanner.explain_plan()
+    assert scanner.to_table()["value"].to_pylist() == values.slice(1, 1).to_pylist()
 
-    query = (
-        "uuid = arrow_cast(0x32333435323334353233343532333435, 'FixedSizeBinary(16)')"
-    )
-    assert (
-        "ScalarIndexQuery: query=[uuid = 32333435323334353233...]@uuid_idx"
-        in ds.scanner(filter=query).explain_plan()
+    fragment_id = ds.get_fragments()[0].fragment_id
+    segment = ds.create_index_uncommitted(
+        column="value",
+        index_type=index_type,
+        name=f"{index_type.lower()}_segment_idx",
+        fragment_ids=[fragment_id],
     )
-
-    table = ds.scanner(filter=query).to_table()
-    assert table.num_rows == 1
-    assert table.column("uuid").to_pylist() == arr.slice(1, 1).to_pylist()
+    assert segment.fragment_ids == {fragment_id}
 
 
 def test_index_take_batch_size(tmp_path):

rust/arrow-stats/src/lib.rs

@@ -445,6 +445,10 @@ fn find_min_max_indices(array: &ArrayRef) -> Result<(Option<usize>, Option<usize
         Float32 => find_extrema_float!(array, Float32Type),
         Float64 => find_extrema_float!(array, Float64Type),
 
+        // Decimal types
+        Decimal128(_, _) => find_extrema_primitive!(array, Decimal128Type),
+        Decimal256(_, _) => find_extrema_primitive!(array, Decimal256Type),
+
         // Temporal types
         Date32 => find_extrema_primitive!(array, Date32Type),
         Date64 => find_extrema_primitive!(array, Date64Type),
@@ -734,6 +738,26 @@ mod tests {
         Arc::new(Float64Array::from(vec![3.0f64, 1.0, 2.0])) as ArrayRef,
         "1.0", "3.0"
     )]
+    #[case::decimal128(
+        DataType::Decimal128(10, 2),
+        Arc::new(
+            Decimal128Array::from(vec![300_i128, 100, 200])
+                .with_precision_and_scale(10, 2)
+                .unwrap(),
+        ) as ArrayRef,
+        "1.00", "3.00"
+    )]
+    #[case::decimal256(
+        DataType::Decimal256(76, 2),
+        Arc::new(
+            Decimal256Array::from_iter_values(
+                [300_i64, 100, 200].into_iter().map(Into::into),
+            )
+            .with_precision_and_scale(76, 2)
+            .unwrap(),
+        ) as ArrayRef,
+        "1.00", "3.00"
+    )]
     fn test_rstest_primitives(
         #[case] dt: DataType,
         #[case] array: ArrayRef,

rust/lance-datafusion/src/expr.rs

@@ -448,6 +448,25 @@ pub fn safe_coerce_scalar(value: &ScalarValue, ty: &DataType) -> Option<ScalarVa
             DataType::BinaryView => Some(value.clone()),
             _ => None,
         },
+        ScalarValue::LargeBinary(_) => match ty {
+            DataType::LargeBinary => Some(value.clone()),
+            _ => None,
+        },
+        ScalarValue::Decimal128(_, _, _) => match ty {
+            DataType::Decimal128(_, _) => value.cast_to(ty).ok(),
+            _ => None,
+        },
+        ScalarValue::Decimal256(_, _, _) => match ty {
+            DataType::Decimal256(_, _) => value.cast_to(ty).ok(),
+            _ => None,
+        },
+        ScalarValue::DurationSecond(_)
+        | ScalarValue::DurationMillisecond(_)
+        | ScalarValue::DurationMicrosecond(_)
+        | ScalarValue::DurationNanosecond(_) => match ty {
+            DataType::Duration(_) => value.cast_to(ty).ok(),
+            _ => None,
+        },
         // A dictionary-encoded literal (e.g. produced by DataFusion's dictionary
         // cast in the scalar-index path) coerces by unwrapping its underlying value.
         ScalarValue::Dictionary(_, inner) => safe_coerce_scalar(inner, ty),
@@ -457,6 +476,8 @@ pub fn safe_coerce_scalar(value: &ScalarValue, ty: &DataType) -> Option<ScalarVa
 
 #[cfg(test)]
 mod tests {
+    use arrow::datatypes::i256;
+
     use super::*;
 
     #[test]
@@ -732,6 +753,13 @@ mod tests {
             ),
             Some(ScalarValue::Time64Nanosecond(Some(5000000000)))
         );
+        assert_eq!(
+            safe_coerce_scalar(
+                &ScalarValue::DurationNanosecond(Some(2_000_000)),
+                &DataType::Duration(TimeUnit::Millisecond),
+            ),
+            Some(ScalarValue::DurationMillisecond(Some(2)))
+        );
     }
 
     #[test]
@@ -789,6 +817,31 @@ mod tests {
             ),
             Some(ScalarValue::BinaryView(Some(vec![1, 2, 3])))
         );
+        assert_eq!(
+            safe_coerce_scalar(
+                &ScalarValue::LargeBinary(Some(vec![1, 2, 3])),
+                &DataType::LargeBinary
+            ),
+            Some(ScalarValue::LargeBinary(Some(vec![1, 2, 3])))
+        );
+    }
+
+    #[test]
+    fn test_decimal_coerce() {
+        assert_eq!(
+            safe_coerce_scalar(
+                &ScalarValue::Decimal128(Some(2), 10, 0),
+                &DataType::Decimal128(12, 2),
+            ),
+            Some(ScalarValue::Decimal128(Some(200), 12, 2))
+        );
+        assert_eq!(
+            safe_coerce_scalar(
+                &ScalarValue::Decimal256(Some(i256::from_i128(2)), 76, 0),
+                &DataType::Decimal256(76, 2),
+            ),
+            Some(ScalarValue::Decimal256(Some(i256::from_i128(200)), 76, 2))
+        );
     }
 
     #[test]