Validate index rowid position and report corrupt records

Author: kumarUjjawal

core/storage/btree.rs

@@ -5853,6 +5853,10 @@ pub enum IntegrityCheckError {
         index_count: usize,
         table_count: usize,
     },
+    #[error("rowid not at end of record in index {index_name}")]
+    RowidNotAtEndOfRecord { index_name: String },
+    #[error("corrupt index record in index {index_name}: {error}")]
+    IndexRecordCorrupt { index_name: String, error: String },
     #[error("NULL value in {table_name}.{column_name}")]
     NotNullViolation {
         table_name: String,

core/vdbe/execute.rs

@@ -13,12 +13,12 @@ use crate::storage::database::DatabaseFile;
 use crate::storage::journal_mode;
 use crate::storage::page_cache::PageCache;
 use crate::storage::pager::{default_page1, CreateBTreeFlags, PageRef};
-use crate::storage::sqlite3_ondisk::{DatabaseHeader, PageSize, RawVersion};
+use crate::storage::sqlite3_ondisk::{read_varint, DatabaseHeader, PageSize, RawVersion};
 use crate::translate::collate::CollationSeq;
 use crate::translate::pragma::TURSO_CDC_VERSION_TABLE_NAME;
 use crate::types::{
     compare_immutable, compare_records_generic, AsValueRef, Extendable, IOCompletions, IOResult,
-    ImmutableRecord, IndexInfo, SeekResult, Text,
+    ImmutableRecord, IndexInfo, SeekResult, SerialType, SerialTypeKind, Text,
 };
 use crate::util::{
     normalize_ident, rewrite_check_expr_column_refs, rewrite_check_expr_table_refs,
@@ -9423,6 +9423,71 @@ fn format_integrity_check_result(errors: &[IntegrityCheckError]) -> String {
     }
 }
 
+fn index_record_rowid_at_end(record: &ImmutableRecord, index_info: &IndexInfo) -> Result<bool> {
+    if !index_info.has_rowid {
+        return Ok(true);
+    }
+
+    let payload = record.get_payload();
+    if payload.is_empty() {
+        return Err(LimboError::Corrupt("Index record payload is empty".into()));
+    }
+
+    let (header_size, header_varint_len) = read_varint(payload)?;
+    let header_size = header_size as usize;
+    if header_size > payload.len() || header_varint_len > payload.len() {
+        return Err(LimboError::Corrupt(
+            "Index record header exceeds payload size".into(),
+        ));
+    }
+
+    let mut header = &payload[header_varint_len..header_size];
+    let mut field_count = 0usize;
+    let mut data_size = 0usize;
+    let mut last_serial_type: Option<SerialType> = None;
+
+    while !header.is_empty() {
+        let (serial_type_raw, bytes_read) = read_varint(header)?;
+        header = &header[bytes_read..];
+
+        let serial_type = SerialType::try_from(serial_type_raw)?;
+        data_size = data_size
+            .checked_add(serial_type.size())
+            .ok_or_else(|| LimboError::Corrupt("Index record data size overflow".into()))?;
+        field_count += 1;
+        last_serial_type = Some(serial_type);
+    }
+
+    if field_count != index_info.num_cols {
+        return Err(LimboError::Corrupt(format!(
+            "Index record field count {} != expected {}",
+            field_count, index_info.num_cols
+        )));
+    }
+
+    let payload_data_len = payload.len().saturating_sub(header_size);
+    if data_size > payload_data_len {
+        return Err(LimboError::Corrupt(
+            "Index record data section too small".into(),
+        ));
+    }
+
+    let last_serial_type = last_serial_type
+        .ok_or_else(|| LimboError::Corrupt("Index record has no serial types".into()))?;
+
+    Ok(matches!(
+        last_serial_type.kind(),
+        SerialTypeKind::I8
+            | SerialTypeKind::I16
+            | SerialTypeKind::I24
+            | SerialTypeKind::I32
+            | SerialTypeKind::I48
+            | SerialTypeKind::I64
+            | SerialTypeKind::ConstInt0
+            | SerialTypeKind::ConstInt1
+    ))
+}
+
 pub enum OpIntegrityCheckState {
     Start,
     /// Phase 1: B-tree structural checks
@@ -9431,7 +9496,20 @@ pub enum OpIntegrityCheckState {
         current_root_idx: usize,
         state: IntegrityCheckState,
     },
-    /// Phase 2: Index entry count validation (runs for both quick_check and integrity_check)
+    /// Phase 2: Index rowid position validation (skipped when quick=true)
+    /// Scans index entries and ensures the rowid is the last column in each record.
+    ValidatingIndexRowidPosition {
+        errors: Vec<IntegrityCheckError>,
+        /// Index into the tables list
+        current_table_idx: usize,
+        /// Index into the current table's indexes list
+        current_index_idx: usize,
+        /// Cursor used for index scanning (persisted across I/O yields)
+        cursor: Option<Box<BTreeCursor>>,
+        /// State of the current validation sub-phase
+        substate: IndexRowidValidationSubstate,
+    },
+    /// Phase 3: Index entry count validation (runs for both quick_check and integrity_check)
     /// Counts rows in each table and entries in each index, then compares.
     ValidatingIndexCounts {
         errors: Vec<IntegrityCheckError>,
@@ -9448,7 +9526,7 @@ pub enum OpIntegrityCheckState {
         /// Cursor used for counting (persisted across I/O yields)
         cursor: Option<Box<BTreeCursor>>,
     },
-    /// Phase 3: NOT NULL constraint validation (skipped when quick=true)
+    /// Phase 4: NOT NULL constraint validation (skipped when quick=true)
     /// Scans table rows and checks that NOT NULL columns don't contain NULL values.
     ValidatingNotNull {
         errors: Vec<IntegrityCheckError>,
@@ -9459,7 +9537,7 @@ pub enum OpIntegrityCheckState {
         /// State of the current validation sub-phase
         substate: NotNullValidationSubstate,
     },
-    /// Phase 4: UNIQUE constraint validation (skipped when quick=true)
+    /// Phase 5: UNIQUE constraint validation (skipped when quick=true)
     /// Scans unique indexes and checks for duplicate keys.
     ValidatingUnique {
         errors: Vec<IntegrityCheckError>,
@@ -9475,7 +9553,7 @@ pub enum OpIntegrityCheckState {
         /// State of the current validation sub-phase
         substate: UniqueValidationSubstate,
     },
-    /// Phase 5: Row-index consistency validation (skipped when quick=true)
+    /// Phase 6: Row-index consistency validation (skipped when quick=true)
     /// For each table row, verifies it has a corresponding entry in each index.
     ValidatingRowIndexConsistency {
         errors: Vec<IntegrityCheckError>,
@@ -9507,6 +9585,21 @@ pub enum IndexCountValidationSubstate {
     ComparingCounts,
 }
 
+/// Sub-states for index rowid position validation
+#[derive(Debug, Clone)]
+pub enum IndexRowidValidationSubstate {
+    /// Starting validation for a new table
+    StartTable,
+    /// Starting validation for a new index
+    StartIndex,
+    /// Rewinding to the beginning of the index
+    Rewinding,
+    /// Checking rowid position in the current index entry
+    CheckingEntry,
+    /// Advancing to the next entry
+    Advancing,
+}
+
 /// Sub-states for NOT NULL constraint validation
 #[derive(Debug, Clone)]
 pub enum NotNullValidationSubstate {
@@ -9693,7 +9786,25 @@ pub fn op_integrity_check(
                     }
                 }
 
-                // Phase 2: Index count validation (skip for quick_check per SQLite)
+                // Phase 2: Index rowid position validation (skip for quick_check per SQLite)
+                if !*quick
+                    && tables
+                        .iter()
+                        .any(|t| t.indexes.iter().any(|idx| idx.index_info.has_rowid))
+                {
+                    let errors = std::mem::take(errors);
+                    state.op_integrity_check_state =
+                        OpIntegrityCheckState::ValidatingIndexRowidPosition {
+                            errors,
+                            current_table_idx: 0,
+                            current_index_idx: 0,
+                            cursor: None,
+                            substate: IndexRowidValidationSubstate::StartTable,
+                        };
+                    return Ok(InsnFunctionStepResult::Step);
+                }
+
+                // Phase 3: Index count validation (skip for quick_check per SQLite)
                 if !*quick && tables.iter().any(|t| !t.indexes.is_empty()) {
                     // Transition to index count validation phase
                     let errors = std::mem::take(errors);
@@ -9709,7 +9820,7 @@ pub fn op_integrity_check(
                     return Ok(InsnFunctionStepResult::Step);
                 }
 
-                // Phase 3: NOT NULL constraint validation (skip for quick_check per SQLite)
+                // Phase 4: NOT NULL constraint validation (skip for quick_check per SQLite)
                 if !*quick && tables.iter().any(|t| !t.not_null_columns.is_empty()) {
                     // Transition to NOT NULL validation phase
                     let errors = std::mem::take(errors);
@@ -9722,7 +9833,7 @@ pub fn op_integrity_check(
                     return Ok(InsnFunctionStepResult::Step);
                 }
 
-                // Phase 4: UNIQUE constraint validation (skip if quick_check)
+                // Phase 5: UNIQUE constraint validation (skip if quick_check)
                 if !*quick
                     && tables
                         .iter()
@@ -9747,6 +9858,136 @@ pub fn op_integrity_check(
                 state.pc += 1;
             }
         }
+        OpIntegrityCheckState::ValidatingIndexRowidPosition {
+            errors,
+            current_table_idx,
+            current_index_idx,
+            cursor,
+            substate,
+        } => {
+            loop {
+                match substate {
+                    IndexRowidValidationSubstate::StartTable => {
+                        // Skip tables without indexes that include rowid
+                        while *current_table_idx < tables.len()
+                            && tables[*current_table_idx]
+                                .indexes
+                                .iter()
+                                .all(|idx| !idx.index_info.has_rowid)
+                        {
+                            *current_table_idx += 1;
+                        }
+
+                        if *current_table_idx >= tables.len() {
+                            // Done scanning rowid positions; transition to index count validation
+                            let errors = std::mem::take(errors);
+                            state.op_integrity_check_state =
+                                OpIntegrityCheckState::ValidatingIndexCounts {
+                                    errors,
+                                    current_table_idx: 0,
+                                    current_index_idx: 0,
+                                    table_row_count: 0,
+                                    index_entry_count: 0,
+                                    substate: IndexCountValidationSubstate::StartTable,
+                                    cursor: None,
+                                };
+                            return Ok(InsnFunctionStepResult::Step);
+                        }
+
+                        *current_index_idx = 0;
+                        *substate = IndexRowidValidationSubstate::StartIndex;
+                    }
+                    IndexRowidValidationSubstate::StartIndex => {
+                        let table = &tables[*current_table_idx];
+
+                        while *current_index_idx < table.indexes.len()
+                            && !table.indexes[*current_index_idx].index_info.has_rowid
+                        {
+                            *current_index_idx += 1;
+                        }
+
+                        if *current_index_idx >= table.indexes.len() {
+                            // Done with this table, move to next
+                            *current_table_idx += 1;
+                            *substate = IndexRowidValidationSubstate::StartTable;
+                            continue;
+                        }
+
+                        let index = &table.indexes[*current_index_idx];
+                        *cursor = Some(Box::new(BTreeCursor::new(
+                            pager.clone(),
+                            index.root_page,
+                            0,
+                        )));
+                        *substate = IndexRowidValidationSubstate::Rewinding;
+                    }
+                    IndexRowidValidationSubstate::Rewinding => {
+                        let btree_cursor = cursor.as_mut().unwrap();
+                        return_if_io!(btree_cursor.rewind());
+                        if !btree_cursor.has_record() {
+                            // Index is empty, move to next index
+                            *cursor = None;
+                            *current_index_idx += 1;
+                            *substate = IndexRowidValidationSubstate::StartIndex;
+                            continue;
+                        }
+                        *substate = IndexRowidValidationSubstate::CheckingEntry;
+                    }
+                    IndexRowidValidationSubstate::CheckingEntry => {
+                        let btree_cursor = cursor.as_mut().unwrap();
+                        let record = return_if_io!(btree_cursor.record())
+                            .expect("record must exist when has_record() is true");
+                        let index = &tables[*current_table_idx].indexes[*current_index_idx];
+                        match index_record_rowid_at_end(record, &index.index_info) {
+                            Ok(true) => {}
+                            Ok(false) => {
+                                errors.push(IntegrityCheckError::RowidNotAtEndOfRecord {
+                                    index_name: index.name.clone(),
+                                });
+                            }
+                            Err(err) => {
+                                errors.push(IntegrityCheckError::IndexRecordCorrupt {
+                                    index_name: index.name.clone(),
+                                    error: err.to_string(),
+                                });
+
+                                if errors.len() >= *max_errors {
+                                    let message = format_integrity_check_result(errors);
+                                    state.registers[*message_register] =
+                                        Register::Value(Value::build_text(message));
+                                    state.op_integrity_check_state = OpIntegrityCheckState::Start;
+                                    state.pc += 1;
+                                    return Ok(InsnFunctionStepResult::Step);
+                                }
+                            }
+                        }
+
+                        if errors.len() >= *max_errors {
+                            let message = format_integrity_check_result(errors);
+                            state.registers[*message_register] =
+                                Register::Value(Value::build_text(message));
+                            state.op_integrity_check_state = OpIntegrityCheckState::Start;
+                            state.pc += 1;
+                            return Ok(InsnFunctionStepResult::Step);
+                        }
+
+                        *substate = IndexRowidValidationSubstate::Advancing;
+                    }
+                    IndexRowidValidationSubstate::Advancing => {
+                        let btree_cursor = cursor.as_mut().unwrap();
+                        return_if_io!(btree_cursor.next());
+                        if btree_cursor.has_record() {
+                            *substate = IndexRowidValidationSubstate::CheckingEntry;
+                        } else {
+                            // Done with this index, move to next
+                            *cursor = None;
+                            *current_index_idx += 1;
+                            *substate = IndexRowidValidationSubstate::StartIndex;
+                        }
+                    }
+                }
+            }
+        }
         OpIntegrityCheckState::ValidatingIndexCounts {
             errors,
             current_table_idx,
@@ -9770,7 +10011,7 @@ pub fn op_integrity_check(
                             // Done with all tables, transition to NOT NULL validation, UNIQUE validation,
                             // row-index validation, or finish
 
-                            // Phase 3: NOT NULL constraint validation (skip for quick_check)
+                            // Phase 4: NOT NULL constraint validation (skip for quick_check)
                             if !*quick && tables.iter().any(|t| !t.not_null_columns.is_empty()) {
                                 let errors = std::mem::take(errors);
                                 state.op_integrity_check_state =
@@ -9783,7 +10024,7 @@ pub fn op_integrity_check(
                                 return Ok(InsnFunctionStepResult::Step);
                             }
 
-                            // Phase 4: UNIQUE constraint validation (skip for quick_check)
+                            // Phase 5: UNIQUE constraint validation (skip for quick_check)
                             if !*quick
                                 && tables
                                     .iter()
@@ -9802,7 +10043,7 @@ pub fn op_integrity_check(
                                 return Ok(InsnFunctionStepResult::Step);
                             }
 
-                            // Phase 5: Row-index consistency validation (skip if quick_check)
+                            // Phase 6: Row-index consistency validation (skip if quick_check)
                             if !*quick && tables.iter().any(|t| !t.indexes.is_empty()) {
                                 let errors = std::mem::take(errors);
                                 state.op_integrity_check_state =
@@ -9917,7 +10158,7 @@ pub fn op_integrity_check(
                         if *current_table_idx >= tables.len() {
                             // Done with all tables, transition to UNIQUE validation, row-index validation, or finish
 
-                            // Phase 4: UNIQUE constraint validation (skip for quick_check)
+                            // Phase 5: UNIQUE constraint validation (skip for quick_check)
                             // Note: We can only be in ValidatingNotNull if quick=false, but check anyway for clarity
                             if !*quick
                                 && tables
@@ -9937,7 +10178,7 @@ pub fn op_integrity_check(
                                 return Ok(InsnFunctionStepResult::Step);
                             }
 
-                            // Phase 5: Row-index consistency validation (skip if quick_check)
+                            // Phase 6: Row-index consistency validation (skip if quick_check)
                             if !*quick && tables.iter().any(|t| !t.indexes.is_empty()) {
                                 let errors = std::mem::take(errors);
                                 state.op_integrity_check_state =