Add rollback action for destination table for CTAS statements

findinpath · findinpath · commit 540828e8d6e4 · 2025-12-23T09:45:28.000+01:00
While CREATE TABLE is an atomic operation,
CREATE TABLE AS SELECT statement happens in two phases
for JDBC connectors from Trino:

- creating the table
- inserting content into the table

If the insertion of the data fails, the CTAS may leave
residual empty tables behind.

The JDBC connectors, in dealing
with CREATE TABLE AS SELECT scenarios,
remove the destination table in case of dealing with exceptions
while inserting the content.
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java
@@ -818,23 +818,24 @@ protected PreparedQuery applyQueryTransformations(JdbcTableHandle tableHandle, P
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         try {
-            createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
+            return createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
         }
         catch (SQLException e) {
             throw new TrinoException(JDBC_ERROR, e);
         }
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
         try {
             if (shouldUseFaultTolerantExecution(session)) {
                 // Create the target table
-                createTable(session, tableMetadata);
+                JdbcOutputTableHandle destinationTableHandle = createTable(session, tableMetadata);
+                rollbackActionConsumer.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle));
                 // Create the temporary table
                 ColumnMetadata pageSinkIdColumn = getPageSinkIdColumn(
                         tableMetadata.getColumns().stream().map(ColumnMetadata::getName).toList());
@@ -849,6 +850,11 @@ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, Connecto
         }
     }
 
+    protected void rollbackCreateDestinationTable(ConnectorSession session, JdbcOutputTableHandle handle)
+    {
+        dropTable(session, handle.getRemoteTableName(), false);
+    }
+
     protected JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, String targetTableName)
             throws SQLException
     {
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java
@@ -593,16 +593,17 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
-        delegate.createTable(session, tableMetadata);
+        JdbcOutputTableHandle outputTableHandle = delegate.createTable(session, tableMetadata);
         invalidateTableCaches(tableMetadata.getTable());
+        return outputTableHandle;
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
-        return delegate.beginCreateTable(session, tableMetadata);
+        return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
     }
 
     @Override
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java
@@ -137,7 +137,7 @@ public class DefaultJdbcMetadata
     private final boolean precalculateStatisticsForPushdown;
     private final Set<JdbcQueryEventListener> jdbcQueryEventListeners;
 
-    private final List<Runnable> rollbackActions = new ArrayList<>();
+    protected final List<Runnable> rollbackActions = new ArrayList<>();
 
     public DefaultJdbcMetadata(
             JdbcClient jdbcClient,
@@ -1199,7 +1199,7 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
         if (replace) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables");
         }
-        JdbcOutputTableHandle handle = jdbcClient.beginCreateTable(session, tableMetadata);
+        JdbcOutputTableHandle handle = jdbcClient.beginCreateTable(session, tableMetadata, rollbackActions::add);
         rollbackActions.add(() -> jdbcClient.rollbackCreateTable(session, handle));
         return handle;
     }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java
@@ -265,9 +265,9 @@ public Optional<PreparedQuery> legacyImplementJoin(
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
-        return delegate().beginCreateTable(session, tableMetadata);
+        return delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer);
     }
 
     @Override
@@ -448,9 +448,9 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
-        delegate().createTable(session, tableMetadata);
+        return delegate().createTable(session, tableMetadata);
     }
 
     @Override
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java
@@ -208,9 +208,9 @@ default void setTableProperties(ConnectorSession session, JdbcTableHandle handle
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support setting table properties");
     }
 
-    void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);
+    JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);
 
-    JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);
+    JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer);
 
     void commitCreateTable(ConnectorSession session, JdbcOutputTableHandle handle, Set<Long> pageSinkIds);
 
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java
@@ -361,17 +361,17 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         // no retrying as it could be not idempotent operation
-        delegate.createTable(session, tableMetadata);
+        return delegate.createTable(session, tableMetadata);
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
         // no retrying as it could be not idempotent operation
-        return delegate.beginCreateTable(session, tableMetadata);
+        return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
     }
 
     @Override
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java
@@ -347,15 +347,15 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
-        stats.getCreateTable().wrap(() -> delegate().createTable(session, tableMetadata));
+        return stats.getCreateTable().wrap(() -> delegate().createTable(session, tableMetadata));
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
-        return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata));
+        return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer));
     }
 
     @Override
