Add support for table partitions with morf

morf-core/src/main/java/org/alfasoftware/morf/dataset/WithMetaDataAdapter.java

@@ -99,6 +99,7 @@ public Collection<String> tableNames() {
           return sourceSchema.tableNames();
         }
 
+
         @Override
         public Collection<Table> tables() {
           Set<Table> tables = new HashSet<>();

morf-core/src/main/java/org/alfasoftware/morf/jdbc/DatabaseMetaDataProvider.java

@@ -28,16 +28,20 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.alfasoftware.morf.metadata.Column;
 import org.alfasoftware.morf.metadata.DataType;
 import org.alfasoftware.morf.metadata.Index;
+import org.alfasoftware.morf.metadata.PartitioningRule;
+import org.alfasoftware.morf.metadata.Partitions;
 import org.alfasoftware.morf.metadata.Schema;
 import org.alfasoftware.morf.metadata.SchemaUtils;
 import org.alfasoftware.morf.metadata.SchemaUtils.ColumnBuilder;
@@ -116,6 +120,8 @@ public abstract class DatabaseMetaDataProvider implements Schema {
   private final LoadingCache<AName, Sequence> sequenceCache = CacheBuilder.newBuilder().build(CacheLoader.from(this::loadSequence));
 
   private final Supplier<Map<String, String>> databaseInformation = Suppliers.memoize(this::loadDatabaseInformation);
+  protected Supplier<Set<String>> ignoredTables = Suppliers.memoize(this::getIgnoredTables);
+  protected Supplier<Set<String>> partitionedTables = Suppliers.memoize(this::getPartitionedTables);
 
   /**
    * @param connection The database connection from which meta data should be provided.
@@ -148,6 +154,9 @@ private Map<String, String> loadDatabaseInformation() {
     }
   }
 
+  protected Set<String> getIgnoredTables() { return new HashSet<>(); }
+
+  protected Set<String> getPartitionedTables() { return new HashSet<>(); }
 
   /**
    * @see org.alfasoftware.morf.metadata.Schema#isEmptyDatabase()
@@ -306,6 +315,11 @@ protected Map<AName, RealName> loadAllTableNames() {
             throw new RuntimeSqlException("Error reading metadata for table ["+tableName+"]", e);
           }
         }
+        // add partitioned tables to list
+        partitionedTables.get().forEach(table -> {
+          RealName partionedTableName = createRealName(table, table);
+          tableNameMappings.put(partionedTableName, partionedTableName);
+        });
 
         long end = System.currentTimeMillis();
         Map<AName, RealName> tableNameMap = tableNameMappings.build();
@@ -587,7 +601,6 @@ protected ColumnBuilder setColumnAutonumbered(RealName tableName, ColumnBuilder
 
   /**
    * Sets column default value.
-   *
    * Note: Uses an empty string for any column other than version.
    * Database-schema level default values are not supported by ALFA's domain model
    * hence we don't want to include a default value in the definition of tables.
@@ -677,6 +690,7 @@ protected Table loadTable(AName tableName) {
     final Map<AName, Integer> primaryKey = loadTablePrimaryKey(realTableName);
     final Supplier<List<Column>> columns = Suppliers.memoize(() -> loadTableColumns(realTableName, primaryKey));
     final Supplier<List<Index>> indexes = Suppliers.memoize(() -> loadTableIndexes(realTableName, false));
+    final Supplier<Partitions> partitions = Suppliers.memoize(() -> loadTablePartitions(realTableName));
 
     return new Table() {
       @Override
@@ -698,6 +712,19 @@ public List<Index> indexes() {
       public boolean isTemporary() {
         return false;
       }
+
+      @Override
+      public boolean isPartitioned() { return partitions.get() != null; }
+
+      @Override
+      public PartitioningRule partitioningRule() {
+        return partitions.get() == null ? null : partitions.get().partitioningRule();
+      }
+
+      @Override
+      public Partitions partitions() {
+        return partitions.get();
+      }
     };
   }
 
@@ -857,6 +884,11 @@ protected List<Index> loadTableIndexes(RealName tableName, boolean returnIgnored
   }
 
 
+  protected Partitions loadTablePartitions(RealName tableName) {
+   return null;
+  }
+
+
   /**
    * Retrieves index name from a result set.
    *
@@ -1112,8 +1144,8 @@ protected static AName named(String name) {
 
   /**
    * Build the SQL to return sequence information from the metadata.
-   * @param schemaName
-   * @return
+   * @param schemaName the schema name
+   * @return the sql for the sequence
    */
   protected abstract String buildSequenceSql(String schemaName);
 
@@ -1378,6 +1410,11 @@ public int getAutoNumberStart() {
       throw new UnexpectedDataTypeException(this.toString());
     }
 
+    @Override
+    public boolean isPartitioned() {
+      return false;
+    }
+
     @Override
     public String getDefaultValue() {
       throw new UnexpectedDataTypeException(this.toString());
@@ -1424,5 +1461,10 @@ public ColumnBuilder autoNumbered(int from) {
     public ColumnBuilder dataType(DataType dataType) {
       return this;
     }
+
+    @Override
+    public ColumnBuilder partitioned() {
+      return this;
+    }
   }
 }

morf-core/src/main/java/org/alfasoftware/morf/jdbc/SqlDialect.java

@@ -47,6 +47,8 @@
 import org.alfasoftware.morf.metadata.DataType;
 import org.alfasoftware.morf.metadata.DataValueLookup;
 import org.alfasoftware.morf.metadata.Index;
+import org.alfasoftware.morf.metadata.PartitioningRule;
+import org.alfasoftware.morf.metadata.Partitions;
 import org.alfasoftware.morf.metadata.Schema;
 import org.alfasoftware.morf.metadata.SchemaResource;
 import org.alfasoftware.morf.metadata.SchemaUtils;
@@ -4621,5 +4623,18 @@ public List<Column> columns() {
     public boolean isTemporary() {
       return isTemporary;
     }
+
+    @Override
+    public boolean isPartitioned() { return false; }
+
+    @Override
+    public PartitioningRule partitioningRule() {
+      return null;
+    }
+
+    @Override
+    public Partitions partitions() {
+      return null;
+    }
   }
 }

morf-core/src/main/java/org/alfasoftware/morf/metadata/AdditionalMetadata.java

@@ -1,6 +1,7 @@
 package org.alfasoftware.morf.metadata;
 
 import java.util.List;
+import java.util.Collection;
 import java.util.Map;
 
 import org.apache.commons.lang3.NotImplementedException;
@@ -22,4 +23,23 @@ default Map<String, List<Index>> ignoredIndexes() {
     return Map.of();
   }
 
+  /**
+   * Provides the names of all partition tables in the database. This applies for now for postgres. Note that the order of
+   * the tables in the result is not specified. The case of the
+   * table names may be preserved when logging progress, but should not be relied on for schema
+   * processing.
+   *
+   * @return A collection of all partitioned table names available in the database.
+   */
+  default Collection<String> partitionedTableNames() { throw new NotImplementedException("Not implemented yet."); }
+
+  /**
+   * Provides the names of all partition tables in the database. This applies for now for postgres. Note that the order of
+   * the tables in the result is not specified. The case of the
+   * table names may be preserved when logging progress, but should not be relied on for schema
+   * processing.
+   *
+   * @return A collection of all partition table names available in the database.
+   */
+  default Collection<String> partitionTableNames() { throw new NotImplementedException("Not implemented yet."); }
 }

