Handle integer conversions explicitly

Cargo.toml

@@ -26,3 +26,7 @@ unimplemented = "deny"
 unreachable = "deny"
 unwrap_in_result = "deny"
 unwrap_used = "deny"
+
+cast_possible_truncation = "deny"
+cast_possible_wrap = "deny"
+cast_sign_loss = "deny"

src/core/error/internal.rs

@@ -54,6 +54,8 @@ pub enum InvariantViolation {
     EmptyLeafSplit,
     #[error("leaf split target key is missing")]
     LeafSplitTargetMissing,
+    #[error("leaf split target slot {slot_index} exceeds the slot-id range")]
+    LeafSplitTargetSlotOutOfRange { slot_index: usize },
     #[error("page {page_id} pin count overflowed")]
     PagePinCountOverflow { page_id: PageId },
 }

src/core/mod.rs

@@ -29,4 +29,5 @@ pub use types::{
 };
 pub(crate) use types::{Lsn, SlotId, TxnId};
 
-pub(crate) const PAGE_SIZE: usize = 4096;
+pub(crate) const PAGE_SIZE_U16: u16 = 4096;
+pub(crate) const PAGE_SIZE: usize = PAGE_SIZE_U16 as usize;

src/executor/expression.rs

@@ -25,8 +25,10 @@ pub fn evaluate_expression(
 /// Each values row is evaluated against an empty synthetic record. The row's
 /// position in the `VALUES` list becomes the result table key.
 pub(super) fn execute_values(rows: Vec<Vec<PlannedExpression>>) -> ExecutorResult<ExecutionOutput> {
-    let rows = rows.into_iter().enumerate().map(|(table_key, expressions)| {
-        let input = empty_record(table_key as TableKey)?;
+    let rows = rows.into_iter().enumerate().map(|(row_index, expressions)| {
+        let table_key = TableKey::try_from(row_index)
+            .map_err(|_| ExecutorError::ValuesRowIndexOutOfRange { row_index })?;
+        let input = empty_record(table_key)?;
         evaluate_expressions(&expressions, &input)
     });
     Ok(ExecutionOutput::Rows { rows: Box::new(rows) })

src/executor/mod.rs

@@ -102,6 +102,12 @@ pub enum ExecutorError {
         /// Operand value rejected by the operator.
         value: Value,
     },
+    /// A synthetic row index for a `VALUES` result exceeded the table-key range.
+    #[error("VALUES row index {row_index} does not fit in a table key")]
+    ValuesRowIndexOutOfRange {
+        /// Zero-based position of the row in the `VALUES` list.
+        row_index: usize,
+    },
     /// Integer arithmetic overflowed.
     #[error("integer overflow while evaluating operator {op}")]
     IntegerOverflow {

src/planner/planning.rs

@@ -738,7 +738,7 @@ fn exact_single_column_index(
 ) -> bool {
     index.table_id == table.table_id
         && index.columns.len() == 1
-        && index.columns[0].source_column_ordinal as usize == column.ordinal
+        && index.columns[0].source_column_ordinal == column.ordinal
 }
 
 fn value_matches_data_type(value: &Value, data_type: DataType) -> bool {

src/relational/catalog.rs

@@ -142,7 +142,7 @@ pub struct TableSchema {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct IndexColumnSchema {
     /// Ordinal of the source column in the indexed table schema.
-    pub source_column_ordinal: u64,
+    pub source_column_ordinal: usize,
     /// Column metadata copied into the index key schema.
     pub column: ColumnSchema,
 }
@@ -384,10 +384,7 @@ impl IndexSchema {
                     table: table.name.clone(),
                     column: (*column_name).to_owned(),
                 })?;
-            columns.push(IndexColumnSchema {
-                source_column_ordinal: source_column_ordinal as u64,
-                column: column.clone(),
-            });
+            columns.push(IndexColumnSchema { source_column_ordinal, column: column.clone() });
         }
 
         Ok(Self {

src/relational/catalog_manager.rs

@@ -115,10 +115,8 @@ impl CatalogManager {
                         column: (*column_name).to_owned(),
                     })
                 })?;
-            index_columns.push(IndexColumnSchema {
-                source_column_ordinal: source_column_ordinal as u64,
-                column: source_column.clone(),
-            });
+            index_columns
+                .push(IndexColumnSchema { source_column_ordinal, column: source_column.clone() });
             catalog_columns.push(ColumnCatalogRow {
                 column_id,
                 object_kind: CatalogObjectKind::Index,
@@ -276,7 +274,7 @@ impl CatalogManager {
             })
             .collect();
 
-        Ok(index_schema_from_rows(index, columns))
+        index_schema_from_rows(index, columns)
     }
 
     pub(crate) fn index_schemas_for_table(
@@ -302,7 +300,7 @@ impl CatalogManager {
                     })
                     .cloned()
                     .collect();
-                Ok(index_schema_from_rows(index, columns))
+                index_schema_from_rows(index, columns)
             })
             .collect()
     }
@@ -465,23 +463,26 @@ fn column_schema_from_row(row: ColumnCatalogRow) -> ColumnSchema {
 fn index_schema_from_rows(
     index: IndexCatalogRow,
     mut columns: Vec<ColumnCatalogRow>,
-) -> IndexSchema {
+) -> StorageResult<IndexSchema> {
     columns.sort_by_key(|row| row.ordinal);
+    let columns = columns
+        .into_iter()
+        .map(|row| {
+            let source_column_ordinal = row.source_column_ordinal.unwrap_or(row.ordinal);
+            let source_column_ordinal = usize::try_from(source_column_ordinal)
+                .map_err(|error| invalid_catalog_row("sys_columns", error))?;
+            Ok(IndexColumnSchema { source_column_ordinal, column: column_schema_from_row(row) })
+        })
+        .collect::<StorageResult<_>>()?;
 
-    IndexSchema {
+    Ok(IndexSchema {
         index_id: index.index_id,
         name: index.name,
         table_id: index.table_id,
         root_page_id: index.root_page_id,
         unique: index.unique,
-        columns: columns
-            .into_iter()
-            .map(|row| IndexColumnSchema {
-                source_column_ordinal: row.source_column_ordinal.unwrap_or(row.ordinal),
-                column: column_schema_from_row(row),
-            })
-            .collect(),
-    }
+        columns,
+    })
 }
 
 #[cfg(test)]
@@ -608,7 +609,7 @@ mod tests {
         let mut tables = manager.table_cursor(SYS_TABLES_ROOT_PAGE_ID);
         assert_table_catalog_row(&mut tables, table.table_id, "users", table.root_page_id);
 
-        let first_user_column_id = system_column_rows().len() as CatalogId + 1;
+        let first_user_column_id = CatalogId::try_from(system_column_rows().len()).unwrap() + 1;
         let mut columns = manager.table_cursor(SYS_COLUMNS_ROOT_PAGE_ID);
         assert_column_catalog_row(
             &mut columns,
@@ -706,8 +707,9 @@ mod tests {
             },
         );
 
-        let index_column_id =
-            system_column_rows().len() as CatalogId + table.row.columns.len() as CatalogId + 1;
+        let index_column_id = CatalogId::try_from(system_column_rows().len()).unwrap()
+            + CatalogId::try_from(table.row.columns.len()).unwrap()
+            + 1;
         let mut columns = manager.table_cursor(SYS_COLUMNS_ROOT_PAGE_ID);
         assert_column_catalog_row(
             &mut columns,

src/relational/index_manager.rs

@@ -101,7 +101,7 @@ fn index_key_from_record_bytes(
     let mut values = Vec::with_capacity(index.columns.len());
 
     for column in &index.columns {
-        let ordinal = column.source_column_ordinal as usize;
+        let ordinal = column.source_column_ordinal;
         let value = tuple.values().nth(ordinal).ok_or_else(|| {
             invalid_table_record(
                 table,

src/relational/tuple.rs

@@ -53,9 +53,9 @@ const TAG_UNSIGNED_INTEGER: u8 = 0x06;
 
 const NULL_LENGTH: u32 = 0;
 const BOOL_LENGTH: u32 = 1;
-const I32_LENGTH: u32 = size_of::<i32>() as u32;
-const F32_LENGTH: u32 = size_of::<f32>() as u32;
-const U64_LENGTH: u32 = size_of::<u64>() as u32;
+const I32_LENGTH: u32 = i32::BITS / 8;
+const F32_LENGTH: u32 = 4;
+const U64_LENGTH: u32 = u64::BITS / 8;
 
 /// A single typed value stored in a [`Tuple`].
 #[derive(Debug, Clone, PartialEq)]
@@ -539,13 +539,13 @@ fn read_value<R: Read>(reader: &mut R, tag: u8, len: u32) -> io::Result<Value> {
 
 fn validate_value_payload(tag: u8, payload: &[u8]) -> io::Result<()> {
     match tag {
-        TAG_NULL => validate_len(tag, payload.len() as u32, NULL_LENGTH),
+        TAG_NULL => validate_payload_len(tag, payload.len(), 0),
         TAG_STRING => {
             std::str::from_utf8(payload).map_err(invalid_data)?;
             Ok(())
         }
         TAG_BOOLEAN => {
-            validate_len(tag, payload.len() as u32, BOOL_LENGTH)?;
+            validate_payload_len(tag, payload.len(), 1)?;
             match payload[0] {
                 0 | 1 => Ok(()),
                 actual => Err(io::Error::new(
@@ -554,20 +554,31 @@ fn validate_value_payload(tag: u8, payload: &[u8]) -> io::Result<()> {
                 )),
             }
         }
-        TAG_INTEGER => validate_len(tag, payload.len() as u32, I32_LENGTH),
+        TAG_INTEGER => validate_payload_len(tag, payload.len(), size_of::<i32>()),
         TAG_FLOAT => {
-            validate_len(tag, payload.len() as u32, F32_LENGTH)?;
+            validate_payload_len(tag, payload.len(), size_of::<f32>())?;
             let bytes = payload.try_into().map_err(invalid_data)?;
             validate_float(decode_ordered_f32(bytes))
         }
-        TAG_UNSIGNED_INTEGER => validate_len(tag, payload.len() as u32, U64_LENGTH),
+        TAG_UNSIGNED_INTEGER => validate_payload_len(tag, payload.len(), size_of::<u64>()),
         actual => Err(io::Error::new(
             io::ErrorKind::InvalidData,
             format!("unknown tuple value tag: {actual}"),
         )),
     }
 }
 
+fn validate_payload_len(tag: u8, actual: usize, expected: usize) -> io::Result<()> {
+    if actual == expected {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("invalid length {actual} for tuple value tag {tag}; expected {expected}"),
+        ))
+    }
+}
+
 fn validate_len(tag: u8, actual: u32, expected: u32) -> io::Result<()> {
     if actual == expected {
         Ok(())
@@ -625,11 +636,11 @@ fn read_value_len_from_slice(bytes: &[u8], offset: usize) -> io::Result<(u32, us
 }
 
 fn encode_ordered_i32(value: i32) -> [u8; size_of::<i32>()] {
-    ((value as u32) ^ 0x8000_0000).to_be_bytes()
+    (value.cast_unsigned() ^ 0x8000_0000).to_be_bytes()
 }
 
 fn decode_ordered_i32(bytes: [u8; size_of::<i32>()]) -> i32 {
-    (u32::from_be_bytes(bytes) ^ 0x8000_0000) as i32
+    (u32::from_be_bytes(bytes) ^ 0x8000_0000).cast_signed()
 }
 
 fn encode_ordered_f32(value: f32) -> io::Result<[u8; size_of::<f32>()]> {

src/storage/btree/rebalance.rs

@@ -144,15 +144,16 @@ impl TreeCursor {
         let page = pin.read()?;
         let interior = page.open::<Interior>()?;
         let slot_count = interior.slot_count();
-        if child_index > slot_count as usize {
-            let slot_index = child_index.try_into().unwrap_or(u16::MAX);
-            return Err(PageError::InvalidSlotIndex { slot_index, slot_count }.into());
-        }
-        if child_index == slot_count as usize {
+        if child_index == usize::from(slot_count) {
             return Ok(interior.rightmost_child());
         }
+        let slot_index = u16::try_from(child_index)
+            .map_err(|_| PageError::InvalidSlotIndex { slot_index: u16::MAX, slot_count })?;
+        if slot_index > slot_count {
+            return Err(PageError::InvalidSlotIndex { slot_index, slot_count }.into());
+        }
 
-        let (left_child, _, _, _) = interior.cell_payload_parts(child_index as u16)?;
+        let (left_child, _, _, _) = interior.cell_payload_parts(slot_index)?;
         Ok(left_child)
     }
 
@@ -389,12 +390,12 @@ impl TreeCursor {
     ) -> StorageResult<bool> {
         let child_page_ids = self.read_interior_child_page_ids(page_id)?;
         for (slot_index, &child_page_id) in
-            child_page_ids[..child_page_ids.len() - 1].iter().enumerate()
+            (0u16..).zip(&child_page_ids[..child_page_ids.len() - 1])
         {
             let matches = self.with_subtree_max_key(child_page_id, |expected_key| {
                 let expected_key =
                     expected_key.ok_or_else(|| Self::missing_child_max_key_error(page_id))?;
-                self.compare_interior_key(page_id, slot_index as u16, expected_key)
+                self.compare_interior_key(page_id, slot_index, expected_key)
                     .map(|ordering| ordering == Ordering::Equal)
             })?;
             if !matches {

src/storage/btree/rebalance_repair.rs

@@ -112,7 +112,8 @@ impl TreeCursor {
 
         let mut child_index = 0;
         loop {
-            let child_count = self.raw_interior_slot_count(page_id)? as usize + 1;
+            let slot_count = self.raw_interior_slot_count(page_id)?;
+            let child_count = usize::from(slot_count) + 1;
             if child_index >= child_count {
                 break;
             }
@@ -122,7 +123,10 @@ impl TreeCursor {
                 let child_ref = if child_index + 1 == child_count {
                     ChildSlotRef::Rightmost
                 } else {
-                    ChildSlotRef::Slot(child_index as u16)
+                    let slot_index = u16::try_from(child_index).map_err(|_| {
+                        PageError::InvalidSlotIndex { slot_index: u16::MAX, slot_count }
+                    })?;
+                    ChildSlotRef::Slot(slot_index)
                 };
                 let parent_frame = PathFrame { page_id, child_ref };
                 if let Some(parent_pending) = self.insert_into_parent(parent_frame, pending)? {

src/storage/btree/split.rs

@@ -316,7 +316,12 @@ impl TreeCursor {
             .position(|cell| cell.key() == target_key)
             .ok_or(StorageError::Internal(InternalError::InvariantViolation(
                 InvariantViolation::LeafSplitTargetMissing,
-            )))? as u16;
+            )))?;
+        let target_slot_index = u16::try_from(target_slot_index).map_err(|_| {
+            StorageError::Internal(InternalError::InvariantViolation(
+                InvariantViolation::LeafSplitTargetSlotOutOfRange { slot_index: target_slot_index },
+            ))
+        })?;
         self.set_positioned_state(target_page_id, target_slot_index);
 
         Ok(PendingSplit { separator, left_page_id: leaf_page_id, right_page_id })

src/storage/btree/tests.rs

@@ -478,10 +478,10 @@ fn failed_interior_rewrite_leaves_page_unchanged() {
         let page = pin.read().unwrap();
         *page.page()
     };
-    let children: Vec<_> = (0..16)
+    let children: Vec<_> = (0u8..16)
         .map(|index| ChildEntry {
-            page_id: 100 + index,
-            max_key: Some(vec![index as u8; PAGE_SIZE]),
+            page_id: 100 + u64::from(index),
+            max_key: Some(vec![index; PAGE_SIZE]),
         })
         .collect();
 

src/storage/database_header.rs

@@ -1,5 +1,5 @@
 use crate::core::{
-    PAGE_SIZE, PageId,
+    PAGE_SIZE, PAGE_SIZE_U16, PageId,
     error::{CorruptionComponent, CorruptionError, CorruptionKind, StorageError, StorageResult},
 };
 
@@ -17,7 +17,7 @@ impl DatabaseHeader {
         let mut page = [0u8; PAGE_SIZE];
         page[0..8].copy_from_slice(MAGIC);
         page[8..10].copy_from_slice(&FORMAT_VERSION.to_le_bytes());
-        page[10..12].copy_from_slice(&(PAGE_SIZE as u16).to_le_bytes());
+        page[10..12].copy_from_slice(&PAGE_SIZE_U16.to_le_bytes());
         page
     }