Rename rollback action consumer to collector

findepi · findepi · commit dfd75cf4f92c · 2026-01-09T09:06:25.000+01:00
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
@@ -829,13 +829,13 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
         try {
             if (shouldUseFaultTolerantExecution(session)) {
                 // Create the target table
                 JdbcOutputTableHandle destinationTableHandle = createTable(session, tableMetadata, tableMetadata.getTable().getTableName());
-                rollbackActionConsumer.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));
+                rollbackActionCollector.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));
                 // Create the temporary table
                 ColumnMetadata pageSinkIdColumn = getPageSinkIdColumn(
                         tableMetadata.getColumns().stream().map(ColumnMetadata::getName).toList());
@@ -1219,7 +1219,7 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
         if (!supportsMerge()) {
@@ -1239,14 +1239,14 @@ public JdbcMergeTableHandle beginMerge(
         JdbcTableHandle plainTable = new JdbcTableHandle(schemaTableName, remoteTableName, Optional.empty());
 
         JdbcOutputTableHandle outputTableHandle = beginInsertTable(session, plainTable, columns);
-        rollbackActionConsumer.accept(() -> rollbackCreateTable(session, outputTableHandle));
+        rollbackActionCollector.accept(() -> rollbackCreateTable(session, outputTableHandle));
 
         try {
             return new JdbcMergeTableHandle(
                     handle,
                     outputTableHandle,
-                    beginUpdate(session, plainTable, columns, primaryKeys, updateColumnHandles, rollbackActionConsumer),
-                    beginDelete(session, plainTable, primaryKeys, rollbackActionConsumer),
+                    beginUpdate(session, plainTable, columns, primaryKeys, updateColumnHandles, rollbackActionCollector),
+                    beginDelete(session, plainTable, primaryKeys, rollbackActionCollector),
                     primaryKeys,
                     columns,
                     updateColumnHandles);
@@ -1262,7 +1262,7 @@ protected Map<Integer, JdbcOutputTableHandle> beginUpdate(
             List<JdbcColumnHandle> columns,
             List<JdbcColumnHandle> primaryKeys,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer)
+            Consumer<Runnable> rollbackActionCollector)
             throws SQLException
     {
         if (isNonTransactionalMerge(session)) {
@@ -1301,7 +1301,7 @@ protected Map<Integer, JdbcOutputTableHandle> beginUpdate(
                     generateTemporaryTableName(session),
                     Optional.of(getPageSinkIdColumn(updatedColumnNames)));
 
-            rollbackActionConsumer.accept(() -> rollbackCreateTable(session, temporaryTableHandle));
+            rollbackActionCollector.accept(() -> rollbackCreateTable(session, temporaryTableHandle));
             outputHandles.put(caseNumber, temporaryTableHandle);
         }
 
@@ -1312,14 +1312,14 @@ protected Optional<JdbcOutputTableHandle> beginDelete(
             ConnectorSession session,
             ConnectorTableHandle tableHandle,
             List<JdbcColumnHandle> primaryKeys,
-            Consumer<Runnable> rollbackActionConsumer)
+            Consumer<Runnable> rollbackActionCollector)
     {
         if (isNonTransactionalMerge(session)) {
             return Optional.empty();
         }
 
         JdbcOutputTableHandle handle = beginInsertTable(session, (JdbcTableHandle) tableHandle, primaryKeys);
-        rollbackActionConsumer.accept(() -> rollbackCreateTable(session, handle));
+        rollbackActionCollector.accept(() -> rollbackCreateTable(session, handle));
         return Optional.of(handle);
     }
 
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
@@ -425,10 +425,10 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
-        return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);
+        return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);
     }
 
     @Override
@@ -600,9 +600,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
-        return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
+        return delegate.beginCreateTable(session, tableMetadata, rollbackActionCollector);
     }
 
     @Override
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, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
-        return delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer);
+        return delegate().beginCreateTable(session, tableMetadata, rollbackActionCollector);
     }
 
     @Override
@@ -293,10 +293,10 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
-        return delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);
+        return delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);
     }
 
     @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
@@ -210,7 +210,7 @@ default void setTableProperties(ConnectorSession session, JdbcTableHandle handle
 
     void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata);
 
-    JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer);
+    JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector);
 
     void commitCreateTable(ConnectorSession session, JdbcOutputTableHandle handle, Set<Long> pageSinkIds);
 
@@ -222,7 +222,7 @@ default JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
         throw new TrinoException(NOT_SUPPORTED, MODIFYING_ROWS_MESSAGE);
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
@@ -368,10 +368,10 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
         // no retrying as it could be not idempotent operation
-        return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
+        return delegate.beginCreateTable(session, tableMetadata, rollbackActionCollector);
     }
 
     @Override
@@ -400,11 +400,11 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
         // no retrying as it could be not idempotent operation
-        return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);
+        return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);
     }
 
     @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
