follow up on the fix of multiply with overflow

use usize in bitpacker to enable larger columns in the columnar store

Godbolt comparison with u32 vs u64 for get access: https://godbolt.org/z/cjf7nenYP

Add a mini-tool to inspect columnar files created by tantivy. (very basic functionality which can be extended later)

bitpacker/src/bitpacker.rs

@@ -65,7 +65,7 @@ impl BitPacker {
 
 #[derive(Clone, Debug, Default, Copy)]
 pub struct BitUnpacker {
-    num_bits: u32,
+    num_bits: usize,
     mask: u64,
 }
 
@@ -83,7 +83,7 @@ impl BitUnpacker {
             (1u64 << num_bits) - 1u64
         };
         BitUnpacker {
-            num_bits: u32::from(num_bits),
+            num_bits: usize::from(num_bits),
             mask,
         }
     }
@@ -94,14 +94,14 @@ impl BitUnpacker {
 
     #[inline]
     pub fn get(&self, idx: u32, data: &[u8]) -> u64 {
-        let addr_in_bits = idx * self.num_bits;
-        let addr = (addr_in_bits >> 3) as usize;
+        let addr_in_bits = idx as usize * self.num_bits;
+        let addr = addr_in_bits >> 3;
         if addr + 8 > data.len() {
             if self.num_bits == 0 {
                 return 0;
             }
             let bit_shift = addr_in_bits & 7;
-            return self.get_slow_path(addr, bit_shift, data);
+            return self.get_slow_path(addr, bit_shift as u32, data);
         }
         let bit_shift = addr_in_bits & 7;
         let bytes: [u8; 8] = (&data[addr..addr + 8]).try_into().unwrap();
@@ -129,24 +129,25 @@ impl BitUnpacker {
     //
     // This methods panics if `num_bits` is > 32.
     fn get_batch_u32s(&self, start_idx: u32, data: &[u8], output: &mut [u32]) {
+        let start_idx = start_idx as usize;
         assert!(
             self.bit_width() <= 32,
             "Bitwidth must be <= 32 to use this method."
         );
 
-        let end_idx = start_idx + output.len() as u32;
+        let end_idx = start_idx + output.len();
 
         let end_bit_read = end_idx * self.num_bits;
         let end_byte_read = (end_bit_read + 7) / 8;
         assert!(
-            end_byte_read as usize <= data.len(),
+            end_byte_read <= data.len(),
             "Requested index is out of bounds."
         );
 
         // Simple slow implementation of get_batch_u32s, to deal with our ramps.
-        let get_batch_ramp = |start_idx: u32, output: &mut [u32]| {
+        let get_batch_ramp = |start_idx: usize, output: &mut [u32]| {
             for (out, idx) in output.iter_mut().zip(start_idx..) {
-                *out = self.get(idx, data) as u32;
+                *out = self.get(idx as u32, data) as u32;
             }
         };
 
@@ -161,23 +162,23 @@ impl BitUnpacker {
         // so highway start is the closest multiple of 8 that is >= start_idx.
         let entrance_ramp_len = 8 - (start_idx % 8) % 8;
 
-        let highway_start: u32 = start_idx + entrance_ramp_len;
+        let highway_start: usize = start_idx + entrance_ramp_len;
 
-        if highway_start + BitPacker1x::BLOCK_LEN as u32 > end_idx {
+        if highway_start + BitPacker1x::BLOCK_LEN > end_idx {
             // We don't have enough values to have even a single block of highway.
             // Let's just supply the values the simple way.
             get_batch_ramp(start_idx, output);
             return;
         }
 
-        let num_blocks: u32 = (end_idx - highway_start) / BitPacker1x::BLOCK_LEN as u32;
+        let num_blocks: usize = (end_idx - highway_start) / BitPacker1x::BLOCK_LEN;
 
         // Entrance ramp
-        get_batch_ramp(start_idx, &mut output[..entrance_ramp_len as usize]);
+        get_batch_ramp(start_idx, &mut output[..entrance_ramp_len]);
 
         // Highway
-        let mut offset = (highway_start * self.num_bits) as usize / 8;
-        let mut output_cursor = (highway_start - start_idx) as usize;
+        let mut offset = (highway_start * self.num_bits) / 8;
+        let mut output_cursor = highway_start - start_idx;
         for _ in 0..num_blocks {
             offset += BitPacker1x.decompress(
                 &data[offset..],
@@ -188,7 +189,7 @@ impl BitUnpacker {
         }
 
         // Exit ramp
-        let highway_end = highway_start + num_blocks * BitPacker1x::BLOCK_LEN as u32;
+        let highway_end = highway_start + num_blocks * BitPacker1x::BLOCK_LEN;
         get_batch_ramp(highway_end, &mut output[output_cursor..]);
     }
 

columnar/columnar-cli-inspect/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "tantivy-columnar-inspect"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+
+[dependencies]
+tantivy = {path="../..", package="tantivy"}
+columnar = {path="../", package="tantivy-columnar"}
+common = {path="../../common", package="tantivy-common"}
+
+[workspace]
+members = []
+
+[profile.release]
+debug = true
+#debug-assertions = true
+#overflow-checks = true

columnar/columnar-cli-inspect/src/main.rs

@@ -0,0 +1,54 @@
+use columnar::ColumnarReader;
+use common::file_slice::{FileSlice, WrapFile};
+use std::io;
+use std::path::Path;
+use tantivy::directory::footer::Footer;
+
+fn main() -> io::Result<()> {
+    println!("Opens a columnar file written by tantivy and validates it.");
+    let path = std::env::args().nth(1).unwrap();
+
+    let path = Path::new(&path);
+    println!("Reading {:?}", path);
+    let _reader = open_and_validate_columnar(path.to_str().unwrap())?;
+
+    Ok(())
+}
+
+pub fn validate_columnar_reader(reader: &ColumnarReader) {
+    let num_rows = reader.num_rows();
+    println!("num_rows: {}", num_rows);
+    let columns = reader.list_columns().unwrap();
+    println!("num columns: {:?}", columns.len());
+    for (col_name, dynamic_column_handle) in columns {
+        let col = dynamic_column_handle.open().unwrap();
+        match col {
+            columnar::DynamicColumn::Bool(_)
+            | columnar::DynamicColumn::I64(_)
+            | columnar::DynamicColumn::U64(_)
+            | columnar::DynamicColumn::F64(_)
+            | columnar::DynamicColumn::IpAddr(_)
+            | columnar::DynamicColumn::DateTime(_)
+            | columnar::DynamicColumn::Bytes(_) => {}
+            columnar::DynamicColumn::Str(str_column) => {
+                let num_vals = str_column.ords().values.num_vals();
+                let num_terms_dict = str_column.num_terms() as u64;
+                let max_ord = str_column.ords().values.iter().max().unwrap_or_default();
+                println!("{col_name:35}  num_vals {num_vals:10} \t num_terms_dict {num_terms_dict:8} max_ord: {max_ord:8}",);
+                for ord in str_column.ords().values.iter() {
+                    assert!(ord < num_terms_dict);
+                }
+            }
+        }
+    }
+}
+
+/// Opens a columnar file that was written by tantivy and validates it.
+pub fn open_and_validate_columnar(path: &str) -> io::Result<ColumnarReader> {
+    let wrap_file = WrapFile::new(std::fs::File::open(path)?)?;
+    let slice = FileSlice::new(std::sync::Arc::new(wrap_file));
+    let (_footer, slice) = Footer::extract_footer(slice.clone()).unwrap();
+    let reader = ColumnarReader::open(slice).unwrap();
+    validate_columnar_reader(&reader);
+    Ok(reader)
+}

common/src/file_slice.rs

@@ -1,5 +1,6 @@
 use std::fs::File;
 use std::ops::{Deref, Range, RangeBounds};
+use std::path::Path;
 use std::sync::Arc;
 use std::{fmt, io};
 
@@ -177,6 +178,12 @@ fn combine_ranges<R: RangeBounds<usize>>(orig_range: Range<usize>, rel_range: R)
 }
 
 impl FileSlice {
+    /// Creates a FileSlice from a path.
+    pub fn open(path: &Path) -> io::Result<FileSlice> {
+        let wrap_file = WrapFile::new(File::open(path)?)?;
+        Ok(FileSlice::new(Arc::new(wrap_file)))
+    }
+
     /// Wraps a FileHandle.
     pub fn new(file_handle: Arc<dyn FileHandle>) -> Self {
         let num_bytes = file_handle.len();

src/directory/footer.rs

@@ -1,3 +1,9 @@
+//! The footer is a small metadata structure that is appended at the end of every file.
+//!
+//! The footer is used to store a checksum of the file content.
+//! The footer also stores the version of the index format.
+//! This version is used to detect incompatibility between the index and the library version.
+
 use std::io;
 use std::io::Write;
 
@@ -20,20 +26,22 @@ type CrcHashU32 = u32;
 /// A Footer is appended to every file
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Footer {
+    /// The version of the index format
     pub version: Version,
+    /// The crc32 hash of the body
     pub crc: CrcHashU32,
 }
 
 impl Footer {
-    pub fn new(crc: CrcHashU32) -> Self {
+    pub(crate) fn new(crc: CrcHashU32) -> Self {
         let version = crate::VERSION.clone();
         Footer { version, crc }
     }
 
-    pub fn crc(&self) -> CrcHashU32 {
+    pub(crate) fn crc(&self) -> CrcHashU32 {
         self.crc
     }
-    pub fn append_footer<W: io::Write>(&self, mut write: &mut W) -> io::Result<()> {
+    pub(crate) fn append_footer<W: io::Write>(&self, mut write: &mut W) -> io::Result<()> {
         let mut counting_write = CountingWriter::wrap(&mut write);
         counting_write.write_all(serde_json::to_string(&self)?.as_ref())?;
         let footer_payload_len = counting_write.written_bytes();
@@ -42,6 +50,7 @@ impl Footer {
         Ok(())
     }
 
+    /// Extracts the tantivy Footer from the file and returns the footer and the rest of the file
     pub fn extract_footer(file: FileSlice) -> io::Result<(Footer, FileSlice)> {
         if file.len() < 4 {
             return Err(io::Error::new(

src/directory/mod.rs

@@ -6,7 +6,7 @@ mod mmap_directory;
 mod directory;
 mod directory_lock;
 mod file_watcher;
-mod footer;
+pub mod footer;
 mod managed_directory;
 mod ram_directory;
 mod watch_event_router;

src/indexer/segment_updater.rs

@@ -287,6 +287,17 @@ impl SegmentUpdater {
         let merge_thread_pool = ThreadPoolBuilder::new()
             .thread_name(|i| format!("merge_thread_{i}"))
             .num_threads(num_merge_threads)
+            .panic_handler(move |panic| {
+                let message = if let Some(msg) = panic.downcast_ref::<&str>() {
+                   *msg
+                } else if let Some(msg) = panic.downcast_ref::<String>() {
+                    msg.as_str()
+                } else {
+                    "UNKNOWN"
+                };
+                eprintln!("merge thread panicked with: {message}")
+
+            })
             .build()
             .map_err(|_| {
                 crate::TantivyError::SystemError(