ci(postgres): add a static pre-commit check for MySQL-only SQL

The Postgres test job is label-gated, so it does not run on every PR. This adds an
always-on pre-commit hook that statically flags the *mechanical* breaks: MySQL-only
functions (timestamp(date,time), timediff, str_to_date, date_format/add/sub,
group_concat, period_diff, SQL IF()), SHOW INDEX/TABLES/COLUMNS, single-quoted
aliases, UPDATE..JOIN, interpolated/f-string SQL carrying MySQL-isms,
set_value/db_set(<Check>, bool), and MySQL SHOW INDEX result keys.

It deliberately does NOT flag the framework auto-translations (ifnull->coalesce,
backtick/locate/REGEXP, .like()->ILIKE) nor the *semantic* divergences (loose GROUP
BY, case-sensitive ==/IN, NULL ordering, tiebreakers) — those need the test suite,
which remains the backstop. AST + structure-gated regex keep false positives near
zero (docstrings and prose skipped); '# pg-ok' exempts intentional MariaDB-only
branches. Scoped to erpnext/ excluding patches/. Includes a unit test of the checker.

.github/helper/postgres_compat.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""Static guard against MySQL-only SQL that breaks on PostgreSQL.
+
+The Postgres test job is label-gated, so it does not run on every PR. This pre-commit
+hook is the always-on first line of defence: it flags the *mechanical* Postgres breaks
+that static analysis can catch reliably with a low false-positive rate.
+
+It deliberately does NOT try to catch the *semantic* divergences (loose GROUP BY,
+case-sensitive ==/IN, NULL ordering, ORDER BY ... LIMIT 1 tiebreakers, integer-division
+intent, savepoint discipline) — those genuinely need the test suite. Run the full suite
+on a Postgres site for those.
+
+Escape hatch: put `# pg-ok` anywhere on the offending statement's line span (e.g. on a
+`SHOW INDEX` query that lives inside an `if frappe.db.db_type == "mariadb":` branch).
+
+Usage: postgres_compat.py <file.py> [<file.py> ...]   (pre-commit passes staged files)
+"""
+
+from __future__ import annotations
+
+import ast
+import re
+import sys
+
+IGNORE = "pg-ok"
+
+# Strings are only scanned for the patterns below when they have real SQL *structure*
+# (not just an English word like "select" or "from"), to keep false positives near zero.
+SQL_HINT = re.compile(
+	r"\bselect\b[\s\S]{0,800}\bfrom\b"        # SELECT ... FROM
+	r"|\bupdate\b[\s\S]{0,400}\bset\b"         # UPDATE ... SET
+	r"|\bdelete\s+from\b"
+	r"|\binsert\s+into\b"
+	r"|\bshow\s+(?:index|tables|columns)\b"
+	r"|\bfrom\s+[\"'`]?tab",                   # FROM `tabDocType`
+	re.I,
+)
+
+# MySQL-only constructs with NO frappe auto-translation. (frappe.db.sql already rewrites
+# ifnull->coalesce on all engines and backtick/locate/REGEXP on Postgres, and .like()
+# renders ILIKE — so those are NOT listed here; flagging them would be false positives.)
+SQL_PATTERNS: list[tuple[re.Pattern, str]] = [
+	(re.compile(r"\btimestamp\s*\(\s*[^,()]+,", re.I),
+	 "timestamp(date, time) is MySQL-only -> use CombineDatetime() or a precomputed datetime column"),
+	(re.compile(r"\btimediff\s*\(", re.I),
+	 "timediff() is MySQL-only -> compute the delta in Python"),
+	(re.compile(r"\bstr_to_date\s*\(", re.I),
+	 "str_to_date() is MySQL-only -> parse in Python and pass a real date"),
+	(re.compile(r"\bdate_format\s*\(", re.I),
+	 "date_format() is MySQL-only -> filter on a date range instead"),
+	(re.compile(r"\bdate_(add|sub)\s*\(", re.I),
+	 "date_add()/date_sub() are MySQL-only -> use Python date math or interval arithmetic"),
+	(re.compile(r"\bgroup_concat\s*\(", re.I),
+	 "group_concat() is MySQL-only -> use GroupConcat (string_agg) or aggregate in Python"),
+	(re.compile(r"\bperiod_diff\s*\(", re.I),
+	 "period_diff() is MySQL-only -> compute in Python"),
+	(re.compile(r"\bshow\s+index\b", re.I),
+	 "SHOW INDEX is MySQL-only -> use frappe.db.has_index() / get_column_index()"),
+	(re.compile(r"\bshow\s+(tables|columns)\b", re.I),
+	 "SHOW TABLES/COLUMNS is MySQL-only -> use frappe.db.get_tables()/table_columns / information-schema helpers"),
+	(re.compile(r"\bas\s+'[^']+'", re.I),
+	 "single-quoted column alias breaks on Postgres -> use a bare or double-quoted alias"),
+	(re.compile(r"\bif\s*\(", re.I),
+	 "SQL IF() is MySQL-only -> use CASE WHEN ... THEN ... ELSE ... END (frappe.qb.Case())"),
+]
+
+# UPDATE ... JOIN: both keywords in the same SQL string.
+UPDATE_JOIN = (re.compile(r"\bupdate\b", re.I), re.compile(r"\bjoin\b", re.I))
+
+MYSQL_RESULT_KEYS = {"Column_name", "Key_name", "Seq_in_index", "Non_unique", "Index_type"}
+
+SET_BOOL_FUNCS = {"set_value", "db_set"}
+
+
+def _docstring_ids(tree: ast.AST) -> set[int]:
+	"""ids of Constant nodes that are docstrings (so prose describing the rules isn't flagged)."""
+	ids: set[int] = set()
+	for node in ast.walk(tree):
+		if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+			body = getattr(node, "body", None)
+			if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
+				ids.add(id(body[0].value))
+	return ids
+
+
+class Visitor(ast.NodeVisitor):
+	def __init__(self, lines: list[str], docstrings: set[int]):
+		self.lines = lines
+		self.docstrings = docstrings
+		self.violations: list[tuple[int, str]] = []
+
+	def _ignored(self, node: ast.AST) -> bool:
+		start = getattr(node, "lineno", 1)
+		end = getattr(node, "end_lineno", start) or start
+		# honour `# pg-ok` anywhere on the node's line span, or on the line just above
+		# (the enclosing call, e.g. `frappe.db.sql(  # pg-ok`).
+		lo = max(0, start - 2)
+		return any(IGNORE in self.lines[i] for i in range(lo, min(end, len(self.lines))))
+
+	def _flag(self, node: ast.AST, msg: str) -> None:
+		if not self._ignored(node):
+			self.violations.append((getattr(node, "lineno", 1), msg))
+
+	def _scan_sql(self, text: str, node: ast.AST) -> None:
+		if not SQL_HINT.search(text):
+			return
+		for pattern, msg in SQL_PATTERNS:
+			if pattern.search(text):
+				self._flag(node, msg)
+		if UPDATE_JOIN[0].search(text) and UPDATE_JOIN[1].search(text):
+			self._flag(node, "UPDATE ... JOIN is MySQL-only -> use a correlated subquery (WHERE ... IN/EXISTS)")
+
+	def visit_Constant(self, node: ast.Constant) -> None:
+		# plain string literals, incl. `"...".format()` and `"..." % (...)` templates
+		if isinstance(node.value, str) and id(node) not in self.docstrings:
+			self._scan_sql(node.value, node)
+		self.generic_visit(node)
+
+	def visit_JoinedStr(self, node: ast.JoinedStr) -> None:
+		# f-string: scan its STATIC text (interpolated values become a placeholder) so MySQL-isms
+		# in dynamic SQL are caught, without flagging safe interpolation of identifiers.
+		text = "".join(
+			v.value if isinstance(v, ast.Constant) and isinstance(v.value, str) else " ? "
+			for v in node.values
+		)
+		self._scan_sql(text, node)
+		# don't recurse: child literal chunks would otherwise be re-scanned individually
+
+	def visit_Call(self, node: ast.Call) -> None:
+		fn = node.func
+		name = fn.attr if isinstance(fn, ast.Attribute) else (fn.id if isinstance(fn, ast.Name) else "")
+
+		# row.get("Column_name") — MySQL SHOW INDEX result key
+		if name == "get" and node.args and isinstance(node.args[0], ast.Constant) and node.args[0].value in MYSQL_RESULT_KEYS:
+			self._flag(node, f'"{node.args[0].value}" is a MySQL SHOW INDEX result key -> use frappe.db.has_index()/get_column_index()')
+
+		# set_value(..., True) / db_set("field", True) on a Check (int) column
+		if name in SET_BOOL_FUNCS:
+			for a in node.args:
+				if isinstance(a, ast.Constant) and isinstance(a.value, bool):
+					self._flag(node, f"{name}(..., {a.value}) sets an int/Check column with a bool -> pass 1/0 (Postgres rejects bool->smallint)")
+				elif isinstance(a, ast.Dict):
+					for v in a.values:
+						if isinstance(v, ast.Constant) and isinstance(v.value, bool):
+							self._flag(node, f"{name}(...) sets an int/Check column with a bool in a dict -> pass 1/0 (Postgres rejects bool->smallint)")
+
+		self.generic_visit(node)
+
+	def visit_Subscript(self, node: ast.Subscript) -> None:
+		key = node.slice
+		if isinstance(key, ast.Constant) and key.value in MYSQL_RESULT_KEYS:
+			self._flag(node, f'"{key.value}" is a MySQL SHOW INDEX result key -> use frappe.db.has_index()/get_column_index()')
+		self.generic_visit(node)
+
+
+def check_file(path: str) -> list[str]:
+	try:
+		src = open(path, encoding="utf-8").read()
+	except (OSError, UnicodeDecodeError):
+		return []
+	try:
+		tree = ast.parse(src, filename=path)
+	except SyntaxError:
+		return []  # check-ast hook reports real syntax errors
+	v = Visitor(src.splitlines(), _docstring_ids(tree))
+	v.visit(tree)
+	return [f"{path}:{line}: [pg-compat] {msg}" for line, msg in sorted(set(v.violations))]
+
+
+def main(argv: list[str]) -> int:
+	out: list[str] = []
+	for path in argv:
+		if path.endswith(".py"):
+			out.extend(check_file(path))
+	if out:
+		print("\n".join(out))
+		print(
+			f"\n{len(out)} PostgreSQL-incompatibility issue(s). Fix them, or add `# pg-ok` to a "
+			"line that is intentionally MariaDB-only (e.g. inside an `if frappe.db.db_type == 'mariadb':` branch)."
+		)
+		return 1
+	return 0
+
+
+if __name__ == "__main__":
+	raise SystemExit(main(sys.argv[1:]))

.pre-commit-config.yaml

@@ -66,6 +66,18 @@ repos:
       - id: ruff-format
         name: "Run ruff formatter"
 
+  - repo: local
+    hooks:
+      - id: postgres-compat
+        name: "PostgreSQL compatibility (static check)"
+        description: "Flags MySQL-only SQL that breaks on Postgres; the label-gated PG test job is the backstop for semantic divergences."
+        entry: .github/helper/postgres_compat.py
+        language: script
+        files: ^erpnext/.*\.py$
+        # patches/ are historical, version-gated migrations (skipped on fresh Postgres installs);
+        # out of scope for the always-on gate. test_postgres_compat.py holds intentional bad-SQL fixtures.
+        exclude: ^erpnext/patches/|^erpnext/tests/test_postgres_compat\.py$
+
 ci:
     autoupdate_schedule: weekly
     skip: []

erpnext/accounts/doctype/process_payment_reconciliation/process_payment_reconciliation.py

@@ -503,7 +503,7 @@ def reconcile(doc: None | str = None) -> None:
 								)
 			else:
 				frappe.db.set_value("Process Payment Reconciliation Log", log, "status", "Reconciled")