morf-core/src/main/java/org/alfasoftware/morf/metadata/Column.java

@@ -78,6 +78,12 @@ public default String getUpperCaseName() {
   public int getAutoNumberStart();
 
 
+  /**
+   * @return True if the column is the partition by source
+   */
+  boolean isPartitioned();
+
+
   /**
    * Helper for {@link Object#toString()} implementations.
    *

morf-core/src/main/java/org/alfasoftware/morf/metadata/ColumnBean.java

@@ -34,7 +34,8 @@ class ColumnBean extends ColumnTypeBean implements Column {
   private final String defaultValue;
   private final boolean autoNumber;
   private final int autoNumberStart;
-
+  //TODO change to private with appropriate constructors.
+  protected boolean partitioned;
 
   /**
    * Creates a column with zero precision.
@@ -218,4 +219,11 @@ public int getAutoNumberStart() {
   public String toString() {
     return this.toStringHelper();
   }
+
+  @Override
+  public boolean isPartitioned() {
+    return partitioned;
+  }
+
+
 }

morf-core/src/main/java/org/alfasoftware/morf/metadata/DatePartitionedByPeriodRule.java

@@ -0,0 +1,41 @@
+package org.alfasoftware.morf.metadata;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.joda.time.LocalDate;
+import org.joda.time.Period;
+import org.joda.time.ReadablePeriod;
+
+public class DatePartitionedByPeriodRule extends PartitioningByRangeRule<LocalDate, Period> {
+
+    public DatePartitionedByPeriodRule(String column, LocalDate startValue, Period period, int count) {
+        super(column, DataType.DATE, startValue, period, count);
+    }
+
+    public DatePartitionedByPeriodRule(String column, List<Pair<LocalDate, LocalDate>> ranges) {
+        super(column, DataType.DATE, ranges);
+    }
+
+    @Override
+    public List<Pair<LocalDate, LocalDate>> getRanges() {
+        List<Pair<LocalDate, LocalDate>> ranges = new ArrayList<Pair<LocalDate, LocalDate>>();
+
+        if (startValue != null) {
+
+            ReadablePeriod readablePeriod = increment.toPeriod();
+            startValue.plus(readablePeriod);
+
+            int i = count;
+            for (LocalDate current = startValue; i > 0; i--) {
+                ranges.add(Pair.of(current, current.plus(readablePeriod)));
+                current = current.plus(readablePeriod);
+            }
+        } else {
+            ranges.addAll(this.partitions);
+        }
+
+        return ranges;
+    }
+}

morf-core/src/main/java/org/alfasoftware/morf/metadata/Index.java

@@ -41,6 +41,15 @@ public interface Index {
    */
   public boolean isUnique();
 
+  /**
+   * @return True if the index is globally partitioned
+   */
+  boolean isGlobalPartitioned();
+
+  /**
+   * @return True if the index is locally partitioned
+   */
+  boolean isLocalPartitioned();
 
   /**
    * Helper for {@link Object#toString()} implementations.

morf-core/src/main/java/org/alfasoftware/morf/metadata/IndexBean.java

@@ -42,6 +42,17 @@ class IndexBean implements Index {
   private final boolean unique;
 
 
+  /**
+   * Flags if index is partitioned and global.
+   */
+  //TODO: change this protected properties from protected to private and add the appropriate constructors to consider them
+  protected boolean isGlobalPartitioned;
+
+  /**
+   * Flags if index is partitioned and local.
+   */
+  protected boolean isLocalPartitioned;
+
   /**
    * Creates an index bean.
    *
@@ -110,6 +121,13 @@ public boolean isUnique() {
   }
 
 
+  @Override
+  public boolean isGlobalPartitioned() { return isGlobalPartitioned; }
+
+  @Override
+  public boolean isLocalPartitioned() { return isLocalPartitioned; }
+
+
   @Override
   public String toString() {
     return this.toStringHelper();

morf-core/src/main/java/org/alfasoftware/morf/metadata/Partition.java

@@ -0,0 +1,10 @@
+package org.alfasoftware.morf.metadata;
+
+/**
+ * Defines a partition on a table.
+ *
+ * @author Copyright (c) Alfa Financial Software 2025
+ */
+public interface Partition {
+  String name();
+}

morf-core/src/main/java/org/alfasoftware/morf/metadata/PartitionBean.java

@@ -0,0 +1,20 @@
+package org.alfasoftware.morf.metadata;
+
+/**
+ * Defines the bean for one partitions on a table. {@link Partition}
+ *
+ * @author Copyright (c) Alfa Financial Software 2025
+ */
+public abstract class PartitionBean implements Partition {
+
+  protected String name;
+
+  public PartitionBean(String name) {
+    this.name = name;
+  }
+
+  @Override
+  public String name() {
+    return name;
+  }
+}

morf-core/src/main/java/org/alfasoftware/morf/metadata/PartitionByHash.java

@@ -0,0 +1,12 @@
+package org.alfasoftware.morf.metadata;
+
+/**
+ * Defines a partition by range on a table.
+ *
+ * @author Copyright (c) Alfa Financial Software 2025
+ */
+public interface PartitionByHash extends Partition {
+  String hashFunction();
+  String divider();
+  String remainder();
+}