add nkeycol to index, use for withoutrowid tables

Author: Pavan-Nambi

core/incremental/operator.rs

@@ -72,6 +72,7 @@ pub fn create_dbsp_state_index(root_page: i64) -> Index {
         where_clause: None,
         index_method: None,
         is_primary_key: false,
+         n_key_col: 3,
     }
 }
 

core/schema.rs

@@ -2581,6 +2581,7 @@ pub struct Index {
     pub where_clause: Option<Box<Expr>>,
     pub index_method: Option<Arc<dyn IndexMethodAttachment>>,
     pub is_primary_key: bool,
+     pub n_key_col: usize,
 }
 
 #[allow(dead_code)]
@@ -2649,6 +2650,7 @@ impl Index {
                         where_clause: None,
                         is_primary_key: false,
                         index_method: Some(descriptor),
+                         n_key_col: columns.len(),
                     })
                 } else {
                     Ok(Index {
@@ -2662,6 +2664,7 @@ impl Index {
                         where_clause,
                         index_method: None,
                         is_primary_key: false,
+                         n_key_col: columns.len(),
                     })
                 }
             }
@@ -2738,6 +2741,7 @@ impl Index {
             where_clause: None,
             index_method: None,
             is_primary_key: true,
+            n_key_col: column_count,
         })
     }
 
@@ -2766,6 +2770,7 @@ impl Index {
             })
             .collect::<Vec<_>>();
 
+             let n_key_col = unique_cols.len();
         Ok(Index {
             name: normalize_ident(index_name.as_str()),
             table_name: table.name.clone(),
@@ -2777,6 +2782,7 @@ impl Index {
             where_clause: None,
             index_method: None,
             is_primary_key: false,
+              n_key_col, 
         })
     }
 

core/translate/compound_select.rs

@@ -437,7 +437,7 @@ fn create_dedupe_index(
         };
         column.collation = collation;
     }
