From ff1f79297a6594d078ddc496fc907036616b3d91 Mon Sep 17 00:00:00 2001
From: findinpath <findinpath@gmail.com>
Date: Tue, 25 Jan 2022 20:40:55 +0100
Subject: [PATCH] Deny DML commands on ACID tables within explicit Trino
 transactions

---
 .../io/trino/plugin/hive/HiveMetadata.java    | 10 +++++-
 .../hive/TestHiveTransactionalTable.java      | 33 +++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
index 92f4d86e38c8..2026d70a0011 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
@@ -1624,6 +1624,9 @@ public ConnectorTableHandle beginUpdate(ConnectorSession session, ConnectorTable
         if (!isFullAcidTable(table.getParameters())) {
             throw new TrinoException(NOT_SUPPORTED, "Hive update is only supported for ACID transactional tables");
         }
+        if (!autoCommit) {
+            throw new TrinoException(NOT_SUPPORTED, "Updating transactional tables is not supported in explicit transactions (use autocommit mode)");
+        }
 
         // Verify that none of the updated columns are partition columns or bucket columns
 
@@ -1723,7 +1726,9 @@ public HiveInsertTableHandle beginInsert(ConnectorSession session, ConnectorTabl
         if (isTransactional && retryMode != NO_RETRIES) {
             throw new TrinoException(NOT_SUPPORTED, "Inserting into Hive transactional tables is not supported with query retries enabled");
         }
-
+        if (isTransactional && !autoCommit) {
+            throw new TrinoException(NOT_SUPPORTED, "Inserting into Hive transactional tables is not supported in explicit transactions (use autocommit mode)");
+        }
         List<HiveColumnHandle> handles = hiveColumnHandles(table, typeManager, getTimestampPrecision(session)).stream()
                 .filter(columnHandle -> !columnHandle.isHidden())
                 .collect(toImmutableList());
@@ -2425,6 +2430,9 @@ public ConnectorTableHandle beginDelete(ConnectorSession session, ConnectorTable
         if (retryMode != NO_RETRIES) {
             throw new TrinoException(NOT_SUPPORTED, "Deleting from Hive tables is not supported with query retries enabled");
         }
+        if (!autoCommit) {
+            throw new TrinoException(NOT_SUPPORTED, "Deleting from Hive transactional tables is not supported in explicit transactions (use autocommit mode)");
+        }
 
         LocationHandle locationHandle = locationService.forExistingTable(metastore, session, table);
 
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTransactionalTable.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTransactionalTable.java
index bec4765e6c44..d643db9d9d00 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTransactionalTable.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTransactionalTable.java
@@ -950,6 +950,39 @@ public void testInsertOnlyMultipleWriters(boolean bucketed, Engine inserter1, En
         });
     }
 
+    @Test(groups = HIVE_TRANSACTIONAL)
+    public void testInsertFailsInExplicitTrinoTransaction()
+    {
+        withTemporaryTable("insert_fail_explicit_transaction", true, false, NONE, tableName -> {
+            onTrino().executeQuery(format("CREATE TABLE %s (a_string varchar) WITH (format = 'ORC', transactional = true)", tableName));
+            onTrino().executeQuery("START TRANSACTION");
+            assertQueryFailure(() -> onTrino().executeQuery(format("INSERT INTO %s (a_string) VALUES ('Commander Bun Bun')", tableName)))
+                    .hasMessageContaining("Inserting into Hive transactional tables is not supported in explicit transactions (use autocommit mode)");
+        });
+    }
+
+    @Test(groups = HIVE_TRANSACTIONAL)
+    public void testUpdateFailsInExplicitTrinoTransaction()
+    {
+        withTemporaryTable("update_fail_explicit_transaction", true, false, NONE, tableName -> {
+            onTrino().executeQuery(format("CREATE TABLE %s (a_string varchar) WITH (format = 'ORC', transactional = true)", tableName));
+            onTrino().executeQuery("START TRANSACTION");
+            assertQueryFailure(() -> onTrino().executeQuery(format("UPDATE %s SET a_string = 'Commander Bun Bun'", tableName)))
+                    .hasMessageContaining("Updating transactional tables is not supported in explicit transactions (use autocommit mode)");
+        });
+    }
+
+    @Test(groups = HIVE_TRANSACTIONAL)
+    public void testDeleteFailsInExplicitTrinoTransaction()
+    {
+        withTemporaryTable("delete_fail_explicit_transaction", true, false, NONE, tableName -> {
+            onTrino().executeQuery(format("CREATE TABLE %s (a_string varchar) WITH (format = 'ORC', transactional = true)", tableName));
+            onTrino().executeQuery("START TRANSACTION");
+            assertQueryFailure(() -> onTrino().executeQuery(format("DELETE FROM %s WHERE a_string = 'Commander Bun Bun'", tableName)))
+                    .hasMessageContaining("Deleting from Hive transactional tables is not supported in explicit transactions (use autocommit mode)");
+        });
+    }
+
     @Test(groups = HIVE_TRANSACTIONAL, dataProvider = "transactionModeProvider")
     public void testColumnRenamesOrcPartitioned(boolean transactional)
     {