fix(global-index): stop reading index keys with a comparator that can panic

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
+    );
+}

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);