Perf improvements for TPC-H ingestion

Cargo.toml

@@ -49,6 +49,7 @@ rand = "0.9.1"
 rayon = "1.10.0"
 regex = "1.12.2"
 roaring = "0.11.2"
+rust_decimal = "1.39.0"
 sqlparser = "0.59.0"
 tempfile = "3.23.0"
 thiserror = "2.0.17"
@@ -57,7 +58,7 @@ rustc-hash = "2.1.1"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 sqllogictest = "0.28.4"
-time = { version = "0.3.44", features = ["parsing"] }
+time = { version = "0.3.44", features = ["formatting", "macros", "parsing"] }
 tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread"] }
 tpchgen = "2.0.1"
 tracing = "0.1.41"

llkv-column-map/README.md

@@ -15,6 +15,10 @@
 - Maintain the column catalog that maps `LogicalFieldId` to the physical keys storing metadata and data pages.
 - Provide gather, scan, and append APIs used by [`llkv-table`](../llkv-table/) and the executor layer.
 
+### Write-Sizing Hints & Loader Tuning
+
+`ColumnStore::write_hints()` exposes the storage layer’s current guidance for chunk sizing, insert batch rows, and variable-width fallbacks. Bulk data loaders should resolve their batch size through this API instead of hard-coding constants, in order to keep memory usage at a minimum and ease write thrashing.
+
 ## Logical vs Physical Keys
 
 - `LogicalFieldId` encodes namespace (user data, row-id shadow, MVCC metadata), table ID, and field ID to avoid collisions across tables.

llkv-column-map/src/store/core.rs

@@ -116,6 +116,11 @@ where
         })
     }
 
+    /// Return heuristics that guide upstream writers when sizing batches.
+    pub fn write_hints(&self) -> ColumnStoreWriteHints {
+        ColumnStoreWriteHints::from_config(&self.cfg)
+    }
+
     /// Creates and persists an index for a column.
     ///
     /// Builds the specified index type for all existing data in the column and

llkv-column-map/src/store/mod.rs

@@ -60,3 +60,6 @@ use dtype_cache::DTypeCache;
 
 pub mod indexing;
 pub use indexing::*;
+
+mod write_hints;
+pub use write_hints::ColumnStoreWriteHints;

llkv-column-map/src/store/write_hints.rs

@@ -0,0 +1,73 @@
+use super::config::ColumnStoreConfig;
+use super::constants::TARGET_CHUNK_BYTES;
+use crate::types::RowId;
+
+/// Heuristics that guide callers when sizing write batches for the column store.
+///
+/// The values are derived from the store's ingest configuration so higher layers can
+/// adapt without duplicating storage-level constants. Callers should treat these
+/// numbers as soft targets: exceeding the maximum batch rows will be clamped, but
+/// smaller batches are always accepted.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ColumnStoreWriteHints {
+    /// Target chunk size used when splitting incoming arrays.
+    pub target_chunk_bytes: usize,
+    /// Preferred number of rows to buffer per insert before flushing.
+    pub recommended_insert_batch_rows: usize,
+    /// Hard ceiling for literal INSERT batches before storage splits them eagerly.
+    pub max_insert_batch_rows: usize,
+    /// Fallback slice size for exotic variable-width arrays lacking offset metadata.
+    pub varwidth_fallback_rows_per_slice: usize,
+}
+
+impl ColumnStoreWriteHints {
+    pub(crate) fn from_config(cfg: &ColumnStoreConfig) -> Self {
+        let row_id_width = std::mem::size_of::<RowId>().max(1);
+        let recommended_rows = (TARGET_CHUNK_BYTES / row_id_width).max(1);
+        let max_rows = recommended_rows.saturating_mul(4).max(recommended_rows);
+        Self {
+            target_chunk_bytes: TARGET_CHUNK_BYTES,
+            recommended_insert_batch_rows: recommended_rows,
+            max_insert_batch_rows: max_rows,
+            varwidth_fallback_rows_per_slice: cfg.varwidth_fallback_rows_per_slice,
+        }
+    }
+
+    /// Clamp a requested batch size to the store's supported envelope.
+    pub fn clamp_insert_batch_rows(&self, requested_rows: usize) -> usize {
+        match requested_rows {
+            0 => 0,
+            _ => requested_rows.min(self.max_insert_batch_rows),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn from_config_uses_target_chunk_bytes() {
+        let cfg = ColumnStoreConfig {
+            varwidth_fallback_rows_per_slice: 2048,
+        };
+        let hints = ColumnStoreWriteHints::from_config(&cfg);
+        let expected_rows = TARGET_CHUNK_BYTES / std::mem::size_of::<RowId>();
+        assert_eq!(hints.target_chunk_bytes, TARGET_CHUNK_BYTES);
+        assert_eq!(hints.varwidth_fallback_rows_per_slice, 2048);
+        assert_eq!(hints.recommended_insert_batch_rows, expected_rows);
+        assert!(hints.max_insert_batch_rows >= hints.recommended_insert_batch_rows);
+    }
+
+    #[test]
+    fn clamp_insert_batch_rows_caps_large_values() {
+        let cfg = ColumnStoreConfig {
+            varwidth_fallback_rows_per_slice: 512,
+        };
+        let hints = ColumnStoreWriteHints::from_config(&cfg);
+        assert_eq!(hints.clamp_insert_batch_rows(0), 0);
+        let max = hints.max_insert_batch_rows;
+        assert_eq!(hints.clamp_insert_batch_rows(max * 10), max);
+        assert_eq!(hints.clamp_insert_batch_rows(max - 1), max - 1);
+    }
+}

llkv-executor/src/lib.rs

@@ -9894,7 +9894,8 @@ where
 
     let transform = match order_plan.sort_type {
         OrderSortType::Native => match column.data_type {
-            DataType::Int64 => ScanOrderTransform::IdentityInteger,
+            DataType::Int64 => ScanOrderTransform::IdentityInt64,
+            DataType::Int32 => ScanOrderTransform::IdentityInt32,
             DataType::Utf8 => ScanOrderTransform::IdentityUtf8,
             ref other => {
                 return Err(Error::InvalidArgumentError(format!(