store/order row acc to pkey

Author: Pavan-Nambi

core/incremental/operator.rs

@@ -72,7 +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,
+        n_key_col: 3,
     }
 }
 

core/schema.rs

@@ -2581,7 +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,
+    pub n_key_col: usize,
 }
 
 #[allow(dead_code)]
@@ -2650,7 +2650,7 @@ impl Index {
                         where_clause: None,
                         is_primary_key: false,
                         index_method: Some(descriptor),
-                         n_key_col: columns.len(),
+                        n_key_col: columns.len(),
                     })
                 } else {
                     Ok(Index {
@@ -2664,7 +2664,7 @@ impl Index {
                         where_clause,
                         index_method: None,
                         is_primary_key: false,
-                         n_key_col: columns.len(),
+                        n_key_col: columns.len(),
                     })
                 }
             }
@@ -2770,7 +2770,7 @@ impl Index {
             })
             .collect::<Vec<_>>();
 
-             let n_key_col = unique_cols.len();
+        let n_key_col = unique_cols.len();
         Ok(Index {
             name: normalize_ident(index_name.as_str()),
             table_name: table.name.clone(),
@@ -2782,7 +2782,7 @@ impl Index {
             where_clause: None,
             index_method: None,
             is_primary_key: false,
-              n_key_col, 
+            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 n_key_col = dedupe_columns.len();
     let dedupe_index = Arc::new(Index {
         columns: dedupe_columns,
         name: "compound_dedupe".to_string(),
@@ -449,7 +449,7 @@ fn create_dedupe_index(
         where_clause: None,
         index_method: None,
         is_primary_key: false,
-           n_key_col,
+        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,7 +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(),
+        n_key_col: columns.len(),
     });
 
     if !idx.validate_where_expr(table) {

core/translate/insert.rs

@@ -2841,32 +2841,82 @@ fn translate_without_rowid_insert(
         description: format_unique_violation_desc(&table.name, pk_index),
     });
 
+
     program.preassign_label_to_next_insn(ok_to_insert_label);
     tracing::debug!("Uniqueness check passed. Continuing with insertion.");
 
+
+    let physically_ordered_cols: Vec<&Column> = {
+        let mut ordered = Vec::with_capacity(table.columns.len());
+        let mut pk_cols_added = std::collections::HashSet::new();
+
+        for pk_col_def in pk_index.columns.iter().take(pk_index.n_key_col) {
+            let (_, col) = table.get_column(&pk_col_def.name).unwrap();
+            ordered.push(col);
+            pk_cols_added.insert(col.name.as_ref().unwrap().as_str());
+            tracing::trace!(
+                "  Physical col {}: {}",
+                ordered.len() - 1,
+                col.name.as_ref().unwrap()
+            );
+        }
+
+        for col in table.columns.iter() {
+            if !pk_cols_added.contains(col.name.as_ref().unwrap().as_str()) {
+                ordered.push(col);
+                tracing::trace!(
+                    "  Physical col {}: {}",
+                    ordered.len() - 1,
+                    col.name.as_ref().unwrap()
+                );
+            }
+        }
+        assert_eq!(
+            ordered.len(),
+            table.columns.len(),
+            "Physical column ordering failed: count mismatch."
+        );
+        ordered
+    };
+
+    let reordered_start_reg = program.alloc_registers(insertion.col_mappings.len());
+    tracing::debug!(
+        "Reordering columns into physical order. New registers start at {}",
+        reordered_start_reg
+    );
+
+    for (i, physical_col) in physically_ordered_cols.iter().enumerate() {
+        let logical_mapping = insertion
+            .get_col_mapping_by_name(physical_col.name.as_ref().unwrap())
+            .unwrap();
+        program.emit_insn(Insn::Copy {
+            src_reg: logical_mapping.register,
+            dst_reg: reordered_start_reg + i,
+            extra_amount: 0,
+        });
+    }
+
     let full_record_reg = program.alloc_register();
-    let affinity_str = insertion
-        .col_mappings
+    let affinity_str = physically_ordered_cols
         .iter()
-        .map(|col_mapping| col_mapping.column.affinity().aff_mask())
+        .map(|col| col.affinity().aff_mask())
         .collect::<String>();
 
+
     program.emit_insn(Insn::MakeRecord {
-        start_reg: insertion.first_col_register(),
+        start_reg: reordered_start_reg,
         count: insertion.col_mappings.len(),
         dest_reg: full_record_reg,
         index_name: Some(pk_index.name.clone()),
         affinity_str: Some(affinity_str),
     });
-    tracing::debug!(
-        "Emitted OP_MakeRecord for full row into register {}",
-        full_record_reg
-    );
+
+
 
     program.emit_insn(Insn::IdxInsert {
         cursor_id: pk_cursor_id,
         record_reg: full_record_reg,
-        unpacked_start: Some(insertion.first_col_register()),
+        unpacked_start: Some(reordered_start_reg),
         unpacked_count: Some(insertion.col_mappings.len() as u16),
         flags: IdxInsertFlags::new().nchange(true),
     });
@@ -2891,7 +2941,6 @@ fn translate_without_rowid_insert(
 
         tracing::debug!("Processing secondary index '{}'", index.name);
 
-
         let num_key_cols = index.n_key_col;
         let key_start_reg = program.alloc_registers(num_key_cols);
 

core/translate/main_loop.rs

@@ -185,8 +185,7 @@ pub fn init_loop(
             where_clause: None,
             index_method: None,
             is_primary_key: false,
-              n_key_col: group_by.as_ref().unwrap().exprs.len(),
-
+            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,7 +1278,7 @@ fn ephemeral_index_build(
             (None, None) => Ordering::Equal,
         }
     });
-     let n_key_col = ephemeral_columns.len();
+    let n_key_col = ephemeral_columns.len();
     let ephemeral_index = Index {
         name: format!(
             "ephemeral_{}_{}",
@@ -1297,7 +1297,7 @@ fn ephemeral_index_build(
             .is_some_and(|btree| btree.has_rowid),
         index_method: None,
         is_primary_key: false,
-         n_key_col,
+        n_key_col,
     };
 
     ephemeral_index

core/translate/order_by.rs

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

core/translate/subquery.rs

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