mirror of
https://github.com/simonw/datasette.git
synced 2026-07-22 15:34:34 +02:00
Rebuild internal catalog for a database in a single atomic write task
populate_schema_tables() previously ran six separate transactions per database: one task deleting that database's catalog rows followed by five execute_write_many inserts. Readers of the internal database could observe the intermediate state where a database had no catalog rows. Schema details are still collected on a read connection of the target database first; the delete-and-reinsert now happens inside one execute_write_fn task so the rebuild is atomic. Refs #2831 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N76afGMhBRQk528VF1LTpR
This commit is contained in:
Claude
parent
26d326c709
commit
a3f8b440e6
3 changed files with 112 additions and 61 deletions
|
|
@ -183,26 +183,6 @@ async def init_internal_db(db):
|
|||
async def populate_schema_tables(internal_db, db):
|
||||
database_name = db.name
|
||||
|
||||
def delete_everything(conn):
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_tables WHERE database_name = ?", [database_name]
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_views WHERE database_name = ?", [database_name]
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_columns WHERE database_name = ?", [database_name]
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_foreign_keys WHERE database_name = ?",
|
||||
[database_name],
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_indexes WHERE database_name = ?", [database_name]
|
||||
)
|
||||
|
||||
await internal_db.execute_write_fn(delete_everything)
|
||||
|
||||
tables = (await db.execute("select * from sqlite_master WHERE type = 'table'")).rows
|
||||
views = (await db.execute("select * from sqlite_master WHERE type = 'view'")).rows
|
||||
|
||||
|
|
@ -266,47 +246,69 @@ async def populate_schema_tables(internal_db, db):
|
|||
indexes_to_insert,
|
||||
) = await db.execute_fn(collect_info)
|
||||
|
||||
await internal_db.execute_write_many(
|
||||
"""
|
||||
INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
|
||||
values (?, ?, ?, ?)
|
||||
""",
|
||||
tables_to_insert,
|
||||
)
|
||||
await internal_db.execute_write_many(
|
||||
"""
|
||||
INSERT INTO catalog_views (database_name, view_name, rootpage, sql)
|
||||
values (?, ?, ?, ?)
|
||||
""",
|
||||
views_to_insert,
|
||||
)
|
||||
await internal_db.execute_write_many(
|
||||
"""
|
||||
INSERT INTO catalog_columns (
|
||||
database_name, table_name, cid, name, type, "notnull", default_value, is_pk, hidden
|
||||
) VALUES (
|
||||
:database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden
|
||||
def update_catalog(conn):
|
||||
# Delete and reinsert as one write task so internal database readers
|
||||
# never observe a database with missing catalog rows
|
||||
# https://github.com/simonw/datasette/issues/2831
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_tables WHERE database_name = ?", [database_name]
|
||||
)
|
||||
""",
|
||||
columns_to_insert,
|
||||
)
|
||||
await internal_db.execute_write_many(
|
||||
"""
|
||||
INSERT INTO catalog_foreign_keys (
|
||||
database_name, table_name, "id", seq, "table", "from", "to", on_update, on_delete, match
|
||||
) VALUES (
|
||||
:database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_views WHERE database_name = ?", [database_name]
|
||||
)
|
||||
""",
|
||||
foreign_keys_to_insert,
|
||||
)
|
||||
await internal_db.execute_write_many(
|
||||
"""
|
||||
INSERT INTO catalog_indexes (
|
||||
database_name, table_name, seq, name, "unique", origin, partial
|
||||
) VALUES (
|
||||
:database_name, :table_name, :seq, :name, :unique, :origin, :partial
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_columns WHERE database_name = ?", [database_name]
|
||||
)
|
||||
""",
|
||||
indexes_to_insert,
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_foreign_keys WHERE database_name = ?",
|
||||
[database_name],
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM catalog_indexes WHERE database_name = ?", [database_name]
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
|
||||
values (?, ?, ?, ?)
|
||||
""",
|
||||
tables_to_insert,
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT INTO catalog_views (database_name, view_name, rootpage, sql)
|
||||
values (?, ?, ?, ?)
|
||||
""",
|
||||
views_to_insert,
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT INTO catalog_columns (
|
||||
database_name, table_name, cid, name, type, "notnull", default_value, is_pk, hidden
|
||||
) VALUES (
|
||||
:database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden
|
||||
)
|
||||
""",
|
||||
columns_to_insert,
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT INTO catalog_foreign_keys (
|
||||
database_name, table_name, "id", seq, "table", "from", "to", on_update, on_delete, match
|
||||
) VALUES (
|
||||
:database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match
|
||||
)
|
||||
""",
|
||||
foreign_keys_to_insert,
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT INTO catalog_indexes (
|
||||
database_name, table_name, seq, name, "unique", origin, partial
|
||||
) VALUES (
|
||||
:database_name, :table_name, :seq, :name, :unique, :origin, :partial
|
||||
)
|
||||
""",
|
||||
indexes_to_insert,
|
||||
)
|
||||
|
||||
await internal_db.execute_write_fn(update_catalog)
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ Unreleased
|
|||
- ``await db.execute_write()`` detects statements that SQLite cannot execute inside a transaction - ``VACUUM``, ``ATTACH``, ``DETACH`` and ``PRAGMA`` - and runs them in autocommit mode instead. (:issue:`2831`)
|
||||
- ``await db.execute_write_script()`` is now transactional, matching its documentation: if any statement in the script fails, none of its statements are applied. Scripts containing statements that cannot run inside a transaction, or that manage transactions themselves, fall back to the previous ``conn.executescript()`` autocommit behavior. (:issue:`2831`)
|
||||
- The JSON write API is now atomic per request: ``/db/-/create`` with initial rows, multi-operation ``/db/table/-/alter`` calls and inserts using ``"return": true`` now either fully apply or roll back entirely if any part fails. Previously a failure part way through could leave earlier writes from the same request permanently committed. (:issue:`2831`)
|
||||
- Rebuilding the internal database catalog for a database is now a single atomic write. Previously the rebuild used six separate transactions, so queries against the internal database could observe a database with missing catalog rows while a rebuild was in progress. (:issue:`2831`)
|
||||
|
||||
.. _v1_0_a36:
|
||||
|
||||
|
|
|
|||
|
|
@ -334,3 +334,51 @@ async def test_orphan_stale_catalog_child_entries_removed(tmp_path):
|
|||
assert response.status_code == 200
|
||||
|
||||
ds2.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_populate_schema_tables_is_a_single_write_task(tmp_path):
|
||||
# https://github.com/simonw/datasette/issues/2831
|
||||
# The catalog rebuild should be one atomic write task, so readers of the
|
||||
# internal database never observe a database with deleted-but-not-yet-
|
||||
# reinserted catalog rows
|
||||
from datasette.app import Datasette
|
||||
from datasette.utils.internal_db import populate_schema_tables
|
||||
|
||||
path = str(tmp_path / "data.db")
|
||||
conn = sqlite3.connect(path)
|
||||
conn.execute("create table t (id integer primary key, name text)")
|
||||
conn.execute("create view v as select * from t")
|
||||
conn.close()
|
||||
ds = Datasette([path])
|
||||
await ds.invoke_startup()
|
||||
await ds.refresh_schemas()
|
||||
internal_db = ds.get_internal_database()
|
||||
|
||||
write_fns = []
|
||||
original_execute_write_fn = internal_db.execute_write_fn
|
||||
|
||||
async def counting_execute_write_fn(fn, *args, **kwargs):
|
||||
write_fns.append(fn)
|
||||
return await original_execute_write_fn(fn, *args, **kwargs)
|
||||
|
||||
internal_db.execute_write_fn = counting_execute_write_fn
|
||||
try:
|
||||
await populate_schema_tables(internal_db, ds.get_database("data"))
|
||||
finally:
|
||||
internal_db.execute_write_fn = original_execute_write_fn
|
||||
|
||||
assert len(write_fns) == 1
|
||||
# And the catalog should be correct
|
||||
tables = await internal_db.execute(
|
||||
"select table_name from catalog_tables where database_name = 'data'"
|
||||
)
|
||||
assert [row["table_name"] for row in tables.rows] == ["t"]
|
||||
views = await internal_db.execute(
|
||||
"select view_name from catalog_views where database_name = 'data'"
|
||||
)
|
||||
assert [row["view_name"] for row in views.rows] == ["v"]
|
||||
columns = await internal_db.execute(
|
||||
"select name from catalog_columns where database_name = 'data' and table_name = 't'"
|
||||
)
|
||||
assert {row["name"] for row in columns.rows} == {"id", "name"}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue