wip: insert with sql

Author: MarceloRobert

backend/kernelCI_app/constants/ingester.py

@@ -48,3 +48,5 @@
 except (ValueError, TypeError):
     logger.warning("Invalid INGEST_QUEUE_MAXSIZE, using default 5000")
     INGEST_QUEUE_MAXSIZE = 5000
+
+PRIO_DB = is_boolean_or_string_true(os.environ.get("PRIO_BD", "True"))

backend/kernelCI_app/management/commands/helpers/kcidbng_ingester.py

@@ -10,6 +10,7 @@
     INGEST_FLUSH_TIMEOUT_SEC,
     INGEST_QUEUE_MAXSIZE,
     VERBOSE,
+    PRIO_DB,
 )
 import threading
 import time
@@ -21,7 +22,7 @@
     extract_log_excerpt,
 )
 import kcidb_io
-from django.db import transaction
+from django.db import connections, transaction
 from kernelCI_app.models import Issues, Checkouts, Builds, Tests, Incidents
 
 from kernelCI_app.management.commands.helpers.process_submissions import (
@@ -111,23 +112,85 @@ def prepare_file_data(
         }
 
 
-def consume_buffer(buffer: list[TableModels], item_type: TableNames) -> None:
+def consume_buffer(buffer: list[TableModels], table_name: TableNames) -> None:
     """
     Consume a buffer of items and insert them into the database.
     This function is called by the db_worker thread.
     """
     if not buffer:
         return
 
-    model = MODEL_MAP[item_type]
+    model = MODEL_MAP[table_name]
+
+    # Get nullable fields for coalesce
+    # Get non-nullable but updateable fields such as timestamp for other functions
+
+    updateable_model_fields = []
+    updateable_db_fields = []
+    query_params_properties: list[tuple] = []
+    for field in model._meta.fields:
+        if field.generated:
+            continue
+
+        field_name = (
+            field.name + "_id"
+            if field.get_internal_type() == "ForeignKey"
+            else field.name
+        )
+        real_name = field.db_column or field_name
+
+        operation = "GREATEST" if real_name == "_timestamp" else "COALESCE"
+
+        query_params_properties.append((real_name, operation))
+
+        updateable_model_fields.append(field_name)
+        updateable_db_fields.append(real_name)
+
+    print("🚀 ~ query_params_properties:", query_params_properties)
+    print("🚀 ~ all_updateable_model_fields:", updateable_model_fields)
+    print("🚀 ~ all_updateable_db_fields:", updateable_db_fields)
+
+    conflict_clauses = []
+    for field, op in query_params_properties:
+        if PRIO_DB:
+            conflict_clauses.append(
+                f"""
+            {field} = {op}({table_name}.{field}, EXCLUDED.{field})"""
+            )
+        else:
+            conflict_clauses.append(
+                f"""
+            {field} = {op}(EXCLUDED.{field}, {table_name}.{field})"""
+            )
+
+    query = f"""
+        INSERT INTO {table_name} (
+            {', '.join(updateable_db_fields)}
+        )
+        VALUES (
+            {', '.join(['%s'] * len(updateable_db_fields))}
+        )
+        ON CONFLICT (id)
+        DO UPDATE SET {', '.join(conflict_clauses)};
+    """
+    print("🚀 ~ query:", query)
+
+    params = []
+    for obj in buffer:
+        obj_values = []
+        for field in updateable_model_fields:
+            value = getattr(obj, field)
+            if isinstance(value, (dict, list)):
+                value = json.dumps(value)
+            obj_values.append(value)
+        params.append(tuple(obj_values))
+    print("🚀 ~ params:", params)
 
     t0 = time.time()
-    model.objects.bulk_create(
-        buffer,
-        batch_size=INGEST_BATCH_SIZE,
-        ignore_conflicts=True,
-    )
-    out("bulk_create %s: n=%d in %.3fs" % (item_type, len(buffer), time.time() - t0))
+    with connections["default"].cursor() as cursor:
+        cursor.executemany(query, params)
+
+    out("bulk_create %s: n=%d in %.3fs" % (table_name, len(buffer), time.time() - t0))
 
 
 def flush_buffers(

backend/kernelCI_app/management/commands/tmp_submissions/archive/base build.json

@@ -0,0 +1,33 @@
+{
+  "checkouts": [],
+  "builds": [
+    {
+      "checkout_id": "ingestertest:687ac2282ce2c1874ed7adad",
+      "id": "ingestertest:0000",
+      "origin": "ingestertest",
+      "comment": "pm-6.16-rc7-234-gc7de79e662b86",
+      "start_time": "2025-07-18T21:55:31.295000+00:00",
+      "architecture": "i386",
+      "compiler": "clang-17",
+      "config_name": "i386_defconfig+allmodconfig+CONFIG_FRAME_WARN=2048",
+      "status": "PASS",
+      "misc": {
+        "platform": "kubernetes",
+        "runtime": "k8s-all",
+        "lab": "k8s-all",
+        "job_id": "kci-687ac2d32ce2c1874ed7b174-kbuild-clang-17-i386-all-lvti1gzs",
+        "job_context": "gke_android-kernelci-external_us-central1-c_kci-us-central1",
+        "kernel_type": "bzimage",
+        "maestro_viewer": "https://api.kernelci.org/viewer?node_id=687ac2d32ce2c1874ed7b174"
+      },
+      "output_files": [],
+      "config_url": "https://files.kernelci.org/kbuild-clang-17-i386-allmodconfig-687ac2d32ce2c1874ed7b174/.config",
+      "log_url": "https://files.kernelci.org/kbuild-clang-17-i386-allmodconfig-687ac2d32ce2c1874ed7b174/build.log.gz",
+      "log_excerpt": ""
+    }
+  ],
+  "tests": [],
+  "issues": [],
+  "incidents": [],
+  "version": { "major": 5, "minor": 3 }
+}

backend/kernelCI_app/models.py

@@ -1,3 +1,6 @@
+"""Defines the models used in the main database.
+All models should have explicit id column for the ingester to work properly."""
+
 from django.db import models
 from django.contrib.postgres.fields import ArrayField
 from django.contrib.postgres.indexes import GinIndex