Fix PostgreSQL deleting too much with complex predicates

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -993,7 +993,13 @@ public OptionalLong delete(ConnectorSession session, JdbcTableHandle handle)
         checkArgument(handle.getSortOrder().isEmpty(), "Unable to delete when sort order is set: %s", handle);
         try (Connection connection = connectionFactory.openConnection(session)) {
             verify(connection.getAutoCommit());
-            PreparedQuery preparedQuery = queryBuilder.prepareDeleteQuery(this, session, connection, handle.getRequiredNamedRelation(), handle.getConstraint());
+            PreparedQuery preparedQuery = queryBuilder.prepareDeleteQuery(
+                    this,
+                    session,
+                    connection,
+                    handle.getRequiredNamedRelation(),
+                    handle.getConstraint(),
+                    getAdditionalPredicate(handle.getConstraintExpressions(), Optional.empty()));
             try (PreparedStatement preparedStatement = queryBuilder.prepareStatement(this, session, connection, preparedQuery)) {
                 return OptionalLong.of(preparedStatement.executeUpdate());
             }

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultQueryBuilder.java

@@ -141,13 +141,20 @@ public PreparedQuery prepareDeleteQuery(
             ConnectorSession session,
             Connection connection,
             JdbcNamedRelationHandle baseRelation,
-            TupleDomain<ColumnHandle> tupleDomain)
+            TupleDomain<ColumnHandle> tupleDomain,
+            Optional<String> additionalPredicate)
     {
         String sql = "DELETE FROM " + getRelation(client, baseRelation.getRemoteTableName());
 
         ImmutableList.Builder<QueryParameter> accumulator = ImmutableList.builder();
 
         List<String> clauses = toConjuncts(client, session, connection, tupleDomain, accumulator::add);
+        if (additionalPredicate.isPresent()) {
+            clauses = ImmutableList.<String>builder()
+                    .addAll(clauses)
+                    .add(additionalPredicate.get())
+                    .build();
+        }
         if (!clauses.isEmpty()) {
             sql += " WHERE " + Joiner.on(" AND ").join(clauses);
         }

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/QueryBuilder.java

@@ -54,7 +54,8 @@ PreparedQuery prepareDeleteQuery(
             ConnectorSession session,
             Connection connection,
             JdbcNamedRelationHandle baseRelation,
-            TupleDomain<ColumnHandle> tupleDomain);
+            TupleDomain<ColumnHandle> tupleDomain,
+            Optional<String> additionalPredicate);
 
     PreparedStatement prepareStatement(
             JdbcClient client,

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcConnectorTest.java

@@ -34,6 +34,7 @@
 import static java.lang.String.format;
 import static java.util.Locale.ENGLISH;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 // Single-threaded because H2 DDL operations can sometimes take a global lock, leading to apparent deadlocks
 // like in https://github.com/trinodb/trino/issues/7209.
@@ -124,6 +125,13 @@ protected Optional<DataMappingTestSetup> filterDataMappingSmokeTestData(DataMapp
         return Optional.of(dataMappingTestSetup);
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testUnknownTypeAsIgnored()
     {

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

@@ -1256,6 +1256,13 @@ public void testDelete()
         }
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("Delete without primary key or partition key is not supported");
+    }
+
     @Override
     public void testDeleteWithComplexPredicate()
     {

plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java

@@ -248,6 +248,13 @@ public void testDelete()
                 .hasStackTraceContaining("Deletes must match whole partitions for non-transactional tables");
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("Deletes must match whole partitions for non-transactional tables");
+    }
+
     @Override
     public void testDeleteWithComplexPredicate()
     {

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -173,6 +173,14 @@ public void testDelete()
                 .hasStackTraceContaining("This connector only supports delete where one or more identity-transformed partitions are deleted entirely");
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        // Deletes are covered with testMetadataDelete test methods
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("This connector only supports delete where one or more identity-transformed partitions are deleted entirely");
+    }
+
     @Override
     public void testDeleteWithComplexPredicate()
     {

plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java

@@ -35,6 +35,7 @@
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -196,6 +197,13 @@ public void testDropTable()
         assertFalse(getQueryRunner().tableExists(getSession(), "test_drop"));
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testViews()
     {

plugin/trino-oracle/src/test/java/io/trino/plugin/oracle/BaseOracleConnectorTest.java

@@ -35,6 +35,7 @@
 import static java.lang.String.format;
 import static java.util.Locale.ENGLISH;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -306,6 +307,13 @@ public void testDropTable()
         assertFalse(getQueryRunner().tableExists(getSession(), tableName));
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testViews()
     {

plugin/trino-phoenix/src/test/java/io/trino/plugin/phoenix/TestPhoenixConnectorTest.java

@@ -290,6 +290,13 @@ public void testCountDistinctWithStringTypes()
         }
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testSchemaOperations()
     {

plugin/trino-phoenix5/src/test/java/io/trino/plugin/phoenix5/TestPhoenixConnectorTest.java

@@ -308,6 +308,13 @@ public void testCountDistinctWithStringTypes()
         }
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testSchemaOperations()
     {

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java

@@ -814,7 +814,13 @@ public OptionalLong delete(ConnectorSession session, JdbcTableHandle handle)
         checkArgument(handle.getSortOrder().isEmpty(), "Unable to delete when sort order is set: %s", handle);
         try (Connection connection = connectionFactory.openConnection(session)) {
             verify(connection.getAutoCommit());
-            PreparedQuery preparedQuery = queryBuilder.prepareDeleteQuery(this, session, connection, handle.getRequiredNamedRelation(), handle.getConstraint());
+            PreparedQuery preparedQuery = queryBuilder.prepareDeleteQuery(
+                    this,
+                    session,
+                    connection,
+                    handle.getRequiredNamedRelation(),
+                    handle.getConstraint(),
+                    getAdditionalPredicate(handle.getConstraintExpressions(), Optional.empty()));
             try (PreparedStatement preparedStatement = queryBuilder.prepareStatement(this, session, connection, preparedQuery)) {
                 int affectedRowsCount = preparedStatement.executeUpdate();
                 // In getPreparedStatement we set autocommit to false so here we need an explicit commit

plugin/trino-singlestore/src/test/java/io/trino/plugin/singlestore/TestSingleStoreConnectorTest.java

@@ -176,6 +176,13 @@ public void testDropTable()
         assertFalse(getQueryRunner().tableExists(getSession(), "test_drop"));
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test
     public void testReadFromView()
     {

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/BaseSqlServerConnectorTest.java

@@ -46,6 +46,7 @@
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
@@ -384,6 +385,13 @@ public void testShowCreateTable()
                         ")");
     }
 
+    @Override
+    public void testDeleteWithLike()
+    {
+        assertThatThrownBy(super::testDeleteWithLike)
+                .hasStackTraceContaining("TrinoException: Unsupported delete");
+    }
+
     @Test(dataProvider = "dataCompression")
     public void testCreateWithDataCompression(DataCompression dataCompression)
     {

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java

@@ -2336,6 +2336,17 @@ public void testDelete()
         }
     }
 
+    @Test
+    public void testDeleteWithLike()
+    {
+        skipTestUnlessSupportsDeletes();
+
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_with_like_", "AS SELECT * FROM nation")) {
+            assertUpdate("DELETE FROM " + table.getName() + " WHERE name LIKE '%a%'", "VALUES 0");
+            assertUpdate("DELETE FROM " + table.getName() + " WHERE name LIKE '%A%'", "SELECT count(*) FROM nation WHERE name LIKE '%A%'");
+        }
+    }
+
     @Test
     public void testDeleteWithComplexPredicate()
     {