@@ -353,9 +353,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
-        return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer));
+        return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionCollector));
     }
 
     @Override
@@ -381,10 +381,10 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
-        return stats.getBeginMergeTable().wrap(() -> delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode));
+        return stats.getBeginMergeTable().wrap(() -> delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode));
     }
 
     @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
@@ -477,7 +477,7 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
         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
@@ -315,7 +315,7 @@ public WriteMapping toWriteMapping(ConnectorSession session, Type type)
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
         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
@@ -382,7 +382,7 @@ private static void checkDateValue(long epochDay)
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
         if (tableMetadata.getComment().isPresent()) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
@@ -424,7 +424,7 @@ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, Connecto
                     columnTypes.build(),
                     Optional.empty(),
                     primaryKeys.isEmpty() ? Optional.of(IGNITE_DUMMY_ID) : Optional.empty());
-            rollbackActionConsumer.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));
+            rollbackActionCollector.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));
             return destinationTableHandle;
         }
         catch (SQLException e) {
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
@@ -1076,7 +1076,7 @@ public JdbcMergeTableHandle beginMerge(
             ConnectorSession session,
             JdbcTableHandle handle,
             Map<Integer, Collection<ColumnHandle>> updateColumnHandles,
-            Consumer<Runnable> rollbackActionConsumer,
+            Consumer<Runnable> rollbackActionCollector,
             RetryMode retryMode)
     {
         if (retryMode != NO_RETRIES) {
@@ -1087,7 +1087,7 @@ public JdbcMergeTableHandle beginMerge(
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support MERGE with transactional execution");
         }
 
-        return super.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);
+        return super.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);
     }
 
     @Override
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
@@ -364,9 +364,9 @@ protected void dropSchema(ConnectorSession session, Connection connection, Strin
     }
 
     @Override
-    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)
+    public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)
     {
-        JdbcOutputTableHandle table = super.beginCreateTable(session, tableMetadata, rollbackActionConsumer);
+        JdbcOutputTableHandle table = super.beginCreateTable(session, tableMetadata, rollbackActionCollector);
         enableTableLockOnBulkLoadTableOption(session, table);
         return table;
     }

Original file line number	Diff line number	Diff line change
`@@ -425,10 +425,10 @@ public JdbcMergeTableHandle beginMerge(`
`425`	`425`	`ConnectorSession session,`
`426`	`426`	`JdbcTableHandle handle,`
`427`	`427`	`Map<Integer, Collection<ColumnHandle>> updateColumnHandles,`
`428`		`- Consumer<Runnable> rollbackActionConsumer,`
	`428`	`+ Consumer<Runnable> rollbackActionCollector,`
`429`	`429`	`RetryMode retryMode)`
`430`	`430`	`{`
`431`		`- return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);`
	`431`	`+ return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);`
