Remove transaction from atomic writes

Author: singhpk234

extension/persistence/relational-jdbc/src/main/java/org/apache/polaris/extension/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -93,11 +93,7 @@ public void writeEntity(
       boolean nameOrParentChanged,
       PolarisBaseEntity originalEntity) {
     try {
-      datasourceOperations.runWithinTransaction(
-          statement -> {
-            persistEntity(callCtx, entity, originalEntity, statement);
-            return true;
-          });
+      persistEntity(callCtx, entity, originalEntity, datasourceOperations);
     } catch (SQLException e) {
       throw new RuntimeException("Error persisting entity", e);
     }
@@ -143,12 +139,12 @@ private void persistEntity(
       @Nonnull PolarisCallContext callCtx,
       @Nonnull PolarisBaseEntity entity,
       PolarisBaseEntity originalEntity,
-      Statement statement)
+      Object executor)
       throws SQLException {
     ModelEntity modelEntity = ModelEntity.fromEntity(entity);
     if (originalEntity == null) {
       try {
-        statement.executeUpdate(generateInsertQuery(modelEntity, realmId));
+        execute(executor, generateInsertQuery(modelEntity, realmId));
       } catch (SQLException e) {
         if (datasourceOperations.isConstraintViolation(e)) {
           PolarisBaseEntity existingEntity =
@@ -176,7 +172,7 @@ private void persistEntity(
               "realm_id",
               realmId);
       try {
-        int rowsUpdated = statement.executeUpdate(generateUpdateQuery(modelEntity, params));
+        int rowsUpdated = execute(executor, generateUpdateQuery(modelEntity, params));
         if (rowsUpdated == 0) {
           throw new RetryOnConcurrencyException(
               "Entity '%s' id '%s' concurrently modified; expected version %s",
@@ -189,6 +185,17 @@ private void persistEntity(
     }
   }
 
+  private int execute(Object executor, String query) throws SQLException {
+    if (executor instanceof Statement) {
+      // used for running in transaction
+      return ((Statement) executor).executeUpdate(query);
+    } else if (executor instanceof DatasourceOperations) {
+      return ((DatasourceOperations) executor).executeUpdate(query);
+    } else {
+      throw new IllegalArgumentException("Unsupported executor: " + executor);
+    }
+  }
+
   @Override
   public void writeToGrantRecords(
       @Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {