SQLite - Bulk insert compare

Working on a way to do a quick compare of similar data that is analogous to a structured row of data from a SQL table. Now the data itself will have a current version and a previous version and it is stored in a JSON format file.

At first I thought I can do a hash across the entirety of the record, but since the data is quite similar I would need to remember to use a significantly diverging hash function. This only comes into play when creating your own custom toy hashing algorithms. Using a standard open source established hashing function eliminates this issue i.e. SHA512. The second part more importantly is that you will need to make sure to keep the relevant fields in the hash to quickly disambiguate the records or apply the hash across the entire record.

I was thinking about this and thought I need a hash table but I want to make it more capable so more like a B-tree. Wait what uses B-trees and is tuned for maximum efficiency and speed? That's right databases and in this case this is the perfect case for using SQLite.

That means I should determine the unique row key field. In this case it is multiple columns and I want it to be a computed stored field for quick reference when doing inserts as I want to insert all the records in a bulk load and pull the results out.

In SQLite that means using the keywords GENERATED/STORED and the hex function across the key columns (some of which can be NULL so serialize them as empty fields)

Create table with generated uniqueness row hash

PRAGMA journal_mode = 'wal';

DROP TABLE IF EXISTS agreements;

CREATE TABLE IF NOT EXISTS agreements (
    file_id TEXT NOT NULL,
    dag TEXT NOT NULL,
    file_code TEXT NOT NULL,
    org_name TEXT,
    uploaded datetime NOT NULL,
    expired datetime,
    resources TEXT,
	compare_id TEXT GENERATED ALWAYS AS (hex(dag || file_code || uploaded || coalesce(expired, '') || coalesce(resources, ''))) STORED,
    UNIQUE(compare_id)
);

Insert all records, updates, duplicates, etc

For the bulk load this works best if you sort the inputs by the set input (previous/current) before loading the records so that the first value can get loaded first and then the overlay of the next value can replace the original value or sit next to it when that portion of the comparison set is loaded.

insert or replace into agreements (file_id, dag, file_code, uploaded, expired, org_name, resources)
values	('previous', 'org-1', 'abc', '2026-01-01', null, null, '1, 2'),
       ('previous', 'org-2', 'abc', '2026-01-01', null, null, null),
       ('previous', 'org-3', 'abc', '2026-01-01', '2026-01-03', null, null),
	   ('current', 'org-3', 'abc', '2026-01-01', '2026-01-03', null, null),
       ('current', 'org-1', 'abc', '2026-01-01', null, null, null),
       ('current', 'org-1', 'def', '2026-01-01', null, null, null),
       ('current', 'org-2', 'abc', '2026-01-01', '2026-10-01', null, null),
       ('current', 'org-4', 'def', '2026-01-01', null, null, null);

Partition records into groups take first value current if there is more than 1 value record has changed

Due to the nature of the way the data was loaded it places all the first records in the set and then by grouping and then the second set of records if matching will replace the current value and if no match sit next to the original and if the set is completely new will add a new grouping.

For example if you have a set of records like this

That would result in the following by doing a bulk load

select row_number() over (partition by dag, file_code order by file_id asc) row_num, dag, file_code, org_name, uploaded, expired, resources	
from agreements
order by dag, file_code, row_num

Now you can group the records and order each set of grouped records by the set id (file_id) so that you pull the relevant updated records. Note that the ordering will depend on which set you wish to show precedence over when loading.

Sample code

def result_set(input_records: list[AgreementExtra]) -> dict[str, list[Any]]:
    added: set[str] = set()
    removed: set[str] = set(['org-4'])
    changed: set[str] = set()
    results = dict(added=[], changed=[], removed=[])
    for key, dag_records in groupby(input_records, lambda z: z.dag):
        all_records = sorted(deque(dag_records), key=lambda z: z.row_num)
        current_dag = peekable(all_records).peek().dag
        latest = sorted(filter(lambda z: z.row_num == 1, all_records), key=lambda z: z.file_code)
        if current_dag in added:
            results['added'].extend(latest)
            continue
        if current_dag in removed:
            results['removed'].extend(latest)
            continue
        if len(list(filter(lambda z: z.row_num > 1, all_records))) > 0:
            results['changed'].extend(latest)
    return results

with pydapper.connect("sqlite://pydapper.db") as commands:
    commands.execute(create_stmt)
    commands.execute("DELETE FROM agreements")
    commands.execute(insert_stmt)
    records = commands.query(sql=select_stmt, model=AgreementExtra)
    ic(result_set(input_records=records))