Feature/master thesis schema evolution in polystores

catalog/src/main/java/org/polypheny/db/catalog/CatalogImpl.java

@@ -17,23 +17,15 @@
 package org.polypheny.db.catalog;
 
 
+import com.google.common.base.Predicates;
+import com.google.common.collect.Collections2;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.io.File;
 import java.io.IOException;
 import java.sql.Timestamp;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -1901,6 +1893,55 @@ public long addTable( String name, long namespaceId, int ownerId, EntityType ent
     }
 
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public long relocateTable( CatalogTable sourceTable, long targetNamespaceId ) {
+        // Clone the source table by changing the ID of the parent namespace
+        CatalogTable targetTable = transferCatalogTable( sourceTable, targetNamespaceId );
+        synchronized ( this ) {
+            // Build the new immutable list for the source namespace by removing the table to transfer
+            ImmutableList<Long> reducedSourceSchemaChildren = ImmutableList
+                    .copyOf( Collections2.filter( schemaChildren.get( sourceTable.namespaceId ),
+                            Predicates.not( Predicates.equalTo( sourceTable.id ) ) ) );
+            // Build the new immutable list for the target namespace by adding the table to transfer
+            ImmutableList<Long> extendedTargetSchemaChildren = new ImmutableList.Builder<Long>()
+                    .addAll( schemaChildren.get( targetNamespaceId ) )
+                    .add( targetTable.id )
+                    .build();
+
+            // Replace the immutable list for both the source and target namespaces
+            schemaChildren.replace( sourceTable.namespaceId, reducedSourceSchemaChildren );
+            schemaChildren.replace( targetNamespaceId, extendedTargetSchemaChildren );
+
+            // Replace the tables' trees with the cloned table
+            tables.replace( sourceTable.id, targetTable );
+            tableNames.remove( new Object[]{ sourceTable.databaseId, sourceTable.namespaceId, sourceTable.name } );
+            tableNames.put( new Object[]{ targetTable.databaseId, targetNamespaceId, targetTable.name }, targetTable );
+
+            // Replace the trees of the tables' columns with cloned columns
+            for ( Long fieldId : sourceTable.fieldIds ) {
+                CatalogColumn targetCatalogColumn = transferCatalogColumn( targetNamespaceId, columns.get( fieldId ) );
+                columns.replace( fieldId, targetCatalogColumn );
+                columnNames.remove( new Object[]{ sourceTable.databaseId, sourceTable.namespaceId, sourceTable.id, targetCatalogColumn.name } );
+                columnNames.put( new Object[]{ sourceTable.databaseId, targetNamespaceId, sourceTable.id, targetCatalogColumn.name }, targetCatalogColumn );
+            }
+
+            // When transferring between document-based namespaces, also replace the collection trees.
+            if ( getSchema( sourceTable.namespaceId ).namespaceType == NamespaceType.DOCUMENT ) {
+                CatalogCollection targetCollection = transferCatalogCollection( collections.get( sourceTable.id ), targetNamespaceId );
+                collections.replace( sourceTable.id, targetCollection );
+                collectionNames.remove( new Object[]{ sourceTable.databaseId, sourceTable.namespaceId, sourceTable.name } );
+                collectionNames.put( new Object[]{ targetTable.databaseId, targetNamespaceId, targetTable.name }, targetCollection );
+            }
+        }
+        listeners.firePropertyChange( "table", sourceTable, null );
+
+        return sourceTable.id;
+    }
+
+
     /**
      * {@inheritDoc}
      */
@@ -5481,6 +5522,56 @@ private CatalogKey getKey( long keyId ) {
     }
 
 
+    private static CatalogColumn transferCatalogColumn( long targetNamespaceId, CatalogColumn sourceCatalogColumn ) {
+        CatalogColumn targetCatalogColumn = new CatalogColumn(
+                sourceCatalogColumn.id,
+                sourceCatalogColumn.name,
+                sourceCatalogColumn.tableId,
+                targetNamespaceId,
+                sourceCatalogColumn.databaseId,
+                sourceCatalogColumn.position,
+                sourceCatalogColumn.type,
+                sourceCatalogColumn.collectionsType,
+                sourceCatalogColumn.length,
+                sourceCatalogColumn.scale,
+                sourceCatalogColumn.dimension,
+                sourceCatalogColumn.cardinality,
+                sourceCatalogColumn.nullable,
+                sourceCatalogColumn.collation,
+                sourceCatalogColumn.defaultValue );
+        return targetCatalogColumn;
+    }
+
+
+    private CatalogTable transferCatalogTable( CatalogTable sourceTable, long targetNamespaceId ) {
+        return new CatalogTable(
+                sourceTable.id,
+                sourceTable.name,
+                sourceTable.fieldIds,
+                targetNamespaceId,
+                sourceTable.databaseId,
+                sourceTable.ownerId,
+                sourceTable.entityType,
+                sourceTable.primaryKey,
+                sourceTable.dataPlacements,
+                sourceTable.modifiable,
+                sourceTable.partitionProperty,
+                sourceTable.connectedViews );
+    }
+
+
+    private CatalogCollection transferCatalogCollection( CatalogCollection sourceCollection, long targetNamespaceId ) {
+        return new CatalogCollection(
+                sourceCollection.databaseId,
+                targetNamespaceId,
+                sourceCollection.id,
+                sourceCollection.name,
+                sourceCollection.placements,
+                sourceCollection.entityType,
+                sourceCollection.physicalName );
+    }
+
+
     static class CatalogValidator {
 
         public void validate() throws GenericCatalogException {
@@ -5510,4 +5601,4 @@ public void startCheck() {
 
     }
 
-}
+}

core/src/main/java/org/polypheny/db/algebra/constant/Kind.java

@@ -236,6 +236,11 @@ public enum Kind {
      */
     MERGE,
 
+    /**
+     * TRANSFER statement
+     */
+    TRANSFER,
+
     /**
      * TABLESAMPLE operator
      */
@@ -1713,4 +1718,3 @@ private static <E extends Enum<E>> EnumSet<E> concat( EnumSet<E> set0, EnumSet<E
         return set;
     }
 }
-

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

@@ -489,6 +489,15 @@ protected final boolean isValidIdentifier( final String str ) {
      */
     public abstract long addTable( String name, long namespaceId, int ownerId, EntityType entityType, boolean modifiable );
 
+    /**
+     * Relocate a table from one namespace to another if the model of both namespaces is the same.
+     *
+     * @param table The name of the table to add
+     * @param targetNamespaceId The id of the schema
+     * @return The id of the table (the ID of the target table remains the same as that of the source table)
+     */
+    public abstract long relocateTable( CatalogTable table, long targetNamespaceId );
+
 
     /**
      * Adds a view to a specified schema.
@@ -2411,4 +2420,4 @@ public static List<EntityType> convertTableTypeList( @NonNull final List<String>
         return typeList;
     }
 
-}
+}

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

@@ -193,6 +193,20 @@ public static DdlManager getInstance() {
      */
     public abstract void addForeignKey( CatalogTable catalogTable, CatalogTable refTable, List<String> columnNames, List<String> refColumnNames, String constraintName, ForeignKeyOption onUpdate, ForeignKeyOption onDelete ) throws UnknownColumnException, GenericCatalogException;
 
+    /**
+     * Merge multiple columns into one new column
+     *
+     * @param catalogTable the table
+     * @param sourceColumnNames name of the columns to be merged
+     * @param newColumnName name of the new column to be added
+     * @param joinString the string to place between the values
+     * @param type the SQL data type specification of the merged column
+     * @param nullable if the merged column should be nullable
+     * @param defaultValue the new default value of the merged column
+     * @param statement the initial query statement
+     */
+    public abstract void mergeColumns( CatalogTable catalogTable, List<String> sourceColumnNames, String newColumnName, String joinString, ColumnTypeInformation type, boolean nullable, String defaultValue, Statement statement ) throws UnknownColumnException, ColumnAlreadyExistsException, ColumnNotExistsException;
+
     /**
      * Adds an index to a table
      *
@@ -451,6 +465,17 @@ public static DdlManager getInstance() {
      */
     public abstract void createTable( long schemaId, String tableName, List<FieldInformation> columns, List<ConstraintInformation> constraints, boolean ifNotExists, List<DataStore> stores, PlacementType placementType, Statement statement ) throws EntityAlreadyExistsException, ColumnNotExistsException, UnknownPartitionTypeException, UnknownColumnException, PartitionGroupNamesNotUniqueException;
 
+    /**
+     * Transfer a table between two namespaces.
+     * Currently, the transfer works between namespaces of the same model and between relational and document-based and vice versa.
+     *
+     * @param table the table about to be transfered
+     * @param targetSchemaId the id of the target namespace
+     * @param statement the used statement
+     * @param statement the used statement
+     */
+    public abstract void transferTable( CatalogTable table, long targetSchemaId, Statement statement, Map<String, List<String>> primaryKeyColumnNames ) throws EntityAlreadyExistsException, DdlOnSourceException, UnknownTableException, UnknownColumnException, GenericCatalogException;
+
     /**
      * Create a new view
      *
@@ -751,4 +776,4 @@ public enum DefaultIndexPlacementStrategy {
         POLYPHENY, ONE_DATA_STORE, ALL_DATA_STORES
     }
 
-}
+}

core/src/main/java/org/polypheny/db/processing/DataMigrator.java

@@ -16,6 +16,7 @@
 
 package org.polypheny.db.processing;
 
+import com.google.gson.JsonObject;
 import java.util.List;
 import java.util.Map;
 import org.polypheny.db.algebra.AlgRoot;
@@ -24,10 +25,12 @@
 import org.polypheny.db.catalog.entity.CatalogColumnPlacement;
 import org.polypheny.db.catalog.entity.CatalogGraphDatabase;
 import org.polypheny.db.catalog.entity.CatalogTable;
+import org.polypheny.db.catalog.exceptions.UnknownColumnException;
+import org.polypheny.db.catalog.exceptions.UnknownTableException;
+import org.polypheny.db.ddl.DdlManager;
 import org.polypheny.db.transaction.Statement;
 import org.polypheny.db.transaction.Transaction;
 
-
 public interface DataMigrator {
 
     void copyData(
@@ -76,16 +79,48 @@ void copyPartitionData(
             List<Long> sourcePartitionIds,
             List<Long> targetPartitionIds );
 
+    /**
+     * Used to merge columns in a relational table. The values of the source columns will be selected,
+     * concatenated and inserted into the target column.
+     *
+     * @param transaction   Transactional scope
+     * @param store         Target Store where data should be migrated to
+     * @param sourceColumns Columns to be merged
+     * @param targetColumn  New column to be added
+     * @param joinString    String delimiter between the values to be merged
+     */
+    void mergeColumns( Transaction transaction, CatalogAdapter store, List<CatalogColumn> sourceColumns, CatalogColumn targetColumn, String joinString );
+
     AlgRoot buildInsertStatement( Statement statement, List<CatalogColumnPlacement> to, long partitionId );
 
-    //is used within copyData
+    // is used within copyData
     void executeQuery( List<CatalogColumn> columns, AlgRoot sourceRel, Statement sourceStatement, Statement targetStatement, AlgRoot targetRel, boolean isMaterializedView, boolean doesSubstituteOrderBy );
 
+    // is used within mergeColumns
+    void executeMergeQuery( List<CatalogColumn> primaryKeyColumns, List<CatalogColumn> sourceColumns, CatalogColumn targetColumn, String joinString, AlgRoot sourceRel, Statement sourceStatement, Statement targetStatement, AlgRoot targetRel, boolean isMaterializedView, boolean doesSubstituteOrderBy );
+
     AlgRoot buildDeleteStatement( Statement statement, List<CatalogColumnPlacement> to, long partitionId );
 
     AlgRoot getSourceIterator( Statement statement, Map<Long, List<CatalogColumnPlacement>> placementDistribution );
 
-
     void copyGraphData( CatalogGraphDatabase graph, Transaction transaction, Integer existingAdapterId, CatalogAdapter adapter );
 
+    /**
+     * Does migration when transferring between a relational and a document-based namespace.
+     *
+     * @param transaction Transactional scope
+     * @param sourceTable Source Table from where data is queried
+     * @param targetSchemaId ID of the target namespace
+     */
+    void copyRelationalDataToDocumentData(Transaction transaction , CatalogTable sourceTable, long targetSchemaId);
+
+    /**
+     * Does migration when transferring between a document-based and a relational namespace.
+     *
+     * @param transaction Transactional scope
+     * @param jsonObjects List of the JSON-objects of the source collection
+     * @param table Target tables created in the {@link DdlManager}
+     */
+    void copyDocumentDataToRelationalData( Transaction transaction, List<JsonObject> jsonObjects, List<CatalogTable> table ) throws UnknownColumnException, UnknownTableException;
+
 }

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

@@ -352,6 +352,12 @@ public long addTable( String name, long namespaceId, int ownerId, EntityType ent
     }
 
 
+    @Override
+    public long relocateTable( CatalogTable sourceTable, long targetNamespaceId ) {
+        throw new NotImplementedException();
+    }
+
+
     @Override
     public long addView( String name, long namespaceId, int ownerId, EntityType entityType, boolean modifiable, AlgNode definition, AlgCollation algCollation, Map<Long, List<Long>> underlyingTables, AlgDataType fieldList, String query, QueryLanguage language ) {
         throw new NotImplementedException();
@@ -1371,4 +1377,4 @@ public void updateCollectionPartitionPhysicalNames( long collectionId, int adapt
         throw new NotImplementedException();
     }
 
-}
+}