-				frappe.db.set_value("Process Payment Reconciliation Log", log, "reconciled", True)
+				frappe.db.set_value("Process Payment Reconciliation Log", log, "reconciled", 1)
 				frappe.db.set_value("Process Payment Reconciliation", doc, "status", "Completed")
 
 

erpnext/tests/test_perf.py

@@ -33,7 +33,7 @@ def _is_leading_index_column(doctype: str, field: str) -> bool:
 	# `table` is a trusted constant (from INDEXED_FIELDS); a table identifier can't be a %s
 	# placeholder in SHOW INDEX, so the f-string is unavoidable and safe here.
 	return bool(
-		frappe.db.sql(
+		frappe.db.sql(  # pg-ok: MariaDB-only branch; Postgres is handled above via pg_index
 			f"""SHOW INDEX FROM `{table}` WHERE Column_name = %s AND Seq_in_index = 1""",
 			(field,),
 		)

erpnext/tests/test_postgres_compat.py

@@ -0,0 +1,103 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+"""Unit tests for the .github/helper/postgres_compat.py pre-commit checker.
+
+This file is excluded from the postgres-compat hook itself (see .pre-commit-config.yaml)
+because the fixtures below intentionally contain MySQL-only SQL.
+"""
+
+import importlib.util
+import os
+import tempfile
+import unittest
+
+import frappe
+
+_HELPER = os.path.join(frappe.get_app_path("erpnext"), "..", ".github", "helper", "postgres_compat.py")
+_spec = importlib.util.spec_from_file_location("postgres_compat", _HELPER)
+pgc = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(pgc)
+
+
+def violations(code: str) -> list[str]:
+	with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
+		f.write(code)
+		path = f.name
+	try:
+		return pgc.check_file(path)
+	finally:
+		os.unlink(path)
+
+
+class TestPostgresCompat(unittest.TestCase):
+	def _assert_flag(self, code: str, needle: str):
+		out = violations(code)
+		self.assertTrue(
+			any(needle in v for v in out), f"expected '{needle}' to be flagged in:\n{code}\ngot: {out}"
+		)
+
+	def _assert_clean(self, code: str):
+		self.assertEqual(violations(code), [], f"expected no flags in:\n{code}")
+
+	# --- catches the mechanical breaks ---
+	def test_timestamp_two_arg(self):
+		self._assert_flag(
+			'frappe.db.sql("select timestamp(posting_date, posting_time) from `tabSLE`")',
+			"timestamp(date, time)",
+		)
+
+	def test_show_index(self):
+		self._assert_flag('frappe.db.sql("show index from `tabItem`")', "SHOW INDEX")
+
+	def test_update_join(self):
+		self._assert_flag(
+			'frappe.db.sql("update `tabA` a join `tabB` b on a.x=b.x set a.y=b.y")', "UPDATE ... JOIN"
+		)
+
+	def test_single_quoted_alias(self):
+		self._assert_flag(
+			"frappe.db.sql(\"select substr(x,1,3) as 'foo' from `tabA`\")", "single-quoted column alias"
+		)
+
+	def test_group_concat(self):
+		self._assert_flag('frappe.db.sql("select group_concat(name) from `tabA`")', "group_concat()")
+
+	def test_set_value_bool(self):
+		self._assert_flag('frappe.db.set_value("Company", c, "some_check", True)', "bool")
+
+	def test_db_set_bool(self):
+		self._assert_flag('doc.db_set("is_default", False)', "bool")
+
+	def test_mysql_result_key(self):
+		self._assert_flag('row.get("Column_name")', "Column_name")
+
+	def test_fstring_sql(self):
+		self._assert_flag("frappe.db.sql(f\"select date_format(d, '%Y') from `tab{dt}`\")", "date_format()")
+
+	# --- does not false-positive on safe shapes ---
+	def test_ifnull_is_auto_translated(self):
+		self._assert_clean('frappe.db.sql("select ifnull(qty, 0) from `tabBin`")')
+
+	def test_like_is_ilike(self):
+		self._assert_clean('frappe.db.get_all("Item", filters={"item_name": ["like", "%x%"]})')
+
+	def test_prose_with_sql_words(self):
+		# a translatable message that merely contains "select" and "as '...'"
+		self._assert_clean(
+			"frappe.throw(_(\"Cannot select charge type as 'On Previous Row' for first row\"))"
+		)
+
+	def test_docstring_describing_rule(self):
+		self._assert_clean(
+			'def f():\n\t"""Avoid MariaDB-only DATE_FORMAT(); read from the pg_index catalog instead."""\n\treturn 1\n'
+		)
+
+	def test_qb_is_clean(self):
+		self._assert_clean("frappe.qb.from_(sle).select(CombineDatetime(sle.posting_date, sle.posting_time))")
+
+	def test_pg_ok_suppresses(self):
+		self._assert_clean('frappe.db.sql(  # pg-ok\n\t"show index from `tabItem`"\n)')
+
+
+if __name__ == "__main__":
+	unittest.main()