feat: re-write date/time functions (calls + DEFAULT) in omni inserts

.schema/pgdog.schema.json

@@ -224,6 +224,7 @@
       "$ref": "#/$defs/Rewrite",
       "default": {
         "enabled": false,
+        "omni_non_deterministic_functions": "ignore",
         "primary_key": "ignore",
         "shard_key": "error",
         "split_inserts": "error"
@@ -1939,6 +1940,11 @@
           "type": "boolean",
           "default": false
         },
+        "omni_non_deterministic_functions": {
+          "description": "Behavior when an `INSERT` is headed to an omnisharded table using a function (such as date-time functions)\nthat will not be consistent when performing the functions separately on each shard.\nThus, it re-writes all such functions before performing the `INSERT` with constant values to maintain consistency.\n\nExample: `NOW()` is re-written to `2026-09-15 18:14:09.123456-05` (or whatever the current time is)\nbefore performing the individual `INSERT` operations.\n\nThis applies to both `DEFAULT` table schema and functions called within a VALUES list of an `INSERT`.\n\n`ignore` allows the `INSERT` without modification.\n\n_Default:_ `ignore`\n\n<https://docs.pgdog.dev/configuration/pgdog.toml/rewrite/#omni_non_deterministic_functions>",
+          "$ref": "#/$defs/RewriteMode",
+          "default": "ignore"
+        },
         "primary_key": {
           "description": "Behavior for `INSERT` missing a `BIGINT` primary key: `error` rejects, `rewrite` auto-injects `pgdog.unique_id()`, `ignore` allows without modification.\n\n_Default:_ `ignore`\n\n<https://docs.pgdog.dev/configuration/pgdog.toml/rewrite/#primary_key>",
           "$ref": "#/$defs/RewriteMode",

integration/pgdog.toml

@@ -44,6 +44,7 @@ enabled = false
 shard_key = "ignore"
 split_inserts = "error"
 # primary_key = "rewrite"
+omni_non_deterministic_functions = "rewrite"
 
 # ------------------------------------------------------------------------------
 # ----- Database :: pgdog ------------------------------------------------------

integration/python/test_asyncpg.py

@@ -689,3 +689,45 @@ async def test_pgdog_role_selection():
             pass
 
     assert got_err
+
+
+# Test to make sure everything works with unnamed prepared statements when we need to re-write
+# Bind for `TimeFunction`
+@pytest.mark.asyncio
+async def test_omni_time_function_unnamed_statement():
+    """`statement_cache_size=0` makes asyncpg use unnamed prepared statements.
+    """
+    conn = await asyncpg.connect(
+        user="pgdog",
+        password="pgdog",
+        database="pgdog_sharded",
+        host="127.0.0.1",
+        port=6432,
+        statement_cache_size=0,
+    )
+    row_id = random.randrange(1 << 32, 1 << 63)
+
+    try:
+        result = await conn.execute(
+            "INSERT INTO sharded_omni (id, value, created_at) VALUES ($1, $2, now())",
+            row_id,
+            "unnamed",
+        )
+        assert result == "INSERT 0 1"
+
+        # Every shard must have gotten the same timestamp.
+        created_at = [
+            await conn.fetchval(
+                f"/* pgdog_shard: {shard} */ SELECT created_at"
+                " FROM sharded_omni WHERE id = $1",
+                row_id,
+            )
+            for shard in (0, 1)
+        ]
+        assert created_at[0] is not None
+        assert created_at[0] == created_at[1]
+    finally:
+        await conn.execute("DELETE FROM sharded_omni WHERE id = $1", row_id)
+        await conn.close()
+
+    no_out_of_sync()

integration/rust/Cargo.toml

@@ -25,4 +25,5 @@ libc = "0.2"
 rand = "0.9"
 bytes.workspace = true
 rust_decimal = { version = "1.42.0", features = ["macros"] }
+chrono-tz = "0.10.4"
 pgdog-stats = { path = "../../pgdog-stats" }

integration/rust/tests/integration/mod.rs

@@ -23,6 +23,7 @@ pub mod max;
 pub mod multi_set;
 pub mod notify;
 pub mod offset;
+pub mod omni_timestamps;
 pub mod partial_req;
 pub mod per_stmt_routing;
 pub mod prepared;