Simplify transactions in JdbcPageSink

Explicit transaction management was unnecessary there, since there was
always exactly one statement (`executeBatch`) executed within each
transaction. Leveraging auto-commit allows for simpler code.

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

@@ -57,14 +57,6 @@ public JdbcPageSink(ConnectorSession session, JdbcOutputTableHandle handle, Jdbc
             throw new TrinoException(JDBC_ERROR, e);
         }
 
-        try {
-            connection.setAutoCommit(false);
-        }
-        catch (SQLException e) {
-            closeWithSuppression(connection, e);
-            throw new TrinoException(JDBC_ERROR, e);
-        }
-
         columnTypes = handle.getColumnTypes();
 
         if (handle.getJdbcColumnTypes().isEmpty()) {
@@ -92,6 +84,8 @@ public JdbcPageSink(ConnectorSession session, JdbcOutputTableHandle handle, Jdbc
         }
 
         try {
+            // Per JDBC specification, auto-commit mode is the default. Verify that in case pooling or custom ConnectionFactory is used.
+            verify(connection.getAutoCommit(), "Connection not in auto-commit");
             statement = connection.prepareStatement(jdbcClient.buildInsertSql(handle, columnWriters));
         }
         catch (SQLException e) {
@@ -114,8 +108,6 @@ public CompletableFuture<?> appendPage(Page page)
 
                 if (batchSize >= 1000) {
                     statement.executeBatch();
-                    connection.commit();
-                    connection.setAutoCommit(false);
                     batchSize = 0;
                 }
             }
@@ -165,7 +157,6 @@ public CompletableFuture<Collection<Slice>> finish()
                 PreparedStatement statement = this.statement) {
             if (batchSize > 0) {
                 statement.executeBatch();
-                connection.commit();
             }
         }
         catch (SQLNonTransientException e) {
@@ -186,17 +177,12 @@ public CompletableFuture<Collection<Slice>> finish()
         return completedFuture(ImmutableList.of());
     }
 
-    @SuppressWarnings("unused")
     @Override
     public void abort()
     {
-        // rollback and close
-        try (Connection connection = this.connection;
-                PreparedStatement statement = this.statement) {
-            // skip rollback if implicitly closed due to an error
-            if (!connection.isClosed()) {
-                connection.rollback();
-            }
+        // close statement and connection
+        try (connection) {
+            statement.close();
         }
         catch (SQLException e) {
             throw new TrinoException(JDBC_ERROR, e);