diff --git a/crates/integrations/datafusion/tests/global_index_schema_evolution.rs b/crates/integrations/datafusion/tests/global_index_schema_evolution.rs
new file mode 100644
index 000000000..947b7e0e0
--- /dev/null
+++ b/crates/integrations/datafusion/tests/global_index_schema_evolution.rs
@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A sorted global index is read with a comparator built from the column's *current*
+//! type, and nothing records the type the index was built with. Widening an indexed
+//! column is allowed, so the index must be left alone once its keys can no longer
+//! belong to that type.
+
+mod common;
+
+use common::{exec, row_count, setup_sql_context};
+
+async fn setup_indexed_table(
+    table_name: &str,
+    column_type: &str,
+) -> (tempfile::TempDir, paimon_datafusion::SQLContext) {
+    let (tmp, sql_context) = setup_sql_context().await;
+    exec(
+        &sql_context,
+        &format!(
+            "CREATE TABLE paimon.test_db.{table_name} (id {column_type}, name VARCHAR(100)) WITH (\
+                'row-tracking.enabled' = 'true',\
+                'data-evolution.enabled' = 'true',\
+                'global-index.enabled' = 'true',\
+                'sorted-index.records-per-range' = '10'\
+            )"
+        ),
+    )
+    .await;
+    (tmp, sql_context)
+}
+
+#[tokio::test]
+async fn test_widening_an_indexed_int_column_keeps_answering_queries() {
+    let (_tmp, sql_context) = setup_indexed_table("gi_widen_int", "INT").await;
+    for id in 1..=40 {
+        exec(
+            &sql_context,
+            &format!("INSERT INTO paimon.test_db.gi_widen_int (id, name) VALUES ({id}, 'n{id}')"),
+        )
+        .await;
+    }
+    exec(
+        &sql_context,
+        "CALL sys.create_global_index(table => 'test_db.gi_widen_int', index_column => 'id')",
+    )
+    .await;
+    assert_eq!(
+        row_count(
+            &sql_context,
+            "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 7"
+        )
+        .await,
+        1,
+        "the index must answer the query before the type change"
+    );
+
+    // Allowed: `UpdateColumnType` guards partition, primary-key, bucket-key and
+    // primary-key-index columns, but not global-index columns.
+    exec(
+        &sql_context,
+        "ALTER TABLE paimon.test_db.gi_widen_int ALTER COLUMN id TYPE BIGINT",
+    )
+    .await;
+
+    // The index keys are still 4 bytes wide; the BIGINT comparator used to read 8 and
+    // panic with "range end index 8 out of range for slice of length 4".
+    assert_eq!(
+        row_count(
+            &sql_context,
+            "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 7"
+        )
+        .await,
+        1,
+        "widening an indexed column must fall back to a scan, not panic"
+    );
+    assert_eq!(
+        row_count(
+            &sql_context,
+            "SELECT * FROM paimon.test_db.gi_widen_int WHERE id > 35"
+        )
+        .await,
+        5
+    );
+    assert_eq!(
+        row_count(
+            &sql_context,
+            "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 99"
+        )
+        .await,
+        0
+    );
+}
diff --git a/crates/paimon/src/btree/block.rs b/crates/paimon/src/btree/block.rs
index 950995f75..6f5c513b0 100644
--- a/crates/paimon/src/btree/block.rs
+++ b/crates/paimon/src/btree/block.rs
@@ -641,10 +641,11 @@ impl BlockReader {
 
     /// Binary search for the given target key. Returns an iterator positioned at the
     /// first entry whose key >= target_key.
-    /// The comparator compares two key byte slices.
-    pub fn seek_and_iter<F>(&self, target_key: &[u8], cmp: &F) -> (bool, BlockIter<'_>)
+    /// The comparator compares two key byte slices, and fails when the stored keys are
+    /// not keys of the column's current type.
+    pub fn seek_and_iter<F>(&self, target_key: &[u8], cmp: &F) -> io::Result<(bool, BlockIter<'_>)>
     where
-        F: Fn(&[u8], &[u8]) -> Ordering,
+        F: Fn(&[u8], &[u8]) -> io::Result<Ordering>,
     {
         let mut left: i32 = 0;
         let mut right: i32 = self.record_count as i32 - 1;
@@ -657,7 +658,7 @@ impl BlockReader {
             let byte_offset = self.seek_to_position(mid as usize);
             let (key, _next_offset) = self.read_key_at(byte_offset);
 
-            match cmp(key, target_key) {
+            match cmp(key, target_key)? {
                 Ordering::Equal => {
                     found = true;
                     best_index = Some(mid as usize);
@@ -675,7 +676,7 @@ impl BlockReader {
             }
         }
 
-        match (best_index, best_offset) {
+        Ok(match (best_index, best_offset) {
             (Some(idx), Some(off)) => (
                 found,
                 BlockIter {
@@ -692,7 +693,7 @@ impl BlockReader {
                     index: self.record_count,
                 },
             ),
-        }
+        })
     }
 }
 
@@ -847,27 +848,46 @@ mod tests {
         let block = writer.finish();
         let reader = BlockReader::create(&block).unwrap();
 
-        let cmp = |a: &[u8], b: &[u8]| a.cmp(b);
+        let cmp = |a: &[u8], b: &[u8]| Ok(a.cmp(b));
 
         // Exact match
-        let (found, mut iter) = reader.seek_and_iter(b"banana", &cmp);
+        let (found, mut iter) = reader.seek_and_iter(b"banana", &cmp).unwrap();
         assert!(found);
         let (k, v) = iter.next().unwrap();
         assert_eq!(k, b"banana");
         assert_eq!(v, b"2");
 
         // Seek to position >= "bz" -> should land on "cherry"
-        let (found, mut iter) = reader.seek_and_iter(b"bz", &cmp);
+        let (found, mut iter) = reader.seek_and_iter(b"bz", &cmp).unwrap();
         assert!(!found);
         let (k, _) = iter.next().unwrap();
         assert_eq!(k, b"cherry");
 
         // Seek past all entries
-        let (found, iter) = reader.seek_and_iter(b"zzz", &cmp);
+        let (found, iter) = reader.seek_and_iter(b"zzz", &cmp).unwrap();
         assert!(!found);
         assert!(!iter.has_next());
     }
 
+    /// A comparator that rejects the stored keys must abort the search, not settle on
+    /// whatever entry the half-finished binary search last looked at.
+    #[test]
+    fn test_block_seek_propagates_a_comparator_failure() {
+        let mut writer = BlockWriter::new(1024);
+        writer.add(b"apple", b"1");
+        writer.add(b"banana", b"2");
+
+        let block = writer.finish();
+        let reader = BlockReader::create(&block).unwrap();
+
+        let cmp =
+            |_: &[u8], _: &[u8]| Err(io::Error::new(io::ErrorKind::InvalidData, "not my key"));
+        let Err(error) = reader.seek_and_iter(b"banana", &cmp) else {
+            panic!("a failing comparator must abort the seek");
+        };
+        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
+    }
+
     #[test]
     fn test_block_handle_roundtrip() {
         let handle = BlockHandle::new(12345, 6789);
diff --git a/crates/paimon/src/btree/key_serde.rs b/crates/paimon/src/btree/key_serde.rs
index d1b7834fb..3ef651dd7 100644
--- a/crates/paimon/src/btree/key_serde.rs
+++ b/crates/paimon/src/btree/key_serde.rs
@@ -19,7 +19,7 @@
 //!
 //! Reference: [org.apache.paimon.globalindex.btree.KeySerializer](https://github.com/apache/paimon/blob/master/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java)
 
-use crate::btree::var_len::{decode_var_int_from_slice, encode_var_int};
+use crate::btree::var_len::{encode_var_int, try_decode_var_int_from_slice};
 use crate::spec::{DataType, Datum, VariantType};
 use std::cmp::Ordering;
 
@@ -29,85 +29,174 @@ const TIMESTAMP_COMPACT_PRECISION: u32 = 3;
 const DECIMAL_COMPACT_PRECISION: u32 = 18;
 
 /// Key comparator type alias.
-pub type KeyComparator = Box<dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync>;
+///
+/// Fallible because the bytes come from an index file that may have been written when
+/// the column had a different type. [`make_key_comparator`] picks its arm from the
+/// column's *current* type, and nothing records the type the index was built with:
+/// neither `IndexManifestEntry` nor `IndexFileMeta` carries a schema id.
+/// `SchemaChange::UpdateColumnType` guards partition, primary-key, bucket-key and
+/// primary-key-index columns but not global-index columns, so `INT` to `BIGINT` is
+/// accepted and every stored key is then 4 bytes short of what the new arm reads. That
+/// used to index out of bounds and panic inside a query; an `Err` lets the caller give
+/// up on the index and fall back instead.
+pub type KeyComparator = Box<dyn Fn(&[u8], &[u8]) -> crate::Result<Ordering> + Send + Sync>;
+
+/// A borrowed key comparator, for signatures that only call one. [`KeyComparator`] is
+/// the owned form callers build from a [`DataType`].
+pub type DynKeyComparator<'a> = dyn Fn(&[u8], &[u8]) -> crate::Result<Ordering> + 'a;
+
+/// Take the fixed-width body of a key, or explain why it cannot belong to `type_name`.
+///
+/// The width must match exactly: [`serialize_datum`] emits exactly `N` bytes for these
+/// types, so a longer key is as much a foreign encoding as a shorter one, and silently
+/// comparing its prefix would answer the query from the wrong bytes.
+pub(crate) fn fixed_key_bytes<const N: usize>(
+    key: &[u8],
+    type_name: &str,
+) -> crate::Result<[u8; N]> {
+    key.try_into().map_err(|_| crate::Error::DataInvalid {
+        message: format!(
+            "Global index key of {} byte(s) cannot be a {type_name} key of {N}; the index was \
+             built before the column's type changed and cannot be used",
+            key.len()
+        ),
+        source: None,
+    })
+}
+
+fn variable_key_body(key: &[u8], min_len: usize, type_name: &str) -> crate::Result<()> {
+    if key.len() < min_len {
+        return Err(crate::Error::DataInvalid {
+            message: format!(
+                "Global index key of {} byte(s) cannot be a {type_name} key of at least \
+                 {min_len}; the index was built before the column's type changed and cannot be \
+                 used",
+                key.len()
+            ),
+            source: None,
+        });
+    }
+    Ok(())
+}
+
+/// A [`KeyComparator`] failure travelling through a reader's [`std::io::Result`].
+///
+/// The BTree and bitmap readers interleave key comparison with file I/O and report both
+/// as [`std::io::Error`], but their caller has to treat the two differently: a comparison
+/// failure means the stored bytes are not keys of this type, so the index simply cannot
+/// answer and the query falls back to a scan, while a real I/O failure must fail the
+/// query. Wrapping keeps both in one channel and lets
+/// [`is_key_comparison_failure`] tell them apart.
+#[derive(Debug)]
+pub struct KeyComparisonFailure {
+    message: String,
+}
+
+impl std::fmt::Display for KeyComparisonFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for KeyComparisonFailure {}
+
+/// Wrap a comparator failure for a reader that returns [`std::io::Result`].
+pub(crate) fn key_comparison_io_error(error: crate::Error) -> std::io::Error {
+    std::io::Error::new(
+        std::io::ErrorKind::InvalidData,
+        KeyComparisonFailure {
+            message: error.to_string(),
+        },
+    )
+}
+
+/// Whether `error` is a key comparison failure rather than a real I/O failure.
+pub(crate) fn is_key_comparison_failure(error: &std::io::Error) -> bool {
+    error
+        .get_ref()
+        .is_some_and(|source| source.is::<KeyComparisonFailure>())
+}
 
 /// Create a key comparator based on the data type.
 /// For fixed-size numeric types, compares by decoded value.
 /// For variable-length types (string, bytes), uses lexicographic byte comparison.
 pub fn make_key_comparator(data_type: &DataType) -> KeyComparator {
     match data_type {
-        DataType::TinyInt(_) => Box::new(|a: &[u8], b: &[u8]| (a[0] as i8).cmp(&(b[0] as i8))),
+        DataType::TinyInt(_) => Box::new(|a: &[u8], b: &[u8]| {
+            let av = i8::from_le_bytes(fixed_key_bytes::<1>(a, "TINYINT")?);
+            let bv = i8::from_le_bytes(fixed_key_bytes::<1>(b, "TINYINT")?);
+            Ok(av.cmp(&bv))
+        }),
         DataType::SmallInt(_) => Box::new(|a: &[u8], b: &[u8]| {
-            let av = i16::from_le_bytes(a[..2].try_into().unwrap());
-            let bv = i16::from_le_bytes(b[..2].try_into().unwrap());
-            av.cmp(&bv)
+            let av = i16::from_le_bytes(fixed_key_bytes::<2>(a, "SMALLINT")?);
+            let bv = i16::from_le_bytes(fixed_key_bytes::<2>(b, "SMALLINT")?);
+            Ok(av.cmp(&bv))
         }),
         DataType::Int(_) | DataType::Date(_) | DataType::Time(_) => {
             Box::new(|a: &[u8], b: &[u8]| {
-                let av = i32::from_le_bytes(a[..4].try_into().unwrap());
-                let bv = i32::from_le_bytes(b[..4].try_into().unwrap());
-                av.cmp(&bv)
+                let av = i32::from_le_bytes(fixed_key_bytes::<4>(a, "INT")?);
+                let bv = i32::from_le_bytes(fixed_key_bytes::<4>(b, "INT")?);
+                Ok(av.cmp(&bv))
             })
         }
         DataType::BigInt(_) => Box::new(|a: &[u8], b: &[u8]| {
-            let av = i64::from_le_bytes(a[..8].try_into().unwrap());
-            let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
-            av.cmp(&bv)
+            let av = i64::from_le_bytes(fixed_key_bytes::<8>(a, "BIGINT")?);
+            let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b, "BIGINT")?);
+            Ok(av.cmp(&bv))
         }),
         DataType::Float(_) => Box::new(|a: &[u8], b: &[u8]| {
-            let av = f32::from_le_bytes(a[..4].try_into().unwrap());
-            let bv = f32::from_le_bytes(b[..4].try_into().unwrap());
-            av.total_cmp(&bv)
+            let av = f32::from_le_bytes(fixed_key_bytes::<4>(a, "FLOAT")?);
+            let bv = f32::from_le_bytes(fixed_key_bytes::<4>(b, "FLOAT")?);
+            Ok(av.total_cmp(&bv))
         }),
         DataType::Double(_) => Box::new(|a: &[u8], b: &[u8]| {
-            let av = f64::from_le_bytes(a[..8].try_into().unwrap());
-            let bv = f64::from_le_bytes(b[..8].try_into().unwrap());
-            av.total_cmp(&bv)
+            let av = f64::from_le_bytes(fixed_key_bytes::<8>(a, "DOUBLE")?);
+            let bv = f64::from_le_bytes(fixed_key_bytes::<8>(b, "DOUBLE")?);
+            Ok(av.total_cmp(&bv))
         }),
         DataType::Timestamp(t) if t.precision() > TIMESTAMP_COMPACT_PRECISION => {
             // Non-compact: millis (8 bytes LE) + nanoOfMillisecond (varint)
-            Box::new(|a: &[u8], b: &[u8]| {
-                let a_millis = i64::from_le_bytes(a[..8].try_into().unwrap());
-                let b_millis = i64::from_le_bytes(b[..8].try_into().unwrap());
-                let (a_nanos, _) = decode_var_int_from_slice(a, 8);
-                let (b_nanos, _) = decode_var_int_from_slice(b, 8);
-                a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos))
-            })
+            Box::new(|a: &[u8], b: &[u8]| compare_non_compact_timestamps(a, b))
         }
         DataType::LocalZonedTimestamp(t) if t.precision() > TIMESTAMP_COMPACT_PRECISION => {
-            Box::new(|a: &[u8], b: &[u8]| {
-                let a_millis = i64::from_le_bytes(a[..8].try_into().unwrap());
-                let b_millis = i64::from_le_bytes(b[..8].try_into().unwrap());
-                let (a_nanos, _) = decode_var_int_from_slice(a, 8);
-                let (b_nanos, _) = decode_var_int_from_slice(b, 8);
-                a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos))
-            })
+            Box::new(|a: &[u8], b: &[u8]| compare_non_compact_timestamps(a, b))
         }
         DataType::Decimal(d) if d.precision() > DECIMAL_COMPACT_PRECISION => {
             // Non-compact Decimal keys use Java BigInteger.toByteArray() bytes.
             Box::new(|a: &[u8], b: &[u8]| {
-                decode_java_big_integer_i128(a).cmp(&decode_java_big_integer_i128(b))
+                Ok(decode_java_big_integer_i128(a)?.cmp(&decode_java_big_integer_i128(b)?))
             })
         }
         // Compact Timestamp/LocalZonedTimestamp (precision <= 3): millis as i64 LE
         DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_) => {
             Box::new(|a: &[u8], b: &[u8]| {
-                let av = i64::from_le_bytes(a[..8].try_into().unwrap());
-                let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
-                av.cmp(&bv)
+                let av = i64::from_le_bytes(fixed_key_bytes::<8>(a, "TIMESTAMP")?);
+                let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b, "TIMESTAMP")?);
+                Ok(av.cmp(&bv))
             })
         }
         // Compact Decimal (precision <= 18): unscaled as i64 LE
         DataType::Decimal(_) => Box::new(|a: &[u8], b: &[u8]| {
-            let av = i64::from_le_bytes(a[..8].try_into().unwrap());
-            let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
-            av.cmp(&bv)
+            let av = i64::from_le_bytes(fixed_key_bytes::<8>(a, "DECIMAL")?);
+            let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b, "DECIMAL")?);
+            Ok(av.cmp(&bv))
         }),