-
+ let n_key_col = dedupe_columns.len();
     let dedupe_index = Arc::new(Index {
         columns: dedupe_columns,
         name: "compound_dedupe".to_string(),
@@ -449,6 +449,7 @@ fn create_dedupe_index(
         where_clause: None,
         index_method: None,
         is_primary_key: false,
+           n_key_col,
     });
     let cursor_id = program.alloc_cursor_id(CursorType::BTreeIndex(dedupe_index.clone()));
     program.emit_insn(Insn::OpenEphemeral {

core/translate/index.rs

@@ -150,6 +150,7 @@ pub fn translate_create_index(
         where_clause: where_clause.clone(),
         index_method: index_method.clone(),
         is_primary_key: false,
+         n_key_col: columns.len(),
     });
 
     if !idx.validate_where_expr(table) {

core/translate/insert.rs

@@ -2764,9 +2764,10 @@ fn translate_without_rowid_insert(
         })?;
 
     tracing::debug!(
-        "Found primary key index '{}' with root page {}",
+        "Found WITHOUT ROWID primary key index '{}' with root page {} and {} key columns.",
         pk_index.name,
-        pk_index.root_page
+        pk_index.root_page,
+        pk_index.n_key_col
     );
 
     let pk_cursor_id = program.alloc_cursor_index(None, pk_index)?;
@@ -2806,16 +2807,15 @@ fn translate_without_rowid_insert(
 
     emit_notnulls(&mut program, &ctx, &insertion);
 
-    let pk_columns = &pk_index.columns;
-    let num_pk_cols = pk_columns.len();
+    let num_pk_cols = pk_index.n_key_col;
     let pk_registers_start = program.alloc_registers(num_pk_cols);
     tracing::debug!(
         "Allocated registers {}-{} for PK uniqueness check",
         pk_registers_start,
         pk_registers_start + num_pk_cols - 1
     );
 
-    for (i, pk_col) in pk_columns.iter().enumerate() {
+    for (i, pk_col) in pk_index.columns.iter().take(num_pk_cols).enumerate() {
         let mapping = insertion.get_col_mapping_by_name(&pk_col.name).unwrap();
         program.emit_insn(Insn::Copy {
             src_reg: mapping.register,
@@ -2878,13 +2878,7 @@ fn translate_without_rowid_insert(
 
     tracing::debug!("Beginning update of secondary indices.");
     for index in resolver.schema.get_indices(table.name.as_str()) {
-        let is_pk = index.columns.len() == pk_index.columns.len()
-            && index
-                .columns
-                .iter()
-                .zip(pk_index.columns.iter())
-                .all(|(a, b)| a.name == b.name);
-        if is_pk {
+        if index.is_primary_key {
             continue;
         }
 
@@ -2897,13 +2891,12 @@ fn translate_without_rowid_insert(
 
         tracing::debug!("Processing secondary index '{}'", index.name);
 
-        let secondary_key_cols = &index.columns;
-        let num_key_cols = secondary_key_cols.len();
+
+        let num_key_cols = index.n_key_col;
         let key_start_reg = program.alloc_registers(num_key_cols);
 
         let mut current_reg = key_start_reg;
-
-        for idx_col in secondary_key_cols {
+        for idx_col in index.columns.iter().take(num_key_cols) {
             let mapping = insertion.get_col_mapping_by_name(&idx_col.name).unwrap();
             program.emit_insn(Insn::Copy {
                 src_reg: mapping.register,

core/translate/main_loop.rs

@@ -105,6 +105,7 @@ pub fn init_distinct(program: &mut ProgramBuilder, plan: &SelectPlan) -> Result<
         where_clause: None,
         index_method: None,
         is_primary_key: false,
+        n_key_col: plan.result_columns.len(),
     });
     let cursor_id = program.alloc_cursor_id(CursorType::BTreeIndex(index.clone()));
     let ctx = DistinctCtx {
@@ -184,6 +185,8 @@ pub fn init_loop(
             where_clause: None,
             index_method: None,
             is_primary_key: false,
+              n_key_col: group_by.as_ref().unwrap().exprs.len(),
+
         });
         let cursor_id = program.alloc_cursor_id(CursorType::BTreeIndex(index.clone()));
         if group_by.is_none() {

core/translate/optimizer/mod.rs

@@ -1278,6 +1278,7 @@ fn ephemeral_index_build(
             (None, None) => Ordering::Equal,
         }
     });
+     let n_key_col = ephemeral_columns.len();
     let ephemeral_index = Index {
         name: format!(
             "ephemeral_{}_{}",
@@ -1296,6 +1297,7 @@ fn ephemeral_index_build(
             .is_some_and(|btree| btree.has_rowid),
         index_method: None,
         is_primary_key: false,
+         n_key_col,
     };
 
     ephemeral_index

core/translate/order_by.rs

@@ -100,6 +100,7 @@ pub fn init_order_by(
                 default: None,
             })
         }
+           let n_key_col = index_columns.len();
         let index = Arc::new(Index {
             name: index_name.clone(),
             table_name: String::new(),
@@ -111,6 +112,7 @@ pub fn init_order_by(
             where_clause: None,
             index_method: None,
             is_primary_key: false,
+              n_key_col,
         });
         program.alloc_cursor_id(CursorType::BTreeIndex(index))
     } else {

core/translate/subquery.rs

@@ -382,6 +382,7 @@ fn get_subquery_parser<'a>(
                     )?;
                 }
 
+                let  n_key_col=columns.len();
                 let ephemeral_index = Arc::new(Index {
                     columns,
                     name: format!("ephemeral_index_where_sub_{subquery_id}"),
@@ -393,6 +394,7 @@ fn get_subquery_parser<'a>(
                     where_clause: None,
                     index_method: None,
                     is_primary_key: false,
+                     n_key_col,
                 });
 
                 let cursor_id =