fix(migration): preserve quoted index semantics

This commit is contained in:
DEls 2026-09-19 19:36:50 +02:00
parent 485443a1e4
commit 1a97d7db44

View file

@ -54,20 +54,70 @@ def field(name, field_type="text", **options):
def index_name(statement):
"""Extract an SQLite/PocketBase index name from a CREATE INDEX statement."""
match = re.search(r"\bINDEX\s+`?([^`\s]+)`?", statement, re.IGNORECASE)
match = re.search(r"\bINDEX\s+(?:`([^`]+)`|\"([^\"]+)\"|([^\s(]+))", statement, re.IGNORECASE)
if not match:
raise ValueError(f"not a CREATE INDEX statement: {statement!r}")
return match.group(1).strip().lower()
return next(value for value in match.groups() if value is not None).strip().lower()
def normalise_index_sql(statement):
"""Compare index SQL by meaning, not PocketBase's formatting choices."""
value = statement.strip().rstrip(";")
value = re.sub(r"`([^`]*)`", lambda match: f"`{match.group(1).lower()}`", value)
chunks = []
literals = []
index = 0
while index < len(value):
character = value[index]
if character == "'":
start = index
index += 1
while index < len(value):
if value[index] == "'":
if index + 1 < len(value) and value[index + 1] == "'":
index += 2
continue
index += 1
break
index += 1
literals.append(value[start:index])
chunks.append(f"\x00{len(literals) - 1}\x00")
continue
if character in '`"':
quote = character
index += 1
identifier = []
while index < len(value):
if value[index] == quote:
if index + 1 < len(value) and value[index + 1] == quote:
identifier.append(quote)
index += 2
continue
index += 1
break
identifier.append(value[index])
index += 1
chunks.append(f"`{''.join(identifier).lower()}`")
continue
if character.isalnum() or character in "_$":
end = index + 1
while end < len(value) and (value[end].isalnum() or value[end] in "_$"):
end += 1
chunks.append(value[index:end].lower())
index = end
continue
chunks.append(character)
index += 1
value = "".join(chunks)
value = re.sub(r"\s+", " ", value).strip().lower()
# SQLite/PocketBase may insert or remove spaces around these tokens while
# preserving the same columns, uniqueness and WHERE predicate.
value = re.sub(r"\s*([(),=<>!])\s*", r"\1", value)
# Quoted identifiers and safe unquoted identifiers share one canonical
# representation; single-quoted literals are placeholders until restore.
value = value.replace("`", "")
for index, literal in enumerate(literals):
value = value.replace(f"\x00{index}\x00", literal)
return value
@ -208,6 +258,15 @@ def self_test():
]
for stored, near_miss in zip(pocketbase_stored, near_misses):
assert merge_indexes([stored], [near_miss])[1] == [("replace", near_miss)]
identifier_styles = (
"CREATE UNIQUE INDEX `idx_style` ON `Delivery_Notes` (`Batch_Note`) WHERE `Batch_Note` != ''",
'create unique index "IDX_STYLE" on "DELIVERY_NOTES" ("BATCH_NOTE") where "BATCH_NOTE" != \'\'',
"CREATE UNIQUE INDEX idx_style ON delivery_notes (batch_note) WHERE batch_note != ''",
)
assert merge_indexes([identifier_styles[0]], [identifier_styles[1]])[1] == []
assert merge_indexes([identifier_styles[0]], [identifier_styles[2]])[1] == []
literal_near_miss = identifier_styles[2].replace("!= ''", "!= ' '")
assert merge_indexes([identifier_styles[0]], [literal_near_miss])[1] == [("replace", literal_near_miss)]
print("self-test: index replacement and idempotence passed")
return 0