-        // String, VarChar, Char, Bytes — lexicographic
-        _ => Box::new(|a: &[u8], b: &[u8]| a.cmp(b)),
+        // String, VarChar, Char, Bytes — lexicographic, so any width is readable
+        _ => Box::new(|a: &[u8], b: &[u8]| Ok(a.cmp(b))),
     }
 }
 
+/// Millis (8 bytes LE) then the varint nano-of-millisecond, both bounds-checked.
+fn compare_non_compact_timestamps(a: &[u8], b: &[u8]) -> crate::Result<Ordering> {
+    variable_key_body(a, 9, "TIMESTAMP")?;
+    variable_key_body(b, 9, "TIMESTAMP")?;
+    let a_millis = i64::from_le_bytes(fixed_key_bytes::<8>(&a[..8], "TIMESTAMP")?);
+    let b_millis = i64::from_le_bytes(fixed_key_bytes::<8>(&b[..8], "TIMESTAMP")?);
+    let (a_nanos, _) = try_decode_var_int_from_slice(a, 8)?;
+    let (b_nanos, _) = try_decode_var_int_from_slice(b, 8)?;
+    Ok(a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos)))
+}
+
 /// Serialize a Datum to BTree key bytes (little-endian, matching Java Paimon's KeySerializer).
 pub fn serialize_datum(datum: &Datum, data_type: &DataType) -> Vec<u8> {
     match datum {
@@ -174,16 +263,35 @@ fn encode_java_big_integer_i128(value: i128) -> Vec<u8> {
     bytes[start..].to_vec()
 }
 
-fn decode_java_big_integer_i128(bytes: &[u8]) -> i128 {
-    if bytes.is_empty() {
-        return 0;
+/// Read the Java `BigInteger.toByteArray()` bytes of a non-compact DECIMAL key.
+///
+/// Anything outside 1..=16 bytes cannot be one: [`encode_java_big_integer_i128`] never
+/// emits an empty slice, and more than 16 bytes does not fit the `i128` unscaled value.
+/// Such a key belongs to the type the column had when the index was built.
+///
+/// The bound is all this arm can check, and 1..=16 covers every fixed-width key this
+/// module writes, so a stale `INT` or `BIGINT` key is accepted here and read as a
+/// big-endian magnitude -- `numeric` to `DECIMAL(p > 18)` is an implicit cast, so that
+/// is reachable. It gives a wrong ordering rather than a panic, which puts it in the
+/// same undetectable class as `INT` to `FLOAT`: telling a foreign key from a real one
+/// needs the type the index was built with, and no index file records it.
+fn decode_java_big_integer_i128(bytes: &[u8]) -> crate::Result<i128> {
+    if bytes.is_empty() || bytes.len() > 16 {
+        return Err(crate::Error::DataInvalid {
+            message: format!(
+                "Global index key of {} byte(s) cannot be a non-compact DECIMAL key of 1 to 16; \
+                 the index was built before the column's type changed and cannot be used",
+                bytes.len()
+            ),
+            source: None,
+        });
     }
     let negative = bytes[0] & 0x80 != 0;
     let mut value = if negative { -1 } else { 0 };
     for &byte in bytes {
         value = (value << 8) | i128::from(byte);
     }
-    value
+    Ok(value)
 }
 
 #[cfg(test)]
@@ -263,8 +371,8 @@ mod tests {
         let key_128 = encode_java_big_integer_i128(128);
         let key_minus_129 = encode_java_big_integer_i128(-129);
 
-        assert_eq!(cmp(&key_127, &key_128), Ordering::Less);
-        assert_eq!(cmp(&key_minus_129, &key_127), Ordering::Less);
+        assert_eq!(cmp(&key_127, &key_128).unwrap(), Ordering::Less);
+        assert_eq!(cmp(&key_minus_129, &key_127).unwrap(), Ordering::Less);
     }
 
     #[test]
@@ -272,4 +380,39 @@ mod tests {
         let key = serialize_datum(&Datum::Int(42), &DataType::Int(IntType::new()));
         assert_eq!(key, 42i32.to_le_bytes());
     }
+
+    /// Every arm that reads a fixed number of bytes must report a key of the wrong
+    /// width instead of indexing out of bounds. Each of these used to panic inside a
+    /// query after `ALTER COLUMN ... TYPE` widened an indexed column.
+    #[test]
+    fn test_comparator_rejects_keys_of_another_type_instead_of_panicking() {
+        use crate::spec::{BigIntType, TimestampType, VarCharType};
+
+        // An index built on INT, read after the column became BIGINT.
+        let big_int = make_key_comparator(&DataType::BigInt(BigIntType::new()));
+        assert!(big_int(&[0; 4], &[0; 8]).is_err());
+        assert!(big_int(&[0; 8], &[0; 4]).is_err());
+        assert_eq!(big_int(&[0; 8], &[0; 8]).unwrap(), Ordering::Equal);
+
+        // An INT arm handed the 8-byte keys of a column that used to be BIGINT.
+        let int = make_key_comparator(&DataType::Int(IntType::new()));
+        assert!(int(&[0; 8], &[0; 4]).is_err());
+
+        // An index built on TIMESTAMP(3), read after the column became TIMESTAMP(6):
+        // the non-compact arm decodes a varint at offset 8.
+        let non_compact = make_key_comparator(&DataType::Timestamp(TimestampType::new(6).unwrap()));
+        assert!(non_compact(&[0; 8], &[0; 9]).is_err());
+        assert_eq!(non_compact(&[0; 9], &[0; 9]).unwrap(), Ordering::Equal);
+        let compact = make_key_comparator(&DataType::Timestamp(TimestampType::new(3).unwrap()));
+        assert!(compact(&[0; 9], &[0; 8]).is_err());
+
+        // Non-compact DECIMAL keys are BigInteger bytes: never empty, never over 16.
+        let decimal = make_key_comparator(&DataType::Decimal(DecimalType::new(20, 0).unwrap()));
+        assert!(decimal(&[], &[0x01]).is_err());
+        assert!(decimal(&[0; 17], &[0x01]).is_err());
+
+        // Character and byte keys are compared lexicographically, so no width is wrong.
+        let varchar = make_key_comparator(&DataType::VarChar(VarCharType::new(10).unwrap()));
+        assert_eq!(varchar(&[], &[0; 3]).unwrap(), Ordering::Less);
+    }
 }
diff --git a/crates/paimon/src/btree/meta.rs b/crates/paimon/src/btree/meta.rs
index c8c0bbe83..e23892f57 100644
--- a/crates/paimon/src/btree/meta.rs
+++ b/crates/paimon/src/btree/meta.rs
@@ -24,6 +24,7 @@
 //! ```
 //! Null key flags distinguish empty serialized keys from absent keys.
 
+use crate::btree::key_serde::DynKeyComparator;
 use crate::spec::PredicateOperator;
 use std::cmp::Ordering;
 use std::io;
@@ -76,66 +77,102 @@ impl BTreeIndexMeta {
     }
 
     /// File-level pruning: check if this BTree file may contain matching keys.
+    ///
+    /// Degradation: pruning predicate. A comparator failure means the stored keys are
+    /// not keys of the column's current type, so this file's recorded bounds order
+    /// nothing and prove nothing -- answer "may match" and leave the decision to the
+    /// read. Turning the failure into "cannot match" would silently drop rows.
     pub fn may_match(
         &self,
         op: PredicateOperator,
         serialized_literals: &[Vec<u8>],
-        cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+        cmp: &DynKeyComparator<'_>,
     ) -> bool {
-        match op {
+        self.try_may_match(op, serialized_literals, cmp)
+            .unwrap_or(true)
+    }
+
+    fn try_may_match(
+        &self,
+        op: PredicateOperator,
+        serialized_literals: &[Vec<u8>],
+        cmp: &DynKeyComparator<'_>,
+    ) -> crate::Result<bool> {
+        Ok(match op {
             PredicateOperator::IsNull => self.has_nulls,
             PredicateOperator::IsNotNull => !self.only_nulls(),
             PredicateOperator::NotEq | PredicateOperator::NotIn => true,
             _ => {
                 if self.only_nulls() {
-                    return false;
+                    return Ok(false);
                 }
                 let (first_key, last_key) = match (&self.first_key, &self.last_key) {
                     (Some(f), Some(l)) => (f.as_slice(), l.as_slice()),
-                    _ => return true,
+                    _ => return Ok(true),
                 };
                 match op {
                     PredicateOperator::Eq => {
-                        cmp(&serialized_literals[0], first_key) != Ordering::Less
-                            && cmp(&serialized_literals[0], last_key) != Ordering::Greater
+                        cmp(&serialized_literals[0], first_key)? != Ordering::Less
+                            && cmp(&serialized_literals[0], last_key)? != Ordering::Greater
                     }
                     PredicateOperator::Lt => {
-                        cmp(first_key, &serialized_literals[0]) == Ordering::Less
+                        cmp(first_key, &serialized_literals[0])? == Ordering::Less
                     }
                     PredicateOperator::LtEq => {
-                        cmp(first_key, &serialized_literals[0]) != Ordering::Greater
+                        cmp(first_key, &serialized_literals[0])? != Ordering::Greater
                     }
                     PredicateOperator::Gt => {
-                        cmp(last_key, &serialized_literals[0]) == Ordering::Greater
+                        cmp(last_key, &serialized_literals[0])? == Ordering::Greater
                     }
                     PredicateOperator::GtEq => {
-                        cmp(last_key, &serialized_literals[0]) != Ordering::Less
+                        cmp(last_key, &serialized_literals[0])? != Ordering::Less
+                    }
+                    PredicateOperator::In => {
+                        let mut any = false;
+                        for key in serialized_literals {
+                            if cmp(key, first_key)? != Ordering::Less
+                                && cmp(key, last_key)? != Ordering::Greater
+                            {
+                                any = true;
+                                break;
+                            }
+                        }
+                        any
                     }
-                    PredicateOperator::In => serialized_literals.iter().any(|key| {
-                        cmp(key, first_key) != Ordering::Less
-                            && cmp(key, last_key) != Ordering::Greater
-                    }),
                     _ => true,
                 }
             }
-        }
+        })
     }
 
     /// File-level pruning for between: file may match if [first_key, last_key] overlaps [from, to].
+    ///
+    /// Degradation: pruning predicate, for the same reason as [`Self::may_match`].
     pub fn may_match_between(
         &self,
         from_key: &[u8],
         to_key: &[u8],
-        cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+        cmp: &DynKeyComparator<'_>,
     ) -> bool {
+        self.try_may_match_between(from_key, to_key, cmp)
+            .unwrap_or(true)
+    }
+
+    fn try_may_match_between(
+        &self,
+        from_key: &[u8],
+        to_key: &[u8],
+        cmp: &DynKeyComparator<'_>,
+    ) -> crate::Result<bool> {
         if self.only_nulls() {
-            return false;
+            return Ok(false);
         }
         let (first_key, last_key) = match (&self.first_key, &self.last_key) {
             (Some(f), Some(l)) => (f.as_slice(), l.as_slice()),
-            _ => return true,
+            _ => return Ok(true),
         };
-        cmp(first_key, to_key) != Ordering::Greater && cmp(last_key, from_key) != Ordering::Less
+        Ok(cmp(first_key, to_key)? != Ordering::Greater
+            && cmp(last_key, from_key)? != Ordering::Less)
     }
 
     /// Serialize to bytes (compatible with Java SortedIndexFileMeta.serialize()).
@@ -306,4 +343,33 @@ mod tests {
         let error = BTreeIndexMeta::deserialize(&encoded).unwrap_err();
         assert_eq!(error.kind(), io::ErrorKind::InvalidData);
     }
+
+    /// Pruning must stay conservative when the recorded keys cannot be compared: the
+    /// index was built before the column's type changed, so its bounds prove nothing.
+    /// Answering "cannot match" here would silently drop rows.
+    #[test]
+    fn test_pruning_says_may_match_when_the_comparator_rejects_the_stored_keys() {
+        let meta = BTreeIndexMeta::new(Some(vec![0; 4]), Some(vec![9; 4]), false);
+        let failing = |_: &[u8], _: &[u8]| {
+            Err(crate::Error::DataInvalid {
+                message: "stored keys are not keys of this type".to_string(),
+                source: None,
+            })
+        };
+
+        for op in [
+            PredicateOperator::Eq,
+            PredicateOperator::Lt,
+            PredicateOperator::LtEq,
+            PredicateOperator::Gt,
+            PredicateOperator::GtEq,
+            PredicateOperator::In,
+        ] {
+            assert!(
+                meta.may_match(op, &[vec![0; 8]], &failing),
+                "{op} must not prune the file"
+            );
+        }
+        assert!(meta.may_match_between(&[0; 8], &[9; 8], &failing));
+    }
 }
diff --git a/crates/paimon/src/btree/query.rs b/crates/paimon/src/btree/query.rs
index cb7688c0a..b5166b5af 100644
--- a/crates/paimon/src/btree/query.rs
+++ b/crates/paimon/src/btree/query.rs
@@ -42,7 +42,7 @@ pub trait IndexQuery: Send + Sync {
 #[async_trait::async_trait]
 impl<F> IndexQuery for BTreeIndexReader<F>
 where
-    F: Fn(&[u8], &[u8]) -> Ordering + Send + Sync,
+    F: Fn(&[u8], &[u8]) -> crate::Result<Ordering> + Send + Sync,
 {
     async fn query(
         &self,
@@ -196,9 +196,14 @@ impl BetweenInfo<'_> {
             &serialize_datum(self.from, self.data_type),
             &serialize_datum(self.to, self.data_type),
         ) {
-            Ordering::Greater => true,
-            Ordering::Equal => !self.from_inclusive || !self.to_inclusive,
-            Ordering::Less => false,
+            Ok(Ordering::Greater) => true,
+            Ok(Ordering::Equal) => !self.from_inclusive || !self.to_inclusive,
+            Ok(Ordering::Less) => false,
+            // Degradation: pure optimisation. This only shortcuts a provably empty
+            // range to "no rows"; without an ordering we cannot prove that, so say
+            // "not empty" and let the index read decide. Answering "empty" on an
+            // error would drop rows.
+            Err(_) => false,
         }
     }
 }
@@ -218,11 +223,20 @@ pub(crate) type ExtractBetweenResult<'a> = (
 pub(crate) fn extract_between<'a>(
     predicates: &[(PredicateOperator, &'a [Datum], &'a DataType)],
 ) -> ExtractBetweenResult<'a> {
+    // Degradation: pure optimisation. Merging bounds needs an ordering on the
+    // serialized literals; without one, hand every predicate back so each is
+    // evaluated on its own, exactly as when no complete range was found.
+    try_extract_between(predicates).unwrap_or_else(|_| (None, predicates.to_vec()))
+}
+
+fn try_extract_between<'a>(
+    predicates: &[(PredicateOperator, &'a [Datum], &'a DataType)],
+) -> crate::Result<ExtractBetweenResult<'a>> {
     let Some((_, _, data_type)) = predicates.first() else {
-        return (None, Vec::new());
+        return Ok((None, Vec::new()));
     };
     if predicates.len() == 1 && predicates[0].0 != PredicateOperator::Between {
-        return (None, predicates.to_vec());
+        return Ok((None, predicates.to_vec()));
     }
     let cmp = crate::btree::make_key_comparator(data_type);
     let mut lower: Option<(&Datum, Vec<u8>, bool)> = None;
@@ -251,7 +265,7 @@ pub(crate) fn extract_between<'a>(
             if let Some((value, inclusive)) = candidate {
                 let key = serialize_datum(value, data_type);
                 match bound {
-                    Some((_, existing, current_inclusive)) => match cmp(&key, existing) {
+                    Some((_, existing, current_inclusive)) => match cmp(&key, existing)? {
                         Ordering::Equal => *current_inclusive &= inclusive,
                         order if order == tighter => *bound = Some((value, key, inclusive)),
                         _ => {}
@@ -261,7 +275,7 @@ pub(crate) fn extract_between<'a>(
             }
         }
     }
-    match (lower, upper) {
+    Ok(match (lower, upper) {
         (Some((from, _, from_inclusive)), Some((to, _, to_inclusive))) => (
             Some(BetweenInfo {
                 from,
@@ -273,5 +287,5 @@ pub(crate) fn extract_between<'a>(
             remaining,
         ),
         _ => (None, predicates.to_vec()),
-    }
+    })
 }
diff --git a/crates/paimon/src/btree/reader.rs b/crates/paimon/src/btree/reader.rs
index 1debf2396..4e2f50e44 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -26,6 +26,7 @@
 use crate::btree::block::{BlockHandle, BlockReader};
 use crate::btree::bloom_filter::BloomFilter;
 use crate::btree::footer::{BTreeFileFooter, BloomFilterHandle, BTREE_FOOTER_ENCODED_LENGTH};
+use crate::btree::key_serde::key_comparison_io_error;
 use crate::btree::meta::BTreeIndexMeta;
 use crate::btree::posting_list;
 use crate::btree::sst_file::{read_block_from_bytes, SstFileReader};
@@ -41,8 +42,28 @@ struct LazyBloomFilter {
     filter: OnceCell<BloomFilter>,
 }
 
+/// `slice::sort_by` with a fallible comparator. The first failure is reported and the
+/// resulting order is unspecified, so the caller must not use the slice afterwards.
+fn try_sort_by<T>(
+    values: &mut [T],
+    mut cmp: impl FnMut(&T, &T) -> io::Result<Ordering>,
+) -> io::Result<()> {
+    let mut failure = None;
+    values.sort_by(|left, right| match cmp(left, right) {
+        Ok(order) => order,
+        Err(error) => {
+            failure.get_or_insert(error);
+            Ordering::Equal
+        }
+    });
+    match failure {
+        Some(error) => Err(error),
+        None => Ok(()),
+    }
+}
+
 /// BTree index reader with on-demand async data block loading.
-pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
     reader: Box<dyn FileRead>,
     sst_reader: SstFileReader,
     null_bitmap: RoaringTreemap,
@@ -53,7 +74,7 @@ pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
     file_version: u32,
 }
 
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BTreeIndexReader<F> {
     /// Open a BTree index reader from a FileRead and file metadata.
     /// Only reads footer, index block, and null bitmap on open.
     /// Data blocks are read on demand during queries.
@@ -115,6 +136,13 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         &self.null_bitmap
     }
 
+    /// Compare two keys, reporting a stored key that cannot belong to the column's
+    /// current type as a distinguishable I/O error so the caller can give up on the
+    /// index instead of failing the query.
+    fn compare(&self, left: &[u8], right: &[u8]) -> io::Result<Ordering> {
+        (self.key_comparator)(left, right).map_err(key_comparison_io_error)
+    }
+
     /// Collect all non-null row ids into a bitmap.
     pub async fn all_non_null_rows(&self) -> io::Result<RoaringTreemap> {
         if self.min_key.is_none() {
@@ -138,9 +166,9 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
             return Ok(RoaringTreemap::new());
         };
 
-        let cmp = &self.key_comparator;
+        let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
         let index_block = self.sst_reader.index_block();
-        let (_, mut index_iter) = index_block.seek_and_iter(min_key, cmp);
+        let (_, mut index_iter) = index_block.seek_and_iter(min_key, &cmp)?;
         let mut result = RoaringTreemap::new();
 
         while let Some((_key, handle_bytes)) = index_iter.next() {
@@ -168,10 +196,10 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         from_inclusive: bool,
         to_inclusive: bool,
     ) -> io::Result<RoaringTreemap> {
-        let cmp = &self.key_comparator;
+        let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
         let mut result = RoaringTreemap::new();
 
-        match cmp(from, to) {
+        match self.compare(from, to)? {
             Ordering::Greater => return Ok(result),
             Ordering::Equal if !from_inclusive || !to_inclusive => return Ok(result),
             _ => {}
@@ -179,7 +207,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
 
         // Seek in index block to find the first data block that may contain `from`
         let index_block = self.sst_reader.index_block();
-        let (_, mut index_iter) = index_block.seek_and_iter(from, cmp);
+        let (_, mut index_iter) = index_block.seek_and_iter(from, &cmp)?;
 
         // First data block: seek within it
         let first_block = match index_iter.next() {
@@ -190,7 +218,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
             None => return Ok(result),
         };
 
-        let (_, seeked) = first_block.seek_and_iter(from, cmp);
+        let (_, seeked) = first_block.seek_and_iter(from, &cmp)?;
         let mut offset = seeked.offset;
 
         // Iterate first block from seeked position
@@ -240,16 +268,15 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         to_inclusive: bool,
         result: &mut RoaringTreemap,
     ) -> io::Result<bool> {
-        let cmp = &self.key_comparator;
         while *offset < block.data.len() {
             let (key, value, next_offset) = block.read_entry_at(*offset);
             *offset = next_offset;
 
-            if !from_inclusive && cmp(key, from) == Ordering::Equal {
+            if !from_inclusive && self.compare(key, from)? == Ordering::Equal {
                 continue;
             }
 
-            let diff = cmp(key, to);
+            let diff = self.compare(key, to)?;
             if diff == Ordering::Greater || (!to_inclusive && diff == Ordering::Equal) {
                 return Ok(true);
             }
@@ -316,16 +343,16 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
 
     /// Equal query: returns row ids for the given key.
     pub async fn query_equal(&self, key: &[u8]) -> io::Result<RoaringTreemap> {
-        let cmp = &self.key_comparator;
-        if self
-            .min_key
-            .as_deref()
-            .is_none_or(|min| cmp(key, min) == Ordering::Less)
-            || self
-                .max_key
-                .as_deref()
-                .is_none_or(|max| cmp(key, max) == Ordering::Greater)
-        {
+        let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
+        let outside_bounds = match self.min_key.as_deref() {
+            None => true,
+            Some(min) if self.compare(key, min)? == Ordering::Less => true,
+            Some(_) => match self.max_key.as_deref() {
+                None => true,
+                Some(max) => self.compare(key, max)? == Ordering::Greater,
+            },
+        };
+        if outside_bounds {
             return Ok(RoaringTreemap::new());
         }
         if !self.bloom_might_contain(key).await? {
@@ -333,13 +360,13 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         }
 
         let index_block = self.sst_reader.index_block();
-        let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
+        let (_, mut index_iter) = index_block.seek_and_iter(key, &cmp)?;
         let Some((_last_key, handle_bytes)) = index_iter.next() else {
             return Ok(RoaringTreemap::new());
         };
         let handle = BlockHandle::decode(handle_bytes)?;
         let block = self.read_data_block(&handle).await?;
-        let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
+        let (found, mut entry_iter) = block.seek_and_iter(key, &cmp)?;
         let mut result = RoaringTreemap::new();
         if let (true, Some((_entry_key, value))) = (found, entry_iter.next()) {
             posting_list::add_to(value, self.file_version, &mut result)?;
@@ -411,25 +438,39 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
             return Ok(RoaringTreemap::new());
         }
 
-        let cmp = &self.key_comparator;
+        let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
         let (Some(min_key), Some(max_key)) = (self.min_key.as_deref(), self.max_key.as_deref())
         else {
             return Ok(RoaringTreemap::new());
         };
 
-        // Sort, deduplicate, and discard keys outside this file's bounds before resolving blocks.
-        let mut sorted_keys: Vec<&[u8]> = keys.to_vec();
-        sorted_keys.sort_by(|a, b| cmp(a, b));
-        sorted_keys.dedup_by(|a, b| cmp(a, b) == Ordering::Equal);
-        sorted_keys.retain(|key| {
-            cmp(key, min_key) != Ordering::Less && cmp(key, max_key) != Ordering::Greater
-        });
-        if sorted_keys.is_empty() {
+        // Discard keys outside this file's bounds, then sort and deduplicate the rest
+        // before resolving blocks.
+        let mut sorted_keys: Vec<&[u8]> = Vec::with_capacity(keys.len());
+        for key in keys {
+            if self.compare(key, min_key)? != Ordering::Less
+                && self.compare(key, max_key)? != Ordering::Greater
+            {
+                sorted_keys.push(key);
+            }
+        }
+        try_sort_by(&mut sorted_keys, |left, right| self.compare(left, right))?;
+        let mut unique_keys: Vec<&[u8]> = Vec::with_capacity(sorted_keys.len());
+        for key in sorted_keys {
+            let duplicate = match unique_keys.last() {
+                Some(previous) => self.compare(previous, key)? == Ordering::Equal,
+                None => false,
+            };
+            if !duplicate {
+                unique_keys.push(key);
+            }
+        }
+        if unique_keys.is_empty() {
             return Ok(RoaringTreemap::new());
         }
 
-        let mut bloom_matches = Vec::with_capacity(sorted_keys.len());
-        for key in sorted_keys {
+        let mut bloom_matches = Vec::with_capacity(unique_keys.len());
+        for key in unique_keys {
             if self.bloom_might_contain(key).await? {
                 bloom_matches.push(key);
             }
@@ -443,7 +484,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         let index_block = self.sst_reader.index_block();
         let mut target_blocks: Vec<(BlockHandle, Vec<&[u8]>)> = Vec::new();
         for key in bloom_matches {
-            let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
+            let (_, mut index_iter) = index_block.seek_and_iter(key, &cmp)?;
             let Some((_last_key, handle_bytes)) = index_iter.next() else {
                 break;
             };
@@ -463,7 +504,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
         for (handle, block_keys) in target_blocks {
             let block = self.read_data_block(&handle).await?;
             for key in block_keys {
-                let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
+                let (found, mut entry_iter) = block.seek_and_iter(key, &cmp)?;
                 if let (true, Some((_entry_key, value))) = (found, entry_iter.next()) {
                     posting_list::add_to(value, self.file_version, &mut result)?;
                 }
diff --git a/crates/paimon/src/btree/tests.rs b/crates/paimon/src/btree/tests.rs
index 9e9d6c92f..105e9b71f 100644
--- a/crates/paimon/src/btree/tests.rs
+++ b/crates/paimon/src/btree/tests.rs
@@ -49,14 +49,14 @@ fn int_key(v: i32) -> Vec<u8> {
     v.to_be_bytes().to_vec()
 }
 
-fn int_cmp(a: &[u8], b: &[u8]) -> std::cmp::Ordering {
+fn int_cmp(a: &[u8], b: &[u8]) -> crate::Result<std::cmp::Ordering> {
     let a_val = i32::from_be_bytes(a.try_into().unwrap());
     let b_val = i32::from_be_bytes(b.try_into().unwrap());
-    a_val.cmp(&b_val)
+    Ok(a_val.cmp(&b_val))
 }
 
 /// Helper: write entries, finish, then open a reader from the in-memory bytes.
-async fn write_and_open<F: Fn(&[u8], &[u8]) -> std::cmp::Ordering>(
+async fn write_and_open<F: Fn(&[u8], &[u8]) -> crate::Result<std::cmp::Ordering>>(
     buf: &VecFileWrite,
     result: &crate::btree::writer::BTreeWriteResult,
     cmp: F,
@@ -667,7 +667,7 @@ async fn test_prefix_query() {
     }
 
     let result = writer.finish().await.unwrap();
-    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| a.cmp(b)).await;
+    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| Ok(a.cmp(b))).await;
 
     let bm = reader.query_prefix(b"ap").await.unwrap();
     assert_eq!(bm.len(), 3);
@@ -709,7 +709,7 @@ async fn test_string_fallback_scan_query() {
     }
 
     let result = writer.finish().await.unwrap();
-    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| a.cmp(b)).await;
+    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| Ok(a.cmp(b))).await;
     let data_type = DataType::VarChar(VarCharType::string_type());
 
     let ends_with = reader
@@ -874,7 +874,7 @@ async fn test_string_keys() {
     }
 
     let result = writer.finish().await.unwrap();
-    let reader = write_and_open(&buf, &result, |a, b| a.cmp(b)).await;
+    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| Ok(a.cmp(b))).await;
 
     let bm = reader.query_equal(b"apple").await.unwrap();
     assert!(bm.contains(0));
@@ -893,7 +893,7 @@ async fn test_string_keys_query() {
     }
 
     let result = writer.finish().await.unwrap();
-    let reader = write_and_open(&buf, &result, |a, b| a.cmp(b)).await;
+    let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]| Ok(a.cmp(b))).await;
 
     let bm = reader.query_equal(b"cherry").await.unwrap();
     assert_eq!(bm.len(), 1);
@@ -1134,10 +1134,10 @@ fn le_int_key(v: i32) -> Vec<u8> {
     v.to_le_bytes().to_vec()
 }
 
-fn le_int_cmp(a: &[u8], b: &[u8]) -> std::cmp::Ordering {
+fn le_int_cmp(a: &[u8], b: &[u8]) -> crate::Result<std::cmp::Ordering> {
     let a_val = i32::from_le_bytes(a.try_into().unwrap());
     let b_val = i32::from_le_bytes(b.try_into().unwrap());
-    a_val.cmp(&b_val)
+    Ok(a_val.cmp(&b_val))
 }
 
 fn load_testdata(name: &str) -> Vec<u8> {
@@ -1145,7 +1145,7 @@ fn load_testdata(name: &str) -> Vec<u8> {
     std::fs::read(&path).unwrap_or_else(|e| panic!("Failed to read {path}: {e}"))
 }
 
-async fn open_testdata<F: Fn(&[u8], &[u8]) -> std::cmp::Ordering>(
+async fn open_testdata<F: Fn(&[u8], &[u8]) -> crate::Result<std::cmp::Ordering>>(
     name: &str,
     meta: &BTreeIndexMeta,
     cmp: F,
@@ -1218,7 +1218,12 @@ async fn test_java_compat_int_with_nulls() {
 #[tokio::test]
 async fn test_java_compat_varchar_no_compress() {
     let meta = BTreeIndexMeta::new(Some(b"a".to_vec()), Some(b"yyyy".to_vec()), false);
-    let reader = open_testdata("btree_varchar_100_no_compress.bin", &meta, |a, b| a.cmp(b)).await;
+    let reader = open_testdata(
+        "btree_varchar_100_no_compress.bin",
+        &meta,
+        |a: &[u8], b: &[u8]| Ok(a.cmp(b)),
+    )
+    .await;
 
     let all = reader.all_non_null_rows().await.unwrap();
     assert_eq!(all.len(), 100);
diff --git a/crates/paimon/src/btree/var_len.rs b/crates/paimon/src/btree/var_len.rs
index 33e71cb17..c0b312cbc 100644
--- a/crates/paimon/src/btree/var_len.rs
+++ b/crates/paimon/src/btree/var_len.rs
@@ -97,6 +97,10 @@ pub fn encode_var_int_to_slice(bytes: &mut [u8], offset: usize, value: i32) -> u
 }
 
 /// Decode var-int from a byte slice, returning (value, bytes_consumed).
+///
+/// Panics on bytes that are not a var-int, so it is only for block-header bytes the
+/// reader has already validated. Use [`try_decode_var_int_from_slice`] for bytes whose
+/// shape is not known, such as a key read from an index file.
 pub fn decode_var_int_from_slice(bytes: &[u8], offset: usize) -> (i32, usize) {
     let mut result: u32 = 0;
     let mut i = 0;
@@ -111,6 +115,42 @@ pub fn decode_var_int_from_slice(bytes: &[u8], offset: usize) -> (i32, usize) {
     panic!("Malformed integer");
 }
 
+/// [`decode_var_int_from_slice`] for bytes that may not hold a var-int at all.
+///
+/// A BTree key is read with a comparator built from the column's *current* type, so a
+/// key written before a type change can be too short for the var-int the new arm
+/// expects. Running off the end or hitting a fifth continuation byte is then a property
+/// of the stored data, not a bug, and must be reported instead of panicking.
+pub fn try_decode_var_int_from_slice(bytes: &[u8], offset: usize) -> crate::Result<(i32, usize)> {
+    let mut result: u32 = 0;
+    let mut i = 0;
+    for shift in (0..32).step_by(7) {
+        let b = offset
+            .checked_add(i)
+            .and_then(|index| bytes.get(index))
+            .ok_or_else(|| crate::Error::DataInvalid {
+                message: format!(
+                    "Var-int at offset {offset} runs past the end of a {}-byte slice",
+                    bytes.len()
+                ),
+                source: None,
+            })?;
+        let b = *b as u32;
+        result |= (b & 0x7F) << shift;
+        i += 1;
+        if (b & 0x80) == 0 {
+            return Ok((result as i32, i));
+        }
+    }
+    Err(crate::Error::DataInvalid {
+        message: format!(
+            "Malformed var-int at offset {offset} of a {}-byte slice",
+            bytes.len()
+        ),
+        source: None,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -143,5 +183,21 @@ mod tests {
         let (decoded, consumed) = decode_var_int_from_slice(&buf, 0);
         assert_eq!(300, decoded);
         assert_eq!(written, consumed);
+        assert_eq!(
+            try_decode_var_int_from_slice(&buf, 0).unwrap(),
+            (300, written)
+        );
+    }
+
+    /// The infallible decoder panics on these; a key read from an index file built
+    /// before the column's type changed reaches the fallible one instead.
+    #[test]
+    fn test_try_decode_var_int_reports_malformed_bytes_instead_of_panicking() {
+        // Offset past the end, and a truncated continuation.
+        assert!(try_decode_var_int_from_slice(&[], 0).is_err());
+        assert!(try_decode_var_int_from_slice(&[0x00], 1).is_err());
+        assert!(try_decode_var_int_from_slice(&[0x80], 0).is_err());
+        // Five continuation bytes never terminate an i32.
+        assert!(try_decode_var_int_from_slice(&[0x80, 0x80, 0x80, 0x80, 0x80], 0).is_err());
     }
 }
diff --git a/crates/paimon/src/btree/writer.rs b/crates/paimon/src/btree/writer.rs
index 0ded13926..47708ef7a 100644
--- a/crates/paimon/src/btree/writer.rs
+++ b/crates/paimon/src/btree/writer.rs
@@ -23,6 +23,7 @@
 
 use crate::btree::block::{BlockCompressionType, BlockHandle};
 use crate::btree::footer::BTreeFileFooter;
+use crate::btree::key_serde::key_comparison_io_error;
 use crate::btree::meta::BTreeIndexMeta;
 use crate::btree::posting_list;
 use crate::btree::sst_file::SstFileWriter;
@@ -38,7 +39,7 @@ const BLOOM_FILTER_FPP: f64 = 0.05;
 /// Usage:
 /// 1. Call `write(key, row_id)` for each entry (keys must be sorted).
 /// 2. Call `finish()` to close the file and get the index meta.
-pub struct BTreeIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub struct BTreeIndexWriter<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
     sst_writer: SstFileWriter,
     current_row_ids: Vec<i64>,
     last_key: Option<Vec<u8>>,
@@ -57,7 +58,7 @@ pub struct BTreeWriteResult {
     pub row_count: u64,
 }
 
-impl BTreeIndexWriter<fn(&[u8], &[u8]) -> Ordering> {
+impl BTreeIndexWriter<fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
     pub fn new(
         writer: Box<dyn FileWrite>,
         block_size: usize,
@@ -84,13 +85,13 @@ impl BTreeIndexWriter<fn(&[u8], &[u8]) -> Ordering> {
             first_key: None,
             null_bitmap: None,
             row_count: 0,
-            key_comparator: |a, b| a.cmp(b),
+            key_comparator: |a, b| Ok(a.cmp(b)),
             file_version: 1,
         }
     }
 }
 
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BTreeIndexWriter<F> {
     /// Create a writer with a custom key comparator.
     pub fn with_comparator(
         writer: Box<dyn FileWrite>,
@@ -179,7 +180,10 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
             }
             Some(k) => {
                 if let Some(ref last) = self.last_key {
-                    if (self.key_comparator)(k, last) != Ordering::Equal {
+                    // The build side serializes both keys from the column's current
+                    // type, so a failure here is a real defect, not schema evolution.
+                    let order = (self.key_comparator)(k, last).map_err(key_comparison_io_error)?;
+                    if order != Ordering::Equal {
                         self.flush_row_ids().await?;
                     }
                 }
diff --git a/crates/paimon/src/table/bitmap_global_index_format.rs b/crates/paimon/src/table/bitmap_global_index_format.rs
index 80fc5bec2..a06ea9953 100644
--- a/crates/paimon/src/table/bitmap_global_index_format.rs
+++ b/crates/paimon/src/table/bitmap_global_index_format.rs
@@ -16,7 +16,7 @@
 
 //! Shared key and wire-format primitives for Java-compatible bitmap indexes.
 
-use crate::btree::key_serde::KeyComparator;
+use crate::btree::key_serde::{fixed_key_bytes, KeyComparator};
 use crate::btree::{make_key_comparator, serialize_datum};
 use crate::spec::{DataType, Datum, PredicateOperator};
 use std::cmp::Ordering;
@@ -39,14 +39,14 @@ pub(super) struct BlockInfo {
 pub(crate) fn make_bitmap_key_comparator(data_type: &DataType) -> KeyComparator {
     match data_type {
         DataType::Float(_) => Box::new(|left, right| {
-            let left = f32::from_le_bytes(left[..4].try_into().unwrap());
-            let right = f32::from_le_bytes(right[..4].try_into().unwrap());
-            compare_float_like_java(left, right)
+            let left = f32::from_le_bytes(fixed_key_bytes::<4>(left, "FLOAT")?);
+            let right = f32::from_le_bytes(fixed_key_bytes::<4>(right, "FLOAT")?);
+            Ok(compare_float_like_java(left, right))
         }),
         DataType::Double(_) => Box::new(|left, right| {
-            let left = f64::from_le_bytes(left[..8].try_into().unwrap());
-            let right = f64::from_le_bytes(right[..8].try_into().unwrap());
-            compare_double_like_java(left, right)
+            let left = f64::from_le_bytes(fixed_key_bytes::<8>(left, "DOUBLE")?);
+            let right = f64::from_le_bytes(fixed_key_bytes::<8>(right, "DOUBLE")?);
+            Ok(compare_double_like_java(left, right))
         }),
         _ => make_key_comparator(data_type),
     }
diff --git a/crates/paimon/src/table/bitmap_global_index_reader.rs b/crates/paimon/src/table/bitmap_global_index_reader.rs
index 73e4c6ad9..24d43f61f 100644
--- a/crates/paimon/src/table/bitmap_global_index_reader.rs
+++ b/crates/paimon/src/table/bitmap_global_index_reader.rs
@@ -25,6 +25,7 @@ use super::bitmap_global_index_format::{
 };
 #[cfg(test)]
 use super::bitmap_global_index_writer::BitmapGlobalIndexWriter;
+use crate::btree::key_serde::key_comparison_io_error;
 use crate::btree::var_len::{decode_var_int, decode_var_long};
 use crate::btree::{compute_crc32, decompress_block, BlockCompressionType};
 use crate::io::FileRead;
@@ -62,6 +63,21 @@ pub(crate) struct BitmapGlobalIndexReader {
     dictionary_block_cache: Mutex<HashMap<BlockInfo, Arc<Vec<DictionaryEntry>>>>,
 }
 
+/// A key comparator for this reader's `io::Result` signatures.
+///
+/// A dictionary key was written with the type the column had when the index was built,
+/// so comparing it against a literal serialized from the column's current type can fail.
+/// The failure is wrapped so the caller can tell it from a real I/O failure and give up
+/// on the index rather than failing the query.
+type IoKeyComparator<'a> = dyn Fn(&[u8], &[u8]) -> io::Result<Ordering> + Send + Sync + 'a;
+
+fn io_key_comparator(
+    data_type: &DataType,
+) -> impl Fn(&[u8], &[u8]) -> io::Result<Ordering> + Send + Sync {
+    let cmp = make_bitmap_key_comparator(data_type);
+    move |left, right| cmp(left, right).map_err(key_comparison_io_error)
+}
+
 impl BitmapGlobalIndexReader {
     pub(crate) async fn open(reader: Box<dyn FileRead>, file_size: u64) -> io::Result<Self> {
         let footer = read_footer(reader.as_ref(), file_size).await?;
@@ -146,29 +162,39 @@ impl BitmapGlobalIndexReader {
             PredicateOperator::IsNotNull => self.is_not_null().await,
             PredicateOperator::Lt => {
                 let key = serialize_bitmap_datum(&literals[0], data_type);
-                self.scan_dictionary(data_type, |candidate, cmp| cmp(candidate, &key).is_lt())
-                    .await
+                self.scan_dictionary(
+                    data_type,
+                    |candidate, cmp| Ok(cmp(candidate, &key)?.is_lt()),
+                )
+                .await
             }
             PredicateOperator::LtEq => {
                 let key = serialize_bitmap_datum(&literals[0], data_type);
-                self.scan_dictionary(data_type, |candidate, cmp| !cmp(candidate, &key).is_gt())
-                    .await
+                self.scan_dictionary(data_type, |candidate, cmp| {
+                    Ok(!cmp(candidate, &key)?.is_gt())
+                })
+                .await
             }
             PredicateOperator::Gt => {
                 let key = serialize_bitmap_datum(&literals[0], data_type);
-                self.scan_dictionary(data_type, |candidate, cmp| cmp(candidate, &key).is_gt())
-                    .await
+                self.scan_dictionary(
+                    data_type,
+                    |candidate, cmp| Ok(cmp(candidate, &key)?.is_gt()),
+                )
+                .await
             }
             PredicateOperator::GtEq => {
                 let key = serialize_bitmap_datum(&literals[0], data_type);
-                self.scan_dictionary(data_type, |candidate, cmp| !cmp(candidate, &key).is_lt())
-                    .await
+                self.scan_dictionary(data_type, |candidate, cmp| {
+                    Ok(!cmp(candidate, &key)?.is_lt())
+                })
+                .await
             }
             PredicateOperator::Between => {
                 let from = serialize_bitmap_datum(&literals[0], data_type);
                 let to = serialize_bitmap_datum(&literals[1], data_type);
                 self.scan_dictionary(data_type, |candidate, cmp| {
-                    !cmp(candidate, &from).is_lt() && !cmp(candidate, &to).is_gt()
+                    Ok(!cmp(candidate, &from)?.is_lt() && !cmp(candidate, &to)?.is_gt())
                 })
                 .await
             }
@@ -178,7 +204,7 @@ impl BitmapGlobalIndexReader {
                 let to = serialize_bitmap_datum(&literals[1], data_type);
                 let inside = self
                     .scan_dictionary(data_type, |candidate, cmp| {
-                        !cmp(candidate, &from).is_lt() && !cmp(candidate, &to).is_gt()
+                        Ok(!cmp(candidate, &from)?.is_lt() && !cmp(candidate, &to)?.is_gt())
                     })
                     .await?;
                 result -= inside;
@@ -195,7 +221,7 @@ impl BitmapGlobalIndexReader {
                 if prefix.is_empty() {
                     return self.is_not_null().await;
                 }
-                self.scan_serialized_dictionary(|candidate| candidate.starts_with(&prefix))
+                self.scan_serialized_dictionary(|candidate| Ok(candidate.starts_with(&prefix)))
                     .await
             }
             PredicateOperator::EndsWith => {
@@ -209,7 +235,7 @@ impl BitmapGlobalIndexReader {
                 if suffix.is_empty() {
                     return self.is_not_null().await;
                 }
-                self.scan_serialized_dictionary(|candidate| candidate.ends_with(&suffix))
+                self.scan_serialized_dictionary(|candidate| Ok(candidate.ends_with(&suffix)))
                     .await
             }
             PredicateOperator::Contains => {
@@ -223,7 +249,7 @@ impl BitmapGlobalIndexReader {
                 if needle.is_empty() {
                     return self.is_not_null().await;
                 }
-                self.scan_serialized_dictionary(|candidate| contains_bytes(candidate, &needle))
+                self.scan_serialized_dictionary(|candidate| Ok(contains_bytes(candidate, &needle)))
                     .await
             }
             PredicateOperator::Like => {
@@ -235,7 +261,8 @@ impl BitmapGlobalIndexReader {
                 }
                 let pattern = string_literal(literals, op)?.to_string();
                 self.scan_serialized_dictionary(|candidate| {
-                    std::str::from_utf8(candidate).is_ok_and(|value| like_match(value, &pattern))
+                    Ok(std::str::from_utf8(candidate)
+                        .is_ok_and(|value| like_match(value, &pattern)))
                 })
                 .await
             }
@@ -254,10 +281,10 @@ impl BitmapGlobalIndexReader {
             return self.is_not_null().await;
         }
         self.scan_dictionary(data_type, |candidate, cmp| {
-            let from_cmp = cmp(candidate, from);
-            let to_cmp = cmp(candidate, to);
-            (from_cmp.is_gt() || (from_inclusive && from_cmp.is_eq()))
-                && (to_cmp.is_lt() || (to_inclusive && to_cmp.is_eq()))
+            let from_cmp = cmp(candidate, from)?;
+            let to_cmp = cmp(candidate, to)?;
+            Ok((from_cmp.is_gt() || (from_inclusive && from_cmp.is_eq()))
+                && (to_cmp.is_lt() || (to_inclusive && to_cmp.is_eq())))
         })
         .await
     }
@@ -271,14 +298,14 @@ impl BitmapGlobalIndexReader {
     }
 
     async fn equal(&self, key: &[u8], data_type: &DataType) -> io::Result<RoaringTreemap> {
-        let logical_cmp = make_bitmap_key_comparator(data_type);
-        self.equal_with_comparator(key, logical_cmp.as_ref()).await
+        let logical_cmp = io_key_comparator(data_type);
+        self.equal_with_comparator(key, &logical_cmp).await
     }
 
     async fn equal_with_comparator(
         &self,
         key: &[u8],
-        logical_cmp: &(dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync),
+        logical_cmp: &IoKeyComparator<'_>,
     ) -> io::Result<RoaringTreemap> {
         match self.find_bitmap_block(key, logical_cmp).await? {
             Some(block) => self.read_bitmap(block).await,
@@ -291,12 +318,10 @@ impl BitmapGlobalIndexReader {
         sorted_keys.sort();
         sorted_keys.dedup();
 
-        let logical_cmp = make_bitmap_key_comparator(data_type);
+        let logical_cmp = io_key_comparator(data_type);
         let mut result = RoaringTreemap::new();
         for key in sorted_keys {
-            result |= self
-                .equal_with_comparator(&key, logical_cmp.as_ref())
-                .await?;
+            result |= self.equal_with_comparator(&key, &logical_cmp).await?;
         }
         Ok(result)
     }
@@ -304,21 +329,21 @@ impl BitmapGlobalIndexReader {
     async fn scan_dictionary(
         &self,
         data_type: &DataType,
-        predicate: impl Fn(&[u8], &dyn Fn(&[u8], &[u8]) -> Ordering) -> bool,
+        predicate: impl Fn(&[u8], &IoKeyComparator<'_>) -> io::Result<bool>,
     ) -> io::Result<RoaringTreemap> {
-        let cmp = make_bitmap_key_comparator(data_type);
-        self.scan_serialized_dictionary(|candidate| predicate(candidate, cmp.as_ref()))
+        let cmp = io_key_comparator(data_type);
+        self.scan_serialized_dictionary(|candidate| predicate(candidate, &cmp))
             .await
     }
 
     async fn scan_serialized_dictionary(
         &self,
-        predicate: impl Fn(&[u8]) -> bool,
+        predicate: impl Fn(&[u8]) -> io::Result<bool>,
     ) -> io::Result<RoaringTreemap> {
         let mut result = RoaringTreemap::new();
         for block_meta in &self.dictionary_blocks {
             for entry in self.read_dictionary_block(block_meta.block).await?.iter() {
-                if predicate(&entry.key) {
+                if predicate(&entry.key)? {
                     result |= self.read_bitmap(entry.bitmap_block).await?;
                 }
             }
@@ -329,13 +354,13 @@ impl BitmapGlobalIndexReader {
     async fn find_bitmap_block(
         &self,
         key: &[u8],
-        logical_cmp: &(dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync),
+        logical_cmp: &IoKeyComparator<'_>,
     ) -> io::Result<Option<BlockInfo>> {
-        let Some(block_meta) = self.find_dictionary_block_meta(key, logical_cmp) else {
+        let Some(block_meta) = self.find_dictionary_block_meta(key, logical_cmp)? else {
             return Ok(None);
         };
         for entry in self.read_dictionary_block(block_meta.block).await?.iter() {
-            match logical_cmp(&entry.key, key) {
+            match logical_cmp(&entry.key, key)? {
                 Ordering::Equal => return Ok(Some(entry.bitmap_block)),
                 Ordering::Greater => return Ok(None),
                 Ordering::Less => {}
@@ -347,23 +372,24 @@ impl BitmapGlobalIndexReader {
     fn find_dictionary_block_meta(
         &self,
         key: &[u8],
-        compare: impl Fn(&[u8], &[u8]) -> Ordering,
-    ) -> Option<&DictionaryBlockMeta> {
+        compare: impl Fn(&[u8], &[u8]) -> io::Result<Ordering>,
+    ) -> io::Result<Option<&DictionaryBlockMeta>> {
         if self.dictionary_blocks.is_empty() {
-            return None;
+            return Ok(None);
         }
         let mut low = 0usize;
         let mut high = self.dictionary_blocks.len();
         while low < high {
             let mid = (low + high) / 2;
-            if compare(&self.dictionary_blocks[mid].first_key, key) != Ordering::Greater {
+            if compare(&self.dictionary_blocks[mid].first_key, key)? != Ordering::Greater {
                 low = mid + 1;
             } else {
                 high = mid;
             }
         }
-        low.checked_sub(1)
-            .and_then(|index| self.dictionary_blocks.get(index))
+        Ok(low
+            .checked_sub(1)
+            .and_then(|index| self.dictionary_blocks.get(index)))
     }
 
     async fn read_dictionary_block(
@@ -1006,9 +1032,10 @@ mod tests {
         ];
         for (op, literals, expected) in cases {
             let key = serialize_bitmap_datum(&literals[0], &data_type);
-            let cmp = make_bitmap_key_comparator(&data_type);
+            let cmp = io_key_comparator(&data_type);
             let expected_block = reader
-                .find_dictionary_block_meta(&key, cmp.as_ref())
+                .find_dictionary_block_meta(&key, cmp)
+                .unwrap()
                 .unwrap()
                 .block;
             let was_cached = reader
diff --git a/crates/paimon/src/table/bitmap_global_index_writer.rs b/crates/paimon/src/table/bitmap_global_index_writer.rs
index 9283c2008..a235c4dc0 100644
--- a/crates/paimon/src/table/bitmap_global_index_writer.rs
+++ b/crates/paimon/src/table/bitmap_global_index_writer.rs
@@ -17,6 +17,7 @@
 //! Writer for Java Paimon's `BitmapGlobalIndexFormat`.
 
 use super::bitmap_global_index_format::{BlockInfo, MAGIC, VERSION};
+use crate::btree::key_serde::key_comparison_io_error;
 use crate::btree::var_len::{encode_var_int, encode_var_long};
 use crate::btree::{compress_block, compute_crc32, BTreeIndexMeta, BlockCompressionType};
 use crate::io::FileWrite;
@@ -32,7 +33,7 @@ pub(crate) struct BitmapWriteResult {
     pub(crate) row_count: u64,
 }
 
-pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
     writer: Box<dyn FileWrite>,
     dictionary_block_size: usize,
     compression_type: BlockCompressionType,
@@ -46,7 +47,7 @@ pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
     row_count: u64,
 }
 
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BitmapGlobalIndexWriter<F> {
     #[cfg(test)]
     pub(crate) fn new(
         writer: Box<dyn FileWrite>,
@@ -99,7 +100,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
                 let row_id = relative_row_id as u64;
                 self.non_null_rows.insert(row_id);
                 self.bitmaps.entry(key.to_vec()).or_default().insert(row_id);
-                self.update_min_max(key);
+                self.update_min_max(key)?;
             }
             None => {
                 self.null_rows.insert(relative_row_id as u64);
@@ -119,7 +120,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
         }
         let row_id = relative_row_id as u64;
         self.bitmaps.entry(key.to_vec()).or_default().insert(row_id);
-        self.update_min_max(key);
+        self.update_min_max(key)?;
         Ok(())
     }
 
@@ -141,7 +142,21 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
         let mut bitmaps = std::mem::take(&mut self.bitmaps)
             .into_iter()
             .collect::<Vec<_>>();
-        bitmaps.sort_by(|(left, _), (right, _)| (self.key_comparator)(left, right));
+        // The build side serializes every key from the column's current type, so a
+        // comparison failure here is a real defect rather than schema evolution.
+        let mut failure = None;
+        bitmaps.sort_by(
+            |(left, _), (right, _)| match (self.key_comparator)(left, right) {
+                Ok(order) => order,
+                Err(error) => {
+                    failure.get_or_insert(error);
+                    Ordering::Equal
+                }
+            },
+        );
+        if let Some(error) = failure {
+            return Err(key_comparison_io_error(error));
+        }
 
         let mut bytes = Vec::new();
         write_bitmap_index_bytes(
@@ -168,21 +183,26 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
         })
     }
 
-    fn update_min_max(&mut self, key: &[u8]) {
-        if self
-            .first_key
-            .as_ref()
-            .is_none_or(|existing| (self.key_comparator)(key, existing).is_lt())
-        {
+    fn update_min_max(&mut self, key: &[u8]) -> io::Result<()> {
+        let replaces_first = match &self.first_key {
+            None => true,
+            Some(existing) => (self.key_comparator)(key, existing)
+                .map_err(key_comparison_io_error)?
+                .is_lt(),
+        };
+        if replaces_first {
             self.first_key = Some(key.to_vec());
         }
-        if self
-            .last_key
-            .as_ref()
-            .is_none_or(|existing| (self.key_comparator)(key, existing).is_gt())
-        {
+        let replaces_last = match &self.last_key {
+            None => true,
+            Some(existing) => (self.key_comparator)(key, existing)
+                .map_err(key_comparison_io_error)?
+                .is_gt(),
+        };
+        if replaces_last {
             self.last_key = Some(key.to_vec());
         }
+        Ok(())
     }
 }
 
diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs
index 5fbeffa67..8c721b8cd 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -54,7 +54,7 @@ use tokio::sync::Semaphore;
 #[cfg(test)]
 use std::sync::atomic::{AtomicUsize as TestAtomicUsize, Ordering as TestOrdering};
 
-type BoxedCmp = Box<dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync>;
+type BoxedCmp = Box<dyn Fn(&[u8], &[u8]) -> Result<Ordering> + Send + Sync>;
 
 const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
 
diff --git a/crates/paimon/src/table/global_index_scanner/all_match.rs b/crates/paimon/src/table/global_index_scanner/all_match.rs
index 57da267b4..1b60fa730 100644
--- a/crates/paimon/src/table/global_index_scanner/all_match.rs
+++ b/crates/paimon/src/table/global_index_scanner/all_match.rs
@@ -18,7 +18,8 @@
 //! Prove complete scalar BTree domains match without opening index files.
 
 use super::entry::{sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind};
-use crate::btree::{make_key_comparator, serialize_datum};
+use crate::btree::key_serde::KeyComparator;
+use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta};
 use crate::spec::{DataType, Datum, PredicateOperator};
 use std::cmp::Ordering::{Equal, Greater, Less};
 use std::collections::HashMap;
@@ -69,28 +70,12 @@ pub(super) fn all_matching_entries(
         }
         for (predicate_index, (op, values, cmp)) in comparisons.iter().enumerate() {
             let all_match = files.iter().all(|&index| {
-                let meta = sorted_entry_meta(entries[index]);
-                let (Some(first), Some(last)) = (&meta.first_key, &meta.last_key) else {
-                    return false;
-                };
-                !meta.has_nulls && {
-                    if cmp(first, last) == Greater {
-                        return false;
-                    }
-                    match (op, values.as_slice()) {
-                        (PredicateOperator::Eq, [value]) => {
-                            cmp(first, value) == Equal && cmp(last, value) == Equal
-                        }
-                        (PredicateOperator::Lt, [value]) => cmp(last, value) == Less,
-                        (PredicateOperator::LtEq, [value]) => cmp(last, value) != Greater,
-                        (PredicateOperator::Gt, [value]) => cmp(first, value) == Greater,
-                        (PredicateOperator::GtEq, [value]) => cmp(first, value) != Less,
-                        (PredicateOperator::Between, [from, to]) => {
-                            cmp(first, from) != Less && cmp(last, to) != Greater
-                        }
-                        _ => false,
-                    }
-                }
+                // Degradation: pure optimisation. This only proves that a file's whole
+                // row range matches so the posting list need not be decoded. Without an
+                // ordering nothing is proven, so the entry is simply not an all-match
+                // and the ordinary query path handles it.
+                entry_all_matches(sorted_entry_meta(entries[index]), *op, values, cmp)
+                    .unwrap_or(false)
             });
             if all_match {
                 for &index in &files {
@@ -101,3 +86,30 @@ pub(super) fn all_matching_entries(
     }
     result
 }
+
+fn entry_all_matches(
+    meta: &BTreeIndexMeta,
+    op: PredicateOperator,
+    values: &[Vec<u8>],
+    cmp: &KeyComparator,
+) -> crate::Result<bool> {
+    let (Some(first), Some(last)) = (&meta.first_key, &meta.last_key) else {
+        return Ok(false);
+    };
+    if meta.has_nulls || cmp(first, last)? == Greater {
+        return Ok(false);
+    }
+    Ok(match (op, values) {
+        (PredicateOperator::Eq, [value]) => {
+            cmp(first, value)? == Equal && cmp(last, value)? == Equal
+        }
+        (PredicateOperator::Lt, [value]) => cmp(last, value)? == Less,
+        (PredicateOperator::LtEq, [value]) => cmp(last, value)? != Greater,
+        (PredicateOperator::Gt, [value]) => cmp(first, value)? == Greater,
+        (PredicateOperator::GtEq, [value]) => cmp(first, value)? != Less,
+        (PredicateOperator::Between, [from, to]) => {
+            cmp(first, from)? != Less && cmp(last, to)? != Greater
+        }
+        _ => false,
+    })
+}
diff --git a/crates/paimon/src/table/global_index_scanner/entry.rs b/crates/paimon/src/table/global_index_scanner/entry.rs
index d402f69b8..4d21e5021 100644
--- a/crates/paimon/src/table/global_index_scanner/entry.rs
+++ b/crates/paimon/src/table/global_index_scanner/entry.rs
@@ -17,12 +17,12 @@
 
 //! Manifest entry parsing and file-level pruning metadata.
 
+use crate::btree::key_serde::DynKeyComparator;
 use crate::btree::BTreeIndexMeta;
 use crate::spec::{DataType, PredicateOperator};
 use crate::table::bitmap_global_index_format::is_bitmap_floating_residual_sensitive_op;
 use crate::table::index_file_path::IndexFileLocation;
 use crate::{Error, Result};
-use std::cmp::Ordering;
 use std::collections::HashMap;
 
 /// A resolved global index entry with parsed metadata.
@@ -167,7 +167,7 @@ pub(super) fn bitmap_meta_may_match(
     op: PredicateOperator,
     data_type: &DataType,
     serialized_literals: &[Vec<u8>],
-    cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+    cmp: &DynKeyComparator<'_>,
 ) -> bool {
     if is_floating_point(data_type) && is_bitmap_floating_residual_sensitive_op(op) {
         !meta.only_nulls()
@@ -181,7 +181,7 @@ pub(super) fn bitmap_meta_may_match_between(
     data_type: &DataType,
     from_key: &[u8],
     to_key: &[u8],
-    cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+    cmp: &DynKeyComparator<'_>,
 ) -> bool {
     if is_floating_point(data_type)
         && is_bitmap_floating_residual_sensitive_op(PredicateOperator::Between)
@@ -196,7 +196,7 @@ pub(super) fn multivalue_meta_may_match(
     meta: &BTreeIndexMeta,
     op: PredicateOperator,
     serialized_literals: &[Vec<u8>],
-    cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+    cmp: &DynKeyComparator<'_>,
 ) -> bool {
     match op {
         PredicateOperator::ArrayContains => {
diff --git a/crates/paimon/src/table/global_index_scanner/reader.rs b/crates/paimon/src/table/global_index_scanner/reader.rs
index 246eeccd0..fc011e106 100644
--- a/crates/paimon/src/table/global_index_scanner/reader.rs
+++ b/crates/paimon/src/table/global_index_scanner/reader.rs
@@ -22,6 +22,7 @@ use super::entry::{
 };
 use super::query_plan::{add_file_size, EntryQueryPlan, EntryQueryResult, FallbackScanPlan};
 use super::{BoxedCmp, GlobalIndexScanner};
+use crate::btree::key_serde::is_key_comparison_failure;
 use crate::btree::query::{BetweenInfo, IndexQuery};
 use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta, BTreeIndexReader};
 use crate::fm_index::FMGlobalIndexReader;
@@ -129,7 +130,7 @@ impl GlobalIndexScanner {
             };
             let from_key = serialize_key(between.from, between.data_type);
             let to_key = serialize_key(between.to, between.data_type);
-            let bitmap = reader
+            let bitmap = match reader
                 .as_ref()
                 .expect("reader is opened when between matches")
                 .range_query(
@@ -140,7 +141,22 @@ impl GlobalIndexScanner {
                     between.to_inclusive,
                 )
                 .await
-                .map_err(|error| Self::query_error(entry, error))?;
+            {
+                Ok(bitmap) => bitmap,
+                // Degradation: this predicate is not evaluated by the global index at
+                // all. The stored keys are not keys of the column's current type, so
+                // this file cannot answer; declining makes `evaluate_leaf` return
+                // `Ok(None)` and the predicate falls through to the read pipeline.
+                // Returning no rows would silently drop rows, and failing the query
+                // would turn a readable table into an error.
+                Err(error) if is_key_comparison_failure(&error) => {
+                    return Ok(EntryQueryResult {
+                        bitmap: None,
+                        declined: true,
+                    })
+                }
+                Err(error) => return Err(Self::query_error(entry, error)),
+            };
             file_result = Some(bitmap);
         }
 
@@ -152,13 +168,18 @@ impl GlobalIndexScanner {
                     .fetch_add(1, super::TestOrdering::SeqCst);
             }
             let (op, literals, data_type) = &effective_predicates[idx];
-            let Some(bitmap) = reader
+            let queried = reader
                 .as_ref()
                 .expect("reader is opened when predicates match")
                 .query(*op, literals, data_type)
-                .await
-                .map_err(|error| Self::query_error(entry, error))?
-            else {
+                .await;
+            let Some(bitmap) = (match queried {
+                Ok(bitmap) => bitmap,
+                // Degradation: this predicate is not evaluated by the global index at
+                // all, for the same reason as the between query above.
+                Err(error) if is_key_comparison_failure(&error) => None,
+                Err(error) => return Err(Self::query_error(entry, error)),
+            }) else {
                 return Ok(EntryQueryResult {
                     bitmap: None,
                     declined: true,
diff --git a/crates/paimon/src/table/global_index_scanner/tests.rs b/crates/paimon/src/table/global_index_scanner/tests.rs
index 4bd731ab6..fc055c95d 100644
--- a/crates/paimon/src/table/global_index_scanner/tests.rs
+++ b/crates/paimon/src/table/global_index_scanner/tests.rs
@@ -1929,12 +1929,12 @@ fn legacy_floating_comparator(data_type: &DataType) -> BoxedCmp {
         DataType::Float(_) => Box::new(|left, right| {
             let left = f32::from_le_bytes(left.try_into().unwrap());
             let right = f32::from_le_bytes(right.try_into().unwrap());
-            left.total_cmp(&right)
+            Ok(left.total_cmp(&right))
         }),
         DataType::Double(_) => Box::new(|left, right| {
             let left = f64::from_le_bytes(left.try_into().unwrap());
             let right = f64::from_le_bytes(right.try_into().unwrap());
-            left.total_cmp(&right)
+            Ok(left.total_cmp(&right))
         }),
         _ => unreachable!("legacy floating comparator requires Float or Double"),
     }
@@ -1955,7 +1955,7 @@ async fn assert_legacy_floating_btree(
         .collect::<Vec<_>>();
     rows.push((zero_key, 3));
     let cmp = legacy_floating_comparator(&data_type);
-    rows.sort_by(|left, right| cmp(&left.0, &right.0));
+    rows.sort_by(|left, right| cmp(&left.0, &right.0).unwrap());
     let expected_first_key = rows.first().unwrap().0.clone();
     let expected_last_key = rows.last().unwrap().0.clone();
 
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
index f50799c93..c136f796a 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
@@ -20,6 +20,7 @@
 use super::planning::SortedGlobalIndexShard;
 use super::validation::checked_row_count;
 use super::{SerializeKeyFn, SortedIndexKeyRow};
+use crate::btree::key_serde::DynKeyComparator;
 use crate::spec::{
     extract_datum_from_array, extract_datum_from_arrow, DataField, DataType, ROW_ID_FIELD_NAME,
 };
@@ -328,16 +329,30 @@ pub(super) fn extract_multivalue_index_rows_from_batches(
     Ok(rows)
 }
 
+/// Sort the extracted rows with the index's key comparator.
+///
+/// Both keys come from [`super::make_index_key_codec`] for the column's current type, so
+/// a comparison failure is a real defect rather than an index built before a type change;
+/// it aborts the build instead of degrading.
 pub(super) fn sort_index_rows(
     rows: &mut [SortedIndexKeyRow],
-    cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
-) {
+    cmp: &DynKeyComparator<'_>,
+) -> Result<()> {
+    let mut failure = None;
     rows.sort_by(|left, right| match (&left.0, &right.0) {
         (None, None) => left.1.cmp(&right.1),
         (None, Some(_)) => Ordering::Less,
         (Some(_), None) => Ordering::Greater,
-        (Some(left_key), Some(right_key)) => {
-            cmp(left_key, right_key).then_with(|| left.1.cmp(&right.1))
-        }
+        (Some(left_key), Some(right_key)) => match cmp(left_key, right_key) {
+            Ok(order) => order.then_with(|| left.1.cmp(&right.1)),
+            Err(error) => {
+                failure.get_or_insert(error);
+                left.1.cmp(&right.1)
+            }
+        },
     });
+    match failure {
+        Some(error) => Err(error),
+        None => Ok(()),
+    }
 }
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index cace6becf..aa0d0551b 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -475,13 +475,13 @@ fn test_index_key_codec_scopes_java_nan_semantics_to_bitmap() {
         let btree_nan_key = btree_serialize(&negative_nan, &data_type);
         let zero_key = btree_serialize(&zero, &data_type);
         assert_eq!(btree_nan_key, raw_nan_key);
-        assert!(btree_cmp(&btree_nan_key, &zero_key).is_lt());
+        assert!(btree_cmp(&btree_nan_key, &zero_key).unwrap().is_lt());
 
         let (bitmap_cmp, bitmap_serialize) =
             make_index_key_codec(BITMAP_GLOBAL_INDEX_TYPE, &data_type);
         let bitmap_nan_key = bitmap_serialize(&negative_nan, &data_type);
         assert_eq!(bitmap_nan_key, canonical_nan_key);
-        assert!(bitmap_cmp(&bitmap_nan_key, &zero_key).is_gt());
+        assert!(bitmap_cmp(&bitmap_nan_key, &zero_key).unwrap().is_gt());
     }
 
     assert_codec(
@@ -528,7 +528,7 @@ fn test_sort_index_rows_orders_nulls_then_keys() {
     ];
     let cmp = make_key_comparator(&DataType::Int(IntType::new()));
 
-    sort_index_rows(&mut rows, &cmp);
+    sort_index_rows(&mut rows, &cmp).unwrap();
 
     assert_eq!(
         rows,
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index aa71e5019..8ef05b2dd 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -81,7 +81,7 @@ impl SortedGlobalIndexBuildBuilder<'_> {
             .await?
         };
         if !rows.is_empty() {
-            sort_index_rows(&mut rows, &cmp);
+            sort_index_rows(&mut rows, &cmp)?;
         }
 
         self.table