Store commit actions in the Transaction

core/src/main/java/org/polypheny/db/catalog/Catalog.java

@@ -24,11 +24,9 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import org.apache.calcite.linq4j.tree.Expression;
 import org.apache.calcite.linq4j.tree.Expressions;
 import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 import org.pf4j.ExtensionPoint;
 import org.polypheny.db.adapter.AbstractAdapterSetting;
 import org.polypheny.db.adapter.Adapter;
@@ -51,7 +49,6 @@
 import org.polypheny.db.catalog.snapshot.Snapshot;
 import org.polypheny.db.iface.QueryInterfaceManager.QueryInterfaceTemplate;
 import org.polypheny.db.transaction.Transaction;
-import org.polypheny.db.util.Pair;
 import org.polypheny.db.util.RunMode;
 
 public abstract class Catalog implements ExtensionPoint {
@@ -111,14 +108,8 @@ public static void afterInit( Runnable action ) {
 
     public abstract String getJson();
 
-    public abstract void executeCommitActions();
-
-    public abstract void clearCommitActions();
-
     public abstract void commit();
 
-    public abstract Pair<@NotNull Boolean, @Nullable String> checkIntegrity();
-
     public abstract void rollback();
 
     public abstract LogicalRelationalCatalog getLogicalRel( long namespaceId );
@@ -282,9 +273,4 @@ public static Snapshot snapshot() {
 
     public abstract void restore( Transaction transaction );
 
-
-    public abstract void attachCommitConstraint( Supplier<Boolean> constraintChecker, String description );
-
-    public abstract void attachCommitAction( Runnable action );
-
 }

core/src/main/java/org/polypheny/db/catalog/impl/PolyCatalog.java

@@ -26,20 +26,15 @@
 import io.activej.serializer.annotations.Deserialize;
 import io.activej.serializer.annotations.Serialize;
 import java.beans.PropertyChangeListener;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 import org.polypheny.db.adapter.AbstractAdapterSetting;
 import org.polypheny.db.adapter.Adapter;
 import org.polypheny.db.adapter.AdapterManager;
@@ -79,7 +74,6 @@
 import org.polypheny.db.catalog.persistance.Persister;
 import org.polypheny.db.catalog.snapshot.Snapshot;
 import org.polypheny.db.catalog.snapshot.impl.SnapshotBuilder;
-import org.polypheny.db.catalog.util.ConstraintCondition;
 import org.polypheny.db.config.RuntimeConfig;
 import org.polypheny.db.iface.QueryInterfaceManager.QueryInterfaceTemplate;
 import org.polypheny.db.nodes.Node;
@@ -109,12 +103,6 @@ public class PolyCatalog extends Catalog implements PolySerializable {
     @Getter
     private final BinarySerializer<PolyCatalog> serializer = PolySerializable.buildSerializer( PolyCatalog.class );
 
-    /**
-     * Constraints which have to be met before a commit can be executed.
-     */
-    private final Collection<ConstraintCondition> commitConstraints = new ConcurrentLinkedDeque<>();
-    private final Collection<Runnable> commitActions = new ConcurrentLinkedDeque<>();
-
     // indicates if the state has advanced and the snapshot has to be recreated or can be reused // trx without ddl
     private long lastCommitSnapshotId = 0;
 
@@ -243,19 +231,6 @@ public String getJson() {
     }
 
 
-    @Override
-    public void executeCommitActions() {
-        // execute physical changes
-        commitActions.forEach( Runnable::run );
-    }
-
-
-    @Override
-    public void clearCommitActions() {
-        commitActions.clear();
-    }
-
-
     public synchronized void commit() {
         if ( !this.dirty.get() ) {
             log.debug( "Nothing changed" );
@@ -275,33 +250,12 @@ public synchronized void commit() {
         updateSnapshot();
         persister.write( backup );
         this.dirty.set( false );
-        this.commitConstraints.clear();
-        this.commitActions.clear();
 
         this.lastCommitSnapshotId = snapshot.id();
     }
 
 
-    @Override
-    public Pair<@NotNull Boolean, @Nullable String> checkIntegrity() {
-        // check constraints e.g. primary key constraints
-        List<Pair<Boolean, String>> fails = commitConstraints
-                .stream()
-                .map( c -> Pair.of( c.condition().get(), c.errorMessage() ) )
-                .filter( c -> !c.left )
-                .toList();
-
-        if ( !fails.isEmpty() ) {
-            commitConstraints.clear();
-            return Pair.of( false, "DDL constraints not met: \n" + fails.stream().map( f -> f.right ).collect( Collectors.joining( ",\n " ) ) + "." );
-        }
-        return Pair.of( true, null );
-    }
-
-
     public void rollback() {
-        commitActions.clear();
-        commitConstraints.clear();
 
         restoreLastState();
 
@@ -570,18 +524,6 @@ private void restoreViews( Transaction transaction ) {
     }
 
 
-    @Override
-    public void attachCommitConstraint( Supplier<Boolean> constraintChecker, String description ) {
-        commitConstraints.add( new ConstraintCondition( constraintChecker, description ) );
-    }
-
-
-    @Override
-    public void attachCommitAction( Runnable action ) {
-        commitActions.add( action );
-    }
-
-
     @Override
     public void close() {
         log.error( "closing" );

core/src/main/java/org/polypheny/db/ddl/DdlManager.java

@@ -45,6 +45,7 @@
 import org.polypheny.db.nodes.Node;
 import org.polypheny.db.partition.raw.RawPartitionInformation;
 import org.polypheny.db.transaction.Statement;
+import org.polypheny.db.transaction.Transaction;
 import org.polypheny.db.transaction.TransactionException;
 import org.polypheny.db.type.PolyType;
 import org.polypheny.db.type.entity.PolyValue;
@@ -116,14 +117,15 @@ public static DdlManager getInstance() {
     /**
      * Adds a new data source(adapter)
      *
+     * @param transaction current transaction
      * @param uniqueName unique name of the newly created source
      * @param adapterName name of source, which is used to create the source
      * @param namespace the target namespace for the adapter
      * @param adapterType the specific {@link AdapterType} for the source to create
      * @param config configuration for the source
      * @param mode the deploy mode
      */
-    public abstract void createSource( String uniqueName, String adapterName, long namespace, AdapterType adapterType, Map<String, String> config, DeployMode mode );
+    public abstract void createSource( Transaction transaction, String uniqueName, String adapterName, long namespace, AdapterType adapterType, Map<String, String> config, DeployMode mode );
 
 
     /**
@@ -248,10 +250,11 @@ public static DdlManager getInstance() {
     /**
      * Drop a specific constraint from a table
      *
+     * @param transaction current transaction
      * @param table the table
      * @param constraintName the name of the constraint to be dropped
      */
-    public abstract void dropConstraint( LogicalTable table, String constraintName );
+    public abstract void dropConstraint( Transaction transaction, LogicalTable table, String constraintName );
 
     /**
      * Drop a foreign key of a table

core/src/main/java/org/polypheny/db/transaction/Transaction.java

@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 import org.jetbrains.annotations.Nullable;
 import org.polypheny.db.adapter.Adapter;
 import org.polypheny.db.adapter.java.JavaTypeFactory;
@@ -44,6 +45,10 @@ public interface Transaction {
 
     LogicalUser getUser();
 
+    void attachCommitAction( Runnable action );
+
+    void attachCommitConstraint( Supplier<Boolean> constraintChecker, String description );
+
     void commit() throws TransactionException;
 
     /**

core/src/test/java/org/polypheny/db/catalog/MockCatalog.java

@@ -20,10 +20,7 @@
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.function.Supplier;
 import org.apache.commons.lang3.NotImplementedException;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 import org.polypheny.db.adapter.DeployMode;
 import org.polypheny.db.catalog.catalogs.AllocationDocumentCatalog;
 import org.polypheny.db.catalog.catalogs.AllocationGraphCatalog;
@@ -37,7 +34,6 @@
 import org.polypheny.db.catalog.entity.LogicalUser;
 import org.polypheny.db.catalog.logistic.DataModel;
 import org.polypheny.db.catalog.snapshot.Snapshot;
-import org.polypheny.db.util.Pair;
 
 
 /**
@@ -212,34 +208,4 @@ public void clear() {
         throw new NotImplementedException();
     }
 
-
-    @Override
-    public void executeCommitActions() {
-        this.commitActions.forEach( Runnable::run );
-    }
-
-
-    @Override
-    public void clearCommitActions() {
-        this.commitActions.clear();
-    }
-
-
-    @Override
-    public void attachCommitConstraint( Supplier<Boolean> constraintChecker, String description ) {
-        // empty on purpose
-    }
-
-
-    @Override
-    public void attachCommitAction( Runnable action ) {
-        commitActions.add( action );
-    }
-
-
-    @Override
-    public Pair<@NotNull Boolean, @Nullable String> checkIntegrity() {
-        return Pair.of( true, null );
-    }
-
 }

dbms/src/main/java/org/polypheny/db/PolyphenyDb.java

@@ -559,7 +559,7 @@ private void restore( Authenticator authenticator, Catalog catalog ) {
 
         Catalog.defaultStore = AdapterTemplate.fromString( defaultStoreName, AdapterType.STORE );
         Catalog.defaultSource = AdapterTemplate.fromString( defaultSourceName, AdapterType.SOURCE );
-        restoreDefaults( catalog, mode );
+        restoreDefaults( trx, catalog, mode );
 
         QueryInterfaceManager.getInstance().restoreInterfaces( catalog.getSnapshot() );
 
@@ -585,9 +585,9 @@ private void commitRestore( Transaction trx ) {
      * @param catalog the current catalog
      * @param mode the current mode
      */
-    private static void restoreDefaults( Catalog catalog, RunMode mode ) {
+    private static void restoreDefaults( Transaction transaction, Catalog catalog, RunMode mode ) {
         catalog.updateSnapshot();
-        DefaultInserter.resetData( DdlManager.getInstance(), mode );
+        DefaultInserter.resetData( transaction, DdlManager.getInstance(), mode );
         DefaultInserter.restoreInterfacesIfNecessary( catalog );
     }