diff --git a/plugin/trino-druid/src/main/java/io/trino/plugin/druid/DruidJdbcClient.java b/plugin/trino-druid/src/main/java/io/trino/plugin/druid/DruidJdbcClient.java
@@ -74,6 +74,7 @@
 import java.util.Set;
 import java.util.TimeZone;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.Iterables.getOnlyElement;
@@ -470,13 +471,13 @@ public OptionalLong update(ConnectorSession session, JdbcTableHandle handle)
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables");
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");
     }
diff --git a/plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolClient.java b/plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolClient.java
@@ -56,6 +56,7 @@
 import java.util.OptionalLong;
 import java.util.Set;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 
 import static io.trino.plugin.jdbc.StandardColumnMappings.bigintColumnMapping;
 import static io.trino.plugin.jdbc.StandardColumnMappings.booleanColumnMapping;
@@ -117,7 +118,7 @@ public void truncateTable(ConnectorSession session, JdbcTableHandle handle)
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables");
     }
@@ -314,7 +315,7 @@ public WriteMapping toWriteMapping(ConnectorSession session, Type type)
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");
     }
diff --git a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteClient.java b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteClient.java
@@ -83,6 +83,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Strings.isNullOrEmpty;
@@ -381,7 +382,7 @@ private static void checkDateValue(long epochDay)
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
         if (tableMetadata.getComment().isPresent()) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
diff --git a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java
@@ -220,7 +220,7 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
         if (saveMode == REPLACE) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables");
         }
-        igniteClient.beginCreateTable(session, tableMetadata);
+        igniteClient.beginCreateTable(session, tableMetadata, _ -> {});
     }
 
     @Override
@@ -232,7 +232,7 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
         if (replace) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables");
         }
-        return igniteClient.beginCreateTable(session, tableMetadata);
+        return igniteClient.beginCreateTable(session, tableMetadata, rollbackActions::add);
     }
 
     @Override
diff --git a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
@@ -36,6 +36,7 @@
 import io.trino.plugin.jdbc.JdbcJoinCondition;
 import io.trino.plugin.jdbc.JdbcMergeTableHandle;
 import io.trino.plugin.jdbc.JdbcMetadata;
+import io.trino.plugin.jdbc.JdbcOutputTableHandle;
 import io.trino.plugin.jdbc.JdbcSortItem;
 import io.trino.plugin.jdbc.JdbcStatisticsConfig;
 import io.trino.plugin.jdbc.JdbcTableHandle;