`432`	`432`	`}`
`433`	`433`
`434`	`434`	`@Override`
`@@ -600,9 +600,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe`
`600`	`600`	`}`
`601`	`601`
`602`	`602`	`@Override`
`603`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`603`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`604`	`604`	`{`
`605`		`- return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
	`605`	`+ return delegate.beginCreateTable(session, tableMetadata, rollbackActionCollector);`
`606`	`606`	`}`
`607`	`607`
`608`	`608`	`@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, Consumer<Runnable> rollbackActionConsumer)`
	`268`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`269`	`269`	`{`
`270`		`- return delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
	`270`	`+ return delegate().beginCreateTable(session, tableMetadata, rollbackActionCollector);`
`271`	`271`	`}`
`272`	`272`
`273`	`273`	`@Override`
`@@ -293,10 +293,10 @@ public JdbcMergeTableHandle beginMerge(`
`293`	`293`	`ConnectorSession session,`
`294`	`294`	`JdbcTableHandle handle,`
`295`	`295`	`Map<Integer, Collection<ColumnHandle>> updateColumnHandles,`
`296`		`- Consumer<Runnable> rollbackActionConsumer,`
	`296`	`+ Consumer<Runnable> rollbackActionCollector,`
`297`	`297`	`RetryMode retryMode)`
`298`	`298`	`{`
`299`		`- return delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);`
	`299`	`+ return delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);`
`300`	`300`	`}`
`301`	`301`
`302`	`302`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -368,10 +368,10 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe`
`368`	`368`	`}`
`369`	`369`
`370`	`370`	`@Override`
`371`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`371`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`372`	`372`	`{`
`373`	`373`	`// no retrying as it could be not idempotent operation`
`374`		`- return delegate.beginCreateTable(session, tableMetadata, rollbackActionConsumer);`
	`374`	`+ return delegate.beginCreateTable(session, tableMetadata, rollbackActionCollector);`
`375`	`375`	`}`
`376`	`376`
`377`	`377`	`@Override`
`@@ -400,11 +400,11 @@ public JdbcMergeTableHandle beginMerge(`
`400`	`400`	`ConnectorSession session,`
`401`	`401`	`JdbcTableHandle handle,`
`402`	`402`	`Map<Integer, Collection<ColumnHandle>> updateColumnHandles,`
`403`		`- Consumer<Runnable> rollbackActionConsumer,`
	`403`	`+ Consumer<Runnable> rollbackActionCollector,`
`404`	`404`	`RetryMode retryMode)`
`405`	`405`	`{`
`406`	`406`	`// no retrying as it could be not idempotent operation`
`407`		`- return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);`
	`407`	`+ return delegate.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);`
`408`	`408`	`}`
`409`	`409`
`410`	`410`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -353,9 +353,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe`
`353`	`353`	`}`
`354`	`354`
`355`	`355`	`@Override`
`356`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`356`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`357`	`357`	`{`
`358`		`- return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionConsumer));`
	`358`	`+ return stats.getBeginCreateTable().wrap(() -> delegate().beginCreateTable(session, tableMetadata, rollbackActionCollector));`
`359`	`359`	`}`
`360`	`360`
`361`	`361`	`@Override`
`@@ -381,10 +381,10 @@ public JdbcMergeTableHandle beginMerge(`
`381`	`381`	`ConnectorSession session,`
`382`	`382`	`JdbcTableHandle handle,`
`383`	`383`	`Map<Integer, Collection<ColumnHandle>> updateColumnHandles,`
`384`		`- Consumer<Runnable> rollbackActionConsumer,`
	`384`	`+ Consumer<Runnable> rollbackActionCollector,`
`385`	`385`	`RetryMode retryMode)`
`386`	`386`	`{`
`387`		`- return stats.getBeginMergeTable().wrap(() -> delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode));`
	`387`	`+ return stats.getBeginMergeTable().wrap(() -> delegate().beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode));`
`388`	`388`	`}`
`389`	`389`
`390`	`390`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -477,7 +477,7 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe`
`477`	`477`	`}`
`478`	`478`
`479`	`479`	`@Override`
`480`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`480`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`481`	`481`	`{`
`482`	`482`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");`
`483`	`483`	`}`
Original file line number	Diff line number	Diff line change
`@@ -315,7 +315,7 @@ public WriteMapping toWriteMapping(ConnectorSession session, Type type)`
`315`	`315`	`}`
`316`	`316`
`317`	`317`	`@Override`
`318`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`318`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`319`	`319`	`{`
`320`	`320`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with data");`
`321`	`321`	`}`
Original file line number	Diff line number	Diff line change
`@@ -382,7 +382,7 @@ private static void checkDateValue(long epochDay)`
`382`	`382`	`}`
`383`	`383`
`384`	`384`	`@Override`
`385`		`- public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionConsumer)`
	`385`	`+ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Consumer<Runnable> rollbackActionCollector)`
`386`	`386`	`{`
`387`	`387`	`if (tableMetadata.getComment().isPresent()) {`
`388`	`388`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");`
`@@ -424,7 +424,7 @@ public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, Connecto`
`424`	`424`	`columnTypes.build(),`
`425`	`425`	`Optional.empty(),`
`426`	`426`	`primaryKeys.isEmpty() ? Optional.of(IGNITE_DUMMY_ID) : Optional.empty());`
`427`		`- rollbackActionConsumer.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));`
	`427`	`+ rollbackActionCollector.accept(() -> rollbackCreateDestinationTable(session, destinationTableHandle.getRemoteTableName()));`
`428`	`428`	`return destinationTableHandle;`
`429`	`429`	`}`
`430`	`430`	`catch (SQLException e) {`
Original file line number	Diff line number	Diff line change
`@@ -1076,7 +1076,7 @@ public JdbcMergeTableHandle beginMerge(`
`1076`	`1076`	`ConnectorSession session,`
`1077`	`1077`	`JdbcTableHandle handle,`
`1078`	`1078`	`Map<Integer, Collection<ColumnHandle>> updateColumnHandles,`
`1079`		`- Consumer<Runnable> rollbackActionConsumer,`
	`1079`	`+ Consumer<Runnable> rollbackActionCollector,`
`1080`	`1080`	`RetryMode retryMode)`
`1081`	`1081`	`{`
`1082`	`1082`	`if (retryMode != NO_RETRIES) {`
`@@ -1087,7 +1087,7 @@ public JdbcMergeTableHandle beginMerge(`
`1087`	`1087`	`throw new TrinoException(NOT_SUPPORTED, "This connector does not support MERGE with transactional execution");`
`1088`	`1088`	`}`
`1089`	`1089`
`1090`		`- return super.beginMerge(session, handle, updateColumnHandles, rollbackActionConsumer, retryMode);`
	`1090`	`+ return super.beginMerge(session, handle, updateColumnHandles, rollbackActionCollector, retryMode);`
`1091`	`1091`	`}`
`1092`	`1092`
`1093`	`1093`	`@Override`