Oracle purges dropped temp tables

Author: mwd410

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

@@ -890,7 +890,7 @@ public void finishInsertTable(ConnectorSession session, JdbcOutputTableHandle ha
 
         // We conditionally create more than the one table, so keep a list of the tables that need to be dropped.
         Closer closer = Closer.create();
-        closer.register(() -> dropTable(session, temporaryTable));
+        closer.register(() -> dropTable(session, temporaryTable, true));
 
         try (Connection connection = getConnection(session, handle)) {
             verify(connection.getAutoCommit());
@@ -906,7 +906,7 @@ public void finishInsertTable(ConnectorSession session, JdbcOutputTableHandle ha
 
             if (handle.getPageSinkIdColumnName().isPresent()) {
                 RemoteTableName pageSinkTable = constructPageSinkIdsTable(session, connection, handle, pageSinkIds);
-                closer.register(() -> dropTable(session, pageSinkTable));
+                closer.register(() -> dropTable(session, pageSinkTable, true));
 
                 insertSql += format(" WHERE EXISTS (SELECT 1 FROM %s page_sink_table WHERE page_sink_table.%s = temp_table.%s)",
                         quoted(pageSinkTable),
@@ -1034,10 +1034,10 @@ public void setColumnType(ConnectorSession session, JdbcTableHandle handle, Jdbc
     public void dropTable(ConnectorSession session, JdbcTableHandle handle)
     {
         verify(handle.getAuthorization().isEmpty(), "Unexpected authorization is required for table: %s".formatted(handle));
-        dropTable(session, handle.asPlainTable().getRemoteTableName());
+        dropTable(session, handle.asPlainTable().getRemoteTableName(), false);
     }
 
-    protected void dropTable(ConnectorSession session, RemoteTableName remoteTableName)
+    protected void dropTable(ConnectorSession session, RemoteTableName remoteTableName, boolean temporaryTable)
     {
         String sql = "DROP TABLE " + quoted(remoteTableName);
         execute(session, sql);
@@ -1047,10 +1047,12 @@ protected void dropTable(ConnectorSession session, RemoteTableName remoteTableNa
     public void rollbackCreateTable(ConnectorSession session, JdbcOutputTableHandle handle)
     {
         if (handle.getTemporaryTableName().isPresent()) {
-            dropTable(session, new JdbcTableHandle(
-                    new SchemaTableName(handle.getSchemaName(), handle.getTemporaryTableName().get()),
-                    new RemoteTableName(Optional.ofNullable(handle.getCatalogName()), Optional.ofNullable(handle.getSchemaName()), handle.getTemporaryTableName().get()),
-                    Optional.empty()));
+            dropTable(session,
+                    new RemoteTableName(
+                            Optional.ofNullable(handle.getCatalogName()),
+                            Optional.ofNullable(handle.getSchemaName()),
+                            handle.getTemporaryTableName().get()),
+                    true);
         }
     }
 

plugin/trino-oracle/src/main/java/io/trino/plugin/oracle/OracleClient.java

@@ -296,6 +296,20 @@ public void dropSchema(ConnectorSession session, String schemaName)
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support dropping schemas");
     }
 
+    @Override
+    protected void dropTable(ConnectorSession session, RemoteTableName remoteTableName, boolean temporaryTable)
+    {
+        String sql = "DROP TABLE " + quoted(remoteTableName);
+        // Oracle puts dropped tables into a recycling bin, which keeps them accessible for a period of time.
+        // PURGE will bypass the bin and completely delete the table immediately.
+        // We should only PURGE the table if it is a temporary table that trino created,
+        // as purging all dropped tables may be unexpected behavior for our clients.
+        if (temporaryTable) {
+            sql += " PURGE";
+        }
+        execute(session, sql);
+    }
+
     @Override
     public void renameSchema(ConnectorSession session, String schemaName, String newSchemaName)
     {