@@ -931,10 +932,10 @@ else if (varcharType.getBoundedLength() <= 16777215) {
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         try {
-            createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
+            return createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
         }
         catch (SQLException e) {
             if (e.getErrorCode() == ER_TABLE_EXISTS_ERROR) {
diff --git a/plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java b/plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java
@@ -391,10 +391,10 @@ public PostgreSqlClient(
     }
 
     @Override
-    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         try {
-            createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
+            return createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
         }
         catch (SQLException e) {
             boolean exists = DUPLICATE_TABLE_SQLSTATE.equals(e.getSQLState());
diff --git a/plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java b/plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java
@@ -129,6 +129,7 @@
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 import static com.google.common.base.MoreObjects.firstNonNull;
@@ -363,9 +364,9 @@ protected void dropSchema(ConnectorSession session, Connection connection, Strin
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
     {
-        JdbcOutputTableHandle table = super.beginCreateTable(session, tableMetadata);
+        JdbcOutputTableHandle table = super.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
         enableTableLockOnBulkLoadTableOption(session, table);
         return table;
     }

Original file line number	Diff line number	Diff line change
`@@ -818,23 +818,24 @@ protected PreparedQuery applyQueryTransformations(JdbcTableHandle tableHandle, P`
`818`	`818`	`}`
`819`	`819`
`820`	`820`	`@Override`
`821`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`821`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`822`	`822`	`{`
`823`	`823`	`try {`
`824`		`- createTable(session, tableMetadata, tableMetadata.getTable().getTableName());`
	`824`	`+ return createTable(session, tableMetadata, tableMetadata.getTable().getTableName());`
`825`	`825`	`}`
`826`	`826`	`catch (SQLException e) {`
`827`	`827`	`throw new TrinoException(JDBC_ERROR, e);`
`828`	`828`	`}`
`829`	`829`	`}`
`830`	`830`
`831`	`831`	`@Override`
`832`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`832`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`833`	`833`	`{`
`834`	`834`	`try {`
`835`	`835`	`if (shouldUseFaultTolerantExecution(session)) {`
`836`	`836`	`// Create the target table`
`837`		`- createTable(session, tableMetadata);`
	`837`	`+ JdbcOutputTableHandle destinationTableHandle = createTable(session, tableMetadata);`
	`838`	`+ rollbackActionConsumer.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle));`
`838`	`839`	`// Create the temporary table`
`839`	`840`	`ColumnMetadata pageSinkIdColumn = getPageSinkIdColumn(`
`840`	`841`	`tableMetadata.getColumns().stream().map(ColumnMetadata::getName).toList());`
`@@ -849,6 +850,11 @@ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, Connecto`
`849`	`850`	`}`
`850`	`851`	`}`
`851`	`852`
	`853`	`+ protected void rollbackCreateDestinationTable(ConnectorSession session, JdbcOutputTableHandle handle)`
	`854`	`+ {`
	`855`	`+ dropTable(session, handle.getRemoteTableName(), false);`
	`856`	`+ }`
	`857`	`+`
`852`	`858`	`protected JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, String targetTableName)`
`853`	`859`	`throws SQLException`
`854`	`860`	`{`
Original file line number	Diff line number	Diff line change
`@@ -593,16 +593,17 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,`
`593`	`593`	`}`
`594`	`594`
`595`	`595`	`@Override`
`596`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`596`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`597`	`597`	`{`
`598`		`- delegate.createTable(session, tableMetadata);`
	`598`	`+ JdbcOutputTableHandle outputTableHandle = delegate.createTable(session, tableMetadata);`
`599`	`599`	`invalidateTableCaches(tableMetadata.getTable());`
	`600`	`+ return outputTableHandle;`
`600`	`601`	`}`
`601`	`602`
`602`	`603`	`@Override`
`603`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`604`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`604`	`605`	`{`
`605`		`- return delegate.beginCreateTable(session, tableMetadata);`
	`606`	`+ return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
`606`	`607`	`}`
`607`	`608`
`608`	`609`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -265,9 +265,9 @@ public Optional<PreparedQuery> legacyImplementJoin(`
`265`	`265`	`}`
`266`	`266`
`267`	`267`	`@Override`
`268`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`268`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`269`	`269`	`{`
`270`		`- return delegate().beginCreateTable(session, tableMetadata);`
	`270`	`+ return delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
`271`	`271`	`}`
`272`	`272`
`273`	`273`	`@Override`
`@@ -448,9 +448,9 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,`
`448`	`448`	`}`
`449`	`449`
`450`	`450`	`@Override`
`451`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`451`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`452`	`452`	`{`
`453`		`- delegate().createTable(session, tableMetadata);`
	`453`	`+ return delegate().createTable(session, tableMetadata);`
`454`	`454`	`}`
`455`	`455`
`456`	`456`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -208,9 +208,9 @@ default void setTableProperties(ConnectorSession session, JdbcTableHandle handle`
`208`	`208`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support setting table properties");`
`209`	`209`	`}`
`210`	`210`
`211`		`- void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);`
	`211`	`+ JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);`
`212`	`212`
`213`		`- JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);`
	`213`	`+ JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer);`
`214`	`214`
`215`	`215`	`void commitCreateTable(ConnectorSession session, JdbcOutputTableHandle handle, Set<Long> pageSinkIds);`
`216`	`216`
Original file line number	Diff line number	Diff line change
`@@ -361,17 +361,17 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,`
`361`	`361`	`}`
`362`	`362`
`363`	`363`	`@Override`
`364`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`364`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`365`	`365`	`{`
`366`	`366`	`// no retrying as it could be not idempotent operation`
`367`		`- delegate.createTable(session, tableMetadata);`
	`367`	`+ return delegate.createTable(session, tableMetadata);`
`368`	`368`	`}`
`369`	`369`
`370`	`370`	`@Override`
`371`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`371`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`372`	`372`	`{`
`373`	`373`	`// no retrying as it could be not idempotent operation`
`374`		`- return delegate.beginCreateTable(session, tableMetadata);`
	`374`	`+ return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
`375`	`375`	`}`
`376`	`376`
`377`	`377`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -347,15 +347,15 @@ public void setTableProperties(ConnectorSession session, JdbcTableHandle handle,`
`347`	`347`	`}`
`348`	`348`
`349`	`349`	`@Override`
`350`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`350`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`351`	`351`	`{`
`352`		`- stats.getCreateTable().wrap(() -> delegate().createTable(session, tableMetadata));`
	`352`	`+ return stats.getCreateTable().wrap(() -> delegate().createTable(session, tableMetadata));`
`353`	`353`	`}`
`354`	`354`
`355`	`355`	`@Override`
`356`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`356`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`357`	`357`	`{`
`358`		`- return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata));`
	`358`	`+ return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer));`
`359`	`359`	`}`
`360`	`360`
`361`	`361`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -74,6 +74,7 @@`
`74`	`74`	`import java.util.Set;`
`75`	`75`	`import java.util.TimeZone;`
`76`	`76`	`import java.util.function.BiFunction;`
	`77`	`+import java.util.function.Consumer;`
`77`	`78`
`78`	`79`	`import static com.google.common.base.Preconditions.checkArgument;`
`79`	`80`	`import static com.google.common.collect.Iterables.getOnlyElement;`
`@@ -470,13 +471,13 @@ public OptionalLong update(ConnectorSession session, JdbcTableHandle handle)`
`470`	`471`	`}`
`471`	`472`
`472`	`473`	`@Override`
`473`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`474`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`474`	`475`	`{`
`475`	`476`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables");`
`476`	`477`	`}`
`477`	`478`
`478`	`479`	`@Override`
`479`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`480`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`480`	`481`	`{`
`481`	`482`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");`
`482`	`483`	`}`
Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,7 @@`
`56`	`56`	`import java.util.OptionalLong;`
`57`	`57`	`import java.util.Set;`
`58`	`58`	`import java.util.function.BiFunction;`
	`59`	`+import java.util.function.Consumer;`
`59`	`60`
`60`	`61`	`import static io.trino.plugin.jdbc.StandardColumnMappings.bigintColumnMapping;`
`61`	`62`	`import static io.trino.plugin.jdbc.StandardColumnMappings.booleanColumnMapping;`
`@@ -117,7 +118,7 @@ public void truncateTable(ConnectorSession session, JdbcTableHandle handle)`
`117`	`118`	`}`
`118`	`119`
`119`	`120`	`@Override`
`120`		`- public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`121`	`+ public JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
`121`	`122`	`{`
`122`	`123`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables");`
`123`	`124`	`}`
`@@ -314,7 +315,7 @@ public WriteMapping toWriteMapping(ConnectorSession session, Type type)`
`314`	`315`	`}`
`315`	`316`
`316`	`317`	`@Override`
`317`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`318`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`318`	`319`	`{`
`319`	`320`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");`
`320`	`321`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,6 +83,7 @@`
`83`	`83`	`import java.util.Optional;`
`84`	`84`	`import java.util.Set;`
`85`	`85`	`import java.util.function.BiFunction;`
	`86`	`+import java.util.function.Consumer;`
`86`	`87`
`87`	`88`	`import static com.google.common.base.Preconditions.checkArgument;`
`88`	`89`	`import static com.google.common.base.Strings.isNullOrEmpty;`
`@@ -381,7 +382,7 @@ private static void checkDateValue(long epochDay)`
`381`	`382`	`}`
`382`	`383`
`383`	`384`	`@Override`
`384`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)`
	`385`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
`385`	`386`	`{`
`386`	`387`	`if (tableMetadata.getComment().isPresent()) {`
`387`	`388`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");`