fix: Accept equal timestamps in the Postgres timestamp conflict check

README.md

@@ -339,7 +339,7 @@ The [JDBC connector for PostgreSQL](connectors/postgresql-connector) (`jdbc-sqrl
 The `sink.on-conflict.action` values map to the following PostgreSQL `ON CONFLICT` behaviors:
 
 - `update`: The existing row is overwritten with the incoming row (`ON CONFLICT (...) DO UPDATE SET ...`). This is the default.
-- `timestamp`: The existing row is overwritten only if the incoming row's `sink.on-conflict.timestamp-column` value is strictly greater than the existing row's. 
+- `timestamp`: The existing row is overwritten only if the incoming row's `sink.on-conflict.timestamp-column` value is greater than or equal to the existing row's. 
   Otherwise, the conflicting insert is silently skipped, so out-of-order events do not roll back newer state.
 - `ignore`: The existing row is kept, and the conflicting insert is silently skipped (`ON CONFLICT (...) DO NOTHING`).
 

connectors/postgresql-connector/src/main/java/com/datasqrl/flinkrunner/connector/postgresql/jdbc/SqrlPostgresDialect.java

@@ -88,6 +88,7 @@ public Optional<String> getUpsertStatement(
                   + ") DO UPDATE SET "
                   + buildUpdateClause(fieldNames));
       case TIMESTAMP -> {
+        // On equal timestamps the incoming row wins, as in Flink's keep-last-row deduplication
         var quotedTsCol = quoteIdentifier(onConflictTimestampCol);
         yield Optional.of(
             insertStatement
@@ -97,7 +98,7 @@ public Optional<String> getUpsertStatement(
                 + buildUpdateClause(fieldNames)
                 + " WHERE EXCLUDED."
                 + quotedTsCol
-                + " > "
+                + " >= "
                 + quoteIdentifier(tableName)
                 + "."
                 + quotedTsCol);

connectors/postgresql-connector/src/test/java/com/datasqrl/flinkrunner/connector/postgresql/jdbc/SqrlPostgresDialectTest.java

@@ -62,7 +62,7 @@ void upsertWithTimestampActionGuardsUpdateWithTimestampPredicate() {
     assertThat(sql)
         .contains("ON CONFLICT (\"id\") DO UPDATE SET ")
         .contains("\"value\"=EXCLUDED.\"value\"")
-        .endsWith(" WHERE EXCLUDED.\"event_time\" > \"events\".\"event_time\"");
+        .endsWith(" WHERE EXCLUDED.\"event_time\" >= \"events\".\"event_time\"");
   }
 
   @Test
@@ -81,6 +81,6 @@ void upsertWithTimestampActionPicksGivenTimestampColumn() {
 
     var sql = dialect.getUpsertStatement(TABLE, FIELDS, SINGLE_KEY).orElseThrow();
 
-    assertThat(sql).endsWith(" WHERE EXCLUDED.\"value\" > \"events\".\"value\"");
+    assertThat(sql).endsWith(" WHERE EXCLUDED.\"value\" >= \"events\".\"value\"");
   }
 }

connectors/postgresql-connector/src/test/java/com/datasqrl/flinkrunner/connector/postgresql/jdbc/SqrlPostgresTimestampConflictTest.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.postgresql.jdbc;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datasqrl.flinkrunner.connector.postgresql.jdbc.SqrlPostgresOptions.OnConflictAction;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatement;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.postgresql.PostgreSQLContainer;
+
+class SqrlPostgresTimestampConflictTest {
+
+  private static final String TABLE = "events";
+  private static final String[] FIELDS = {"id", "value", "event_time"};
+  private static final Timestamp STORED_TIME = Timestamp.valueOf("2026-01-01 00:00:10");
+
+  private static final PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17");
+  private static Connection conn;
+
+  @BeforeAll
+  static void startPostgres() throws SQLException {
+    postgres.start();
+    conn =
+        DriverManager.getConnection(
+            postgres.getJdbcUrl(), postgres.getUsername(), postgres.getPassword());
+  }
+
+  @AfterAll
+  static void stopPostgres() throws SQLException {
+    conn.close();
+    postgres.stop();
+  }
+
+  @BeforeEach
+  void createTableWithStoredRow() throws SQLException {
+    try (var stmt = conn.createStatement()) {
+      stmt.execute("DROP TABLE IF EXISTS \"events\"");
+      stmt.execute(
+          "CREATE TABLE \"events\" (\"id\" BIGINT PRIMARY KEY, \"value\" TEXT, \"event_time\" TIMESTAMP)");
+    }
+    upsert("stored", STORED_TIME);
+  }
+
+  @Test
+  void givenStoredRow_whenUpsertWithNewerTimestamp_thenRowIsUpdated() throws SQLException {
+    upsert("newer", Timestamp.valueOf("2026-01-01 00:00:11"));
+
+    assertThat(storedValue()).isEqualTo("newer");
+  }
+
+  @Test
+  void givenStoredRow_whenUpsertWithEqualTimestamp_thenRowIsUpdated() throws SQLException {
+    upsert("equal", STORED_TIME);
+
+    assertThat(storedValue()).isEqualTo("equal");
+  }
+
+  @Test
+  void givenStoredRow_whenUpsertWithOlderTimestamp_thenRowIsKept() throws SQLException {
+    upsert("older", Timestamp.valueOf("2026-01-01 00:00:09"));
+
+    assertThat(storedValue()).isEqualTo("stored");
+  }
+
+  private static void upsert(String value, Timestamp eventTime) throws SQLException {
+    var dialect = new SqrlPostgresDialect(OnConflictAction.TIMESTAMP, "event_time");
+    var sql = dialect.getUpsertStatement(TABLE, FIELDS, new String[] {"id"}).orElseThrow();
+    try (var stmt = FieldNamedPreparedStatement.prepareStatement(conn, sql, FIELDS)) {
+      stmt.setLong(0, 1L);
+      stmt.setString(1, value);
+      stmt.setTimestamp(2, eventTime);
+      stmt.addBatch();
+      stmt.executeBatch();
+    }
+  }
+
+  private static String storedValue() throws SQLException {
+    try (var stmt = conn.createStatement();
+        var rs = stmt.executeQuery("SELECT \"value\" FROM \"events\" WHERE \"id\" = 1")) {
+      rs.next();
+      return rs.getString(1);
+    }
+  }
+}