Delegate bulk create/update to ryx_core

Cargo.toml

@@ -62,6 +62,7 @@ sqlx = { version = "0.8.6", features = [
 # Full tokio runtime. "full" is fine for a library crate — callers can restrict
 # features if they need a lighter binary.
 tokio = { version = "1.40", features = ["full"] }
+smallvec = "1.13"
 
 # ── Serialization ─────────────────────────────────────────────────────────────
 # serde + serde_json: used to pass structured data between Rust and Python

ryx-query/Cargo.toml

@@ -11,6 +11,7 @@ serde_json = "1"
 thiserror = "2"
 once_cell = "1"
 tracing = "0.1"
+smallvec = "1.13"
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["async_tokio"] }

ryx-query/src/ast.rs

@@ -32,7 +32,7 @@ pub enum SqlValue {
     Text(String),
     /// Used by `__in` and `__range` lookups. The compiler expands it into
     /// multiple bind placeholders.
-    List(Vec<SqlValue>),
+    List(smallvec::SmallVec<[Box<SqlValue>; 4]>),
 }
 
 impl SqlValue {

ryx-query/src/compiler/compiler.rs

@@ -16,6 +16,7 @@ use crate::errors::{QueryError, QueryResult};
 use crate::lookups::date_lookups as date;
 use crate::lookups::json_lookups as json;
 use crate::lookups::{self, LookupContext};
+use smallvec::SmallVec;
 
 pub use super::helpers::{apply_like_wrapping, qualified_col, split_qualified, KNOWN_TRANSFORMS};
 
@@ -24,12 +25,12 @@ use super::helpers;
 #[derive(Debug, Clone)]
 pub struct CompiledQuery {
     pub sql: String,
-    pub values: Vec<SqlValue>,
+    pub values: SmallVec<[SqlValue; 8]>,
     pub db_alias: Option<String>,
 }
 
 pub fn compile(node: &QueryNode) -> QueryResult<CompiledQuery> {
-    let mut values: Vec<SqlValue> = Vec::new();
+    let mut values: SmallVec<[SqlValue; 8]> = SmallVec::new();
     let sql = match &node.operation {
         QueryOperation::Select { columns } => {
             compile_select(node, columns.as_deref(), &mut values)?
@@ -53,7 +54,7 @@ pub fn compile(node: &QueryNode) -> QueryResult<CompiledQuery> {
 fn compile_select(
     node: &QueryNode,
     columns: Option<&[String]>,
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
 ) -> QueryResult<String> {
     let base_cols = match columns {
         None => "*".to_string(),
@@ -134,7 +135,7 @@ fn compile_select(
     Ok(sql)
 }
 
-fn compile_aggregate(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResult<String> {
+fn compile_aggregate(node: &QueryNode, values: &mut SmallVec<[SqlValue; 8]>) -> QueryResult<String> {
     if node.annotations.is_empty() {
         return Err(QueryError::Internal(
             "aggregate() called with no aggregate expressions".into(),
@@ -158,7 +159,7 @@ fn compile_aggregate(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResul
     Ok(sql)
 }
 
-fn compile_count(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResult<String> {
+fn compile_count(node: &QueryNode, values: &mut SmallVec<[SqlValue; 8]>) -> QueryResult<String> {
     let mut sql = format!("SELECT COUNT(*) FROM {}", helpers::quote_col(&node.table));
     if !node.joins.is_empty() {
         sql.push(' ');
@@ -173,7 +174,7 @@ fn compile_count(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResult<St
     Ok(sql)
 }
 
-fn compile_delete(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResult<String> {
+fn compile_delete(node: &QueryNode, values: &mut SmallVec<[SqlValue; 8]>) -> QueryResult<String> {
     let mut sql = format!("DELETE FROM {}", helpers::quote_col(&node.table));
     let where_sql =
         compile_where_combined(&node.filters, node.q_filter.as_ref(), values, node.backend)?;
@@ -187,7 +188,7 @@ fn compile_delete(node: &QueryNode, values: &mut Vec<SqlValue>) -> QueryResult<S
 fn compile_update(
     node: &QueryNode,
     assignments: &[(String, SqlValue)],
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
 ) -> QueryResult<String> {
     if assignments.is_empty() {
         return Err(QueryError::Internal("UPDATE with no assignments".into()));
@@ -217,7 +218,7 @@ fn compile_insert(
     node: &QueryNode,
     cols_vals: &[(String, SqlValue)],
     returning_id: bool,
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
 ) -> QueryResult<String> {
     if cols_vals.is_empty() {
         return Err(QueryError::Internal("INSERT with no values".into()));
@@ -340,7 +341,7 @@ pub fn compile_order_by(clauses: &[crate::ast::OrderByClause]) -> String {
 fn compile_where_combined(
     filters: &[FilterNode],
     q: Option<&QNode>,
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
     backend: Backend,
 ) -> QueryResult<String> {
     let flat = if filters.is_empty() {
@@ -361,7 +362,11 @@ fn compile_where_combined(
     })
 }
 
-pub fn compile_q(q: &QNode, values: &mut Vec<SqlValue>, backend: Backend) -> QueryResult<String> {
+pub fn compile_q(
+    q: &QNode,
+    values: &mut SmallVec<[SqlValue; 8]>,
+    backend: Backend,
+) -> QueryResult<String> {
     match q {
         QNode::Leaf {
             field,
@@ -392,7 +397,7 @@ pub fn compile_q(q: &QNode, values: &mut Vec<SqlValue>, backend: Backend) -> Que
 
 fn compile_filters(
     filters: &[FilterNode],
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
     backend: Backend,
 ) -> QueryResult<String> {
     let parts: Vec<String> = filters
@@ -407,7 +412,7 @@ fn compile_single_filter(
     lookup: &str,
     value: &SqlValue,
     negated: bool,
-    values: &mut Vec<SqlValue>,
+    values: &mut SmallVec<[SqlValue; 8]>,
     backend: Backend,
 ) -> QueryResult<String> {
     let (base_column, applied_transforms, json_key) = if field.contains("__") {
@@ -474,9 +479,9 @@ fn compile_single_filter(
     }
 
     if lookup == "in" {
-        let items = match value {
-            SqlValue::List(v) => v.clone(),
-            other => vec![other.clone()],
+        let items: SmallVec<[SqlValue; 4]> = match value {
+            SqlValue::List(v) => v.iter().map(|x| (**x).clone()).collect(),
+            other => smallvec::smallvec![(*other).clone()],
         };
         if items.is_empty() {
             return Ok("(1 = 0)".into());
@@ -495,9 +500,9 @@ fn compile_single_filter(
     }
 
     if lookup == "has_any" || lookup == "has_all" {
-        let items = match value {
-            SqlValue::List(v) => v.clone(),
-            other => vec![other.clone()],
+        let items: SmallVec<[SqlValue; 4]> = match value {
+            SqlValue::List(v) => v.iter().map(|x| (**x).clone()).collect(),
+            other => smallvec::smallvec![(*other).clone()],
         };
         if items.is_empty() {
             return Ok("(1 = 0)".into());
@@ -537,7 +542,7 @@ fn compile_single_filter(
 
     if lookup == "range" {
         let (lo, hi) = match value {
-            SqlValue::List(v) if v.len() == 2 => (v[0].clone(), v[1].clone()),
+            SqlValue::List(v) if v.len() == 2 => (v[0].as_ref().clone(), v[1].as_ref().clone()),
             _ => return Err(QueryError::Internal("range needs exactly 2 values".into())),
         };
         values.push(lo);

ryx/bulk.py

@@ -25,13 +25,13 @@
 
 from __future__ import annotations
 
-# import asyncio
-# import itertools
 from typing import List, Sequence, Type, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from ryx.models import Model
 
+from ryx import ryx_core as _core
+
 
 def _detect_backend() -> str:
     """Detect the database backend from the RYX_DATABASE_URL env var.
@@ -113,12 +113,20 @@ async def bulk_create(
 
     pk_field = model._meta.pk_field
 
-    # Process in batches
+    # Process in batches — all SQL and execution handled in Rust
     for batch in _chunked(instances, batch_size):
-        pks = await _insert_batch(model, batch, fields, col_names, ignore_conflicts)
-        # Assign returned PKs to instances
-        for inst, pk in zip(batch, pks):
-            object.__setattr__(inst, pk_field.attname, pk)
+        rows = [[f.to_db(getattr(inst, f.attname)) for f in fields] for inst in batch]
+        res = await _core.bulk_insert(
+            model._meta.table_name,
+            col_names,
+            rows,
+            True,  # returning_id
+            ignore_conflicts,
+        )
+        # On PostgreSQL/SQLite res is list of ids; on MySQL res is rows_affected
+        if pk_field and isinstance(res, list):
+            for inst, pk in zip(batch, res):
+                object.__setattr__(inst, pk_field.attname, pk)
 
     return list(instances)
 
@@ -255,11 +263,7 @@ async def bulk_update(
     }
     total = 0
 
-    from ryx import ryx_core as _core
-    from ryx.pool_ext import execute_with_params
-
     for batch in _chunked(instances, batch_size):
-        # Collect valid instances (with pk set)
         valid = [inst for inst in batch if inst.pk is not None]
         if not valid:
             continue
@@ -268,55 +272,27 @@ async def bulk_update(
         pk_col = pk_field.column
         table = model._meta.table_name
 
-        # Build CASE WHEN clauses.
-        # Strategy: inline integers directly in SQL (zero FFI cost),
-        # use ? placeholders only for non-integer values.
-        case_clauses = []
-        all_values = []
-
+        # Collect values per column in the order of pks
+        col_names: List[str] = []
+        field_values: List[List[object]] = []
         for fname in update_fields:
             if fname not in field_objs:
                 continue
             fobj = field_objs[fname]
-            col = fobj.column
-            case_parts = [f'"{col}" = CASE "{pk_col}"']
-            for inst in valid:
-                val = fobj.to_db(getattr(inst, fname))
-                if isinstance(val, int) and not isinstance(val, bool):
-                    # Inline integers — zero FFI overhead
-                    case_parts.append(f"WHEN {inst.pk} THEN {val}")
-                else:
-                    case_parts.append("WHEN ? THEN ?")
-                    all_values.append(inst.pk)
-                    all_values.append(val)
-            case_parts.append("END")
-            case_clauses.append(" ".join(case_parts))
-
-        if not case_clauses:
-            continue
+            col_names.append(fobj.column)
+            vals = [fobj.to_db(getattr(inst, fname)) for inst in valid]
+            field_values.append(vals)
 
-        # WHERE IN — inline integer PKs
-        pk_parts = []
-        for pk in pks:
-            if isinstance(pk, int):
-                pk_parts.append(str(pk))
-            else:
-                pk_parts.append("?")
-                all_values.append(pk)
+        if not col_names:
+            continue
 
-        sql = (
-            f'UPDATE "{table}" SET '
-            f"{', '.join(case_clauses)} "
-            f'WHERE "{pk_col}" IN ({", ".join(pk_parts)})'
+        result = await _core.bulk_update(
+            table,
+            pk_col,
+            list(zip(col_names,field_values)),
+            pks,
         )
-
-        if all_values:
-            await execute_with_params(sql, all_values)
-        else:
-            from ryx.executor_helpers import raw_execute
-
-            await raw_execute(sql)
-        total += len(valid)
+        total += result
 
     return total
 

ryx/models.py

@@ -167,9 +167,6 @@ def __init__(self, alias: Optional[str] = None) -> None:
     def contribute_to_class(self, model: type, name: str) -> None:
         self._model = model
 
-    def contribute_to_class(self, model: type, name: str) -> None:
-        self._model = model
-
     def get_queryset(self):
         from ryx.queryset import QuerySet
 
@@ -292,9 +289,9 @@ async def bulk_create(self, instances: list[Model], batch_size: int = 500) -> li
     async def bulk_update(
         self, instances: list, fields: list, batch_size: int = 500
     ) -> int:
-        from ryx.bulk import bulk_update
+        from ryx.bulk import bulk_update as _update
 
-        return await bulk_update(self._model, instances, fields, batch_size=batch_size)
+        return await _update(self._model, instances, fields, batch_size=batch_size)
 
     async def bulk_delete(
         self, instances: Optional[list] = None, batch_size: int = 500