Merge branch 'all_versions' of https://github.com/prasar-ashutosh/leg…

…end-engine into all_versions
prasar-ashutosh · Oct 11, 2023 · 66514f2 · 66514f2
2 parents e813eeb + 12d022d
commit 66514f2
Show file tree

Hide file tree

Showing 23 changed files with 137 additions and 110 deletions.
diff --git a/...in/java/org/finos/legend/engine/persistence/components/ingestmode/AppendOnlyAbstract.java b/...in/java/org/finos/legend/engine/persistence/components/ingestmode/AppendOnlyAbstract.java
@@ -18,7 +18,7 @@
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.AllVersionsStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.MaxVersionStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.NoVersioningStrategyAbstract;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategyVisitor;
 import org.immutables.value.Value;
 
@@ -61,19 +61,19 @@ public Void visitNoVersioningStrategy(NoVersioningStrategyAbstract noVersioningS
             @Override
             public Void visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionStrategy)
             {
-                if (maxVersionStrategy.versioningComparator() != VersioningComparator.ALWAYS)
+                if (maxVersionStrategy.versionResolver() != VersionResolver.DIGEST_BASED)
                 {
-                    throw new IllegalStateException("Cannot build AppendOnly, versioning comparator can only be Always");
+                    throw new IllegalStateException("Cannot build AppendOnly, Only DIGEST_BASED VersioningResolver allowed for this ingest mode");
                 }
                 return null;
             }
 
             @Override
             public Void visitAllVersionsStrategy(AllVersionsStrategyAbstract allVersionsStrategyAbstract)
             {
-                if (allVersionsStrategyAbstract.versioningComparator() != VersioningComparator.ALWAYS)
+                if (allVersionsStrategyAbstract.versionResolver() != VersionResolver.DIGEST_BASED)
                 {
-                    throw new IllegalStateException("Cannot build AppendOnly, versioning comparator can only be Always");
+                    throw new IllegalStateException("Cannot build AppendOnly, Only DIGEST_BASED VersioningResolver allowed for this ingest mode");
                 }
                 return null;
             }

diff --git a/...va/org/finos/legend/engine/persistence/components/ingestmode/BitemporalDeltaAbstract.java b/...va/org/finos/legend/engine/persistence/components/ingestmode/BitemporalDeltaAbstract.java
@@ -25,7 +25,7 @@
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.AllVersionsStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.MaxVersionStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.NoVersioningStrategyAbstract;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategyVisitor;
 import org.immutables.value.Value;
 
@@ -89,9 +89,9 @@ public Void visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionStrateg
             @Override
             public Void visitAllVersionsStrategy(AllVersionsStrategyAbstract allVersionsStrategyAbstract)
             {
-                if (allVersionsStrategyAbstract.versioningComparator() != VersioningComparator.ALWAYS)
+                if (allVersionsStrategyAbstract.versionResolver() != VersionResolver.DIGEST_BASED)
                 {
-                    throw new IllegalStateException("Cannot build BitemporalDelta, versioning comparator can only be Always");
+                    throw new IllegalStateException("Cannot build BitemporalDelta, Only DIGEST_BASED VersioningResolver allowed for this ingest mode");
                 }
                 if (allVersionsStrategyAbstract.performVersioning())
                 {

diff --git a/...va/org/finos/legend/engine/persistence/components/ingestmode/IngestModeCaseConverter.java b/...va/org/finos/legend/engine/persistence/components/ingestmode/IngestModeCaseConverter.java
@@ -348,7 +348,7 @@ public VersioningStrategy visitMaxVersionStrategy(MaxVersionStrategyAbstract max
         {
             return MaxVersionStrategy
                     .builder()
-                    .versioningComparator(maxVersionStrategy.versioningComparator())
+                    .versionResolver(maxVersionStrategy.versionResolver())
                     .versioningField(strategy.apply(maxVersionStrategy.versioningField()))
                     .performVersioning(maxVersionStrategy.performVersioning())
                     .build();
@@ -359,7 +359,7 @@ public VersioningStrategy visitAllVersionsStrategy(AllVersionsStrategyAbstract a
         {
             return AllVersionsStrategy
                     .builder()
-                    .versioningComparator(allVersionsStrategyAbstract.versioningComparator())
+                    .versionResolver(allVersionsStrategyAbstract.versionResolver())
                     .versioningField(strategy.apply(allVersionsStrategyAbstract.versioningField()))
                     .dataSplitFieldName(strategy.apply(allVersionsStrategyAbstract.dataSplitFieldName()))
                     .performVersioning(allVersionsStrategyAbstract.performVersioning())

diff --git a/...rg/finos/legend/engine/persistence/components/ingestmode/NontemporalSnapshotAbstract.java b/...rg/finos/legend/engine/persistence/components/ingestmode/NontemporalSnapshotAbstract.java
@@ -18,7 +18,7 @@
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.AllVersionsStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.MaxVersionStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.NoVersioningStrategyAbstract;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategyVisitor;
 import org.immutables.value.Value;
 
@@ -58,9 +58,9 @@ public Void visitNoVersioningStrategy(NoVersioningStrategyAbstract noVersioningS
             @Override
             public Void visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionStrategy)
             {
-                if (maxVersionStrategy.versioningComparator() != VersioningComparator.ALWAYS)
+                if (maxVersionStrategy.versionResolver() != VersionResolver.DIGEST_BASED)
                 {
-                    throw new IllegalStateException("Cannot build NontemporalSnapshot, versioning comparator can only be Always");
+                    throw new IllegalStateException("Cannot build NontemporalSnapshot, Only DIGEST_BASED VersioningResolver allowed for this ingest mode");
                 }
                 return null;
             }

diff --git a/...rg/finos/legend/engine/persistence/components/ingestmode/UnitemporalSnapshotAbstract.java b/...rg/finos/legend/engine/persistence/components/ingestmode/UnitemporalSnapshotAbstract.java
@@ -18,6 +18,11 @@
 import org.finos.legend.engine.persistence.components.ingestmode.emptyhandling.EmptyDatasetHandling;
 import org.finos.legend.engine.persistence.components.ingestmode.transactionmilestoning.TransactionMilestoned;
 import org.finos.legend.engine.persistence.components.ingestmode.transactionmilestoning.TransactionMilestoning;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategyVisitor;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.NoVersioningStrategyAbstract;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.AllVersionsStrategyAbstract;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.MaxVersionStrategyAbstract;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver;
 import org.immutables.value.Value;
 
 import java.util.List;
@@ -83,5 +88,32 @@ default void validate()
                 }
             }
         }
+
+        // Allowed Versioning Strategy - NoVersioning, MaxVersioining
+        this.versioningStrategy().accept(new VersioningStrategyVisitor<Void>()
+        {
+            @Override
+            public Void visitNoVersioningStrategy(NoVersioningStrategyAbstract noVersioningStrategy)
+            {
+                return null;
+            }
+
+            @Override
+            public Void visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionStrategy)
+            {
+                if (maxVersionStrategy.versionResolver() != VersionResolver.DIGEST_BASED)
+                {
+                    throw new IllegalStateException("Cannot build UnitemporalSnapshot, Only DIGEST_BASED VersioningResolver allowed for this ingest mode");
+                }
+                return null;
+            }
+
+            @Override
+            public Void visitAllVersionsStrategy(AllVersionsStrategyAbstract allVersionsStrategyAbstract)
+            {
+                throw new IllegalStateException("Cannot build UnitemporalSnapshot, AllVersionsStrategy not supported");
+            }
+        });
+
     }
 }
diff --git a/...gend/engine/persistence/components/ingestmode/versioning/AllVersionsStrategyAbstract.java b/...gend/engine/persistence/components/ingestmode/versioning/AllVersionsStrategyAbstract.java
@@ -14,9 +14,6 @@
 
 package org.finos.legend.engine.persistence.components.ingestmode.versioning;
 
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategy;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningStrategyVisitor;
 import org.immutables.value.Value;
 
 import static org.immutables.value.Value.Immutable;
@@ -39,9 +36,9 @@ public interface AllVersionsStrategyAbstract extends VersioningStrategy
     String versioningField();
 
     @Value.Default
-    default VersioningComparator versioningComparator()
+    default VersionResolver versionResolver()
     {
-        return VersioningComparator.GREATER_THAN;
+        return VersionResolver.DIGEST_BASED;
     }
 
     @Value.Default

diff --git a/...egend/engine/persistence/components/ingestmode/versioning/MaxVersionStrategyAbstract.java b/...egend/engine/persistence/components/ingestmode/versioning/MaxVersionStrategyAbstract.java
@@ -33,9 +33,9 @@ public interface MaxVersionStrategyAbstract extends VersioningStrategy
     String versioningField();
 
     @Value.Default
-    default VersioningComparator versioningComparator()
+    default VersionResolver versionResolver()
     {
-        return VersioningComparator.GREATER_THAN;
+        return VersionResolver.DIGEST_BASED;
     }
 
     @Value.Default

diff --git a/...mode/versioning/VersioningComparator.java → ...ngestmode/versioning/VersionResolver.java b/...mode/versioning/VersioningComparator.java → ...ngestmode/versioning/VersionResolver.java
@@ -14,9 +14,9 @@
 
 package org.finos.legend.engine.persistence.components.ingestmode.versioning;
 
-public enum VersioningComparator
+public enum VersionResolver
 {
-    GREATER_THAN,
-    GREATER_THAN_EQUAL_TO,
-    ALWAYS
+    GREATER_THAN_ACTIVE_VERSION,
+    GREATER_THAN_EQUAL_TO_ACTIVE_VERSION,
+    DIGEST_BASED
 }
diff --git a/...egend/engine/persistence/components/ingestmode/versioning/VersioningConditionVisitor.java b/...egend/engine/persistence/components/ingestmode/versioning/VersioningConditionVisitor.java
@@ -59,9 +59,9 @@ public Condition visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionSt
         FieldValue mainVersioningField = FieldValue.builder().datasetRef(mainDataset.datasetReference()).fieldName(maxVersionStrategy.versioningField()).build();
         FieldValue stagingVersioningField = FieldValue.builder().datasetRef(stagingDataset.datasetReference()).fieldName(maxVersionStrategy.versioningField()).build();
 
-        switch (maxVersionStrategy.versioningComparator())
+        switch (maxVersionStrategy.versionResolver())
         {
-            case GREATER_THAN:
+            case GREATER_THAN_ACTIVE_VERSION:
                 if (invertComparison)
                 {
                     return LessThanEqualTo.of(stagingVersioningField, mainVersioningField);
@@ -70,7 +70,7 @@ public Condition visitMaxVersionStrategy(MaxVersionStrategyAbstract maxVersionSt
                 {
                     return GreaterThan.of(stagingVersioningField, mainVersioningField);
                 }
-            case GREATER_THAN_EQUAL_TO:
+            case GREATER_THAN_EQUAL_TO_ACTIVE_VERSION:
                 if (invertComparison)
                 {
                     return LessThan.of(stagingVersioningField, mainVersioningField);

diff --git a/...va/org/finos/legend/engine/persistence/components/planner/UnitemporalSnapshotPlanner.java b/...va/org/finos/legend/engine/persistence/components/planner/UnitemporalSnapshotPlanner.java
@@ -139,12 +139,13 @@ protected Insert sqlToUpsertRows()
                 .addFields(FieldValue.builder().datasetRef(mainDataset().datasetReference()).fieldName(ingestMode().digestField()).build())
                 .build()));
 
-        List<Value> fieldsToSelect = new ArrayList<>(stagingDataset().schemaReference().fieldValues());
+        List<Value> dataFields = getDataFields();
+        List<Value> fieldsToSelect = new ArrayList<>(dataFields);
         List<Value> milestoneUpdateValues = transactionMilestoningFieldValues();
         fieldsToSelect.addAll(milestoneUpdateValues);
         Dataset selectStage = Selection.builder().source(stagingDataset()).condition(notInSinkCondition).addAllFields(fieldsToSelect).build();
 
-        List<Value> fieldsToInsert = new ArrayList<>(stagingDataset().schemaReference().fieldValues());
+        List<Value> fieldsToInsert = new ArrayList<>(dataFields);
         fieldsToInsert.addAll(transactionMilestoningFields());
 
         return Insert.of(mainDataset(), selectStage, fieldsToInsert);

diff --git a/...c/test/java/org/finos/legend/engine/persistence/components/util/LogicalPlanUtilsTest.java b/...c/test/java/org/finos/legend/engine/persistence/components/util/LogicalPlanUtilsTest.java
@@ -38,7 +38,7 @@
 import java.util.Map;
 import com.fasterxml.jackson.core.JsonProcessingException;
 
-import static org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator.GREATER_THAN;
+import static org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver.GREATER_THAN_ACTIVE_VERSION;
 
 public class LogicalPlanUtilsTest extends IngestModeTest
 {
@@ -57,7 +57,7 @@ public void testDeduplicateByMaxVersion()
         RelationalTransformer transformer = new RelationalTransformer(AnsiSqlSink.get());
 
         List<String> primaryKeys = Arrays.asList("id", "name");
-        VersioningStrategy versioningStrategy = MaxVersionStrategy.builder().versioningField("version").performVersioning(true).versioningComparator(GREATER_THAN).build();
+        VersioningStrategy versioningStrategy = MaxVersionStrategy.builder().versioningField("version").performVersioning(true).versionResolver(GREATER_THAN_ACTIVE_VERSION).build();
         Selection selection = (Selection) versioningStrategy.accept(new DatasetDeduplicator(dataset, primaryKeys));
         LogicalPlan logicalPlan = LogicalPlan.builder().addOps(selection).build();
         SqlPlan physicalPlan = transformer.generatePhysicalPlan(logicalPlan);
@@ -87,7 +87,7 @@ public void testDeduplicateByMaxVersionAndFilterDataset()
                 .addDatasetFilters(DatasetFilter.of("biz_date", FilterType.LESS_THAN, "2020-01-03"))
                 .build();
 
-        VersioningStrategy versioningStrategy = MaxVersionStrategy.builder().versioningField("version").performVersioning(true).versioningComparator(GREATER_THAN).build();
+        VersioningStrategy versioningStrategy = MaxVersionStrategy.builder().versioningField("version").performVersioning(true).versionResolver(GREATER_THAN_ACTIVE_VERSION).build();
         Selection selection = (Selection) versioningStrategy.accept(new DatasetDeduplicator(dataset, primaryKeys));
 
         LogicalPlan logicalPlan = LogicalPlan.builder().addOps(selection).build();

diff --git a/...d/engine/persistence/components/ingestmode/bitemporal/BitemporalDeltaWithBatchIdTest.java b/...d/engine/persistence/components/ingestmode/bitemporal/BitemporalDeltaWithBatchIdTest.java
@@ -24,7 +24,7 @@
 import org.finos.legend.engine.persistence.components.ingestmode.validitymilestoning.derivation.SourceSpecifiesFromAndThruDateTime;
 import org.finos.legend.engine.persistence.components.ingestmode.validitymilestoning.derivation.SourceSpecifiesFromDateTime;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.AllVersionsStrategy;
-import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersioningComparator;
+import org.finos.legend.engine.persistence.components.ingestmode.versioning.VersionResolver;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.DatasetDefinition;
 import org.finos.legend.engine.persistence.components.planner.PlannerOptions;
@@ -516,7 +516,7 @@ void testMilestoningSourceSpecifiesFromSet3WithDataSplit() throws Exception
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -598,7 +598,7 @@ void testMilestoningSourceSpecifiesFromSet3WithDataSplitMultiPasses() throws Exc
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -788,7 +788,7 @@ void testMilestoningSourceSpecifiesFromSet5WithDataSplitFilterDuplicates() throw
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -871,7 +871,7 @@ void testMilestoningSourceSpecifiesFromSet5WithDataSplitFilterDuplicatesMultiPas
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -1121,7 +1121,7 @@ void testMilestoningSourceSpecifiesFromWithDeleteIndicatorSet3WithDataSplit() th
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -1202,7 +1202,7 @@ void testMilestoningSourceSpecifiesFromWithDeleteIndicatorSet3WithDataSplitWithM
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -1390,7 +1390,7 @@ void testMilestoningSourceSpecifiesFromWithDeleteIndicatorSet5WithDataSplitFilte
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()
@@ -1475,7 +1475,7 @@ void testMilestoningSourceSpecifiesFromWithDeleteIndicatorSet5WithDataSplitFilte
             .versioningStrategy(AllVersionsStrategy.builder()
                 .versioningField(versionName)
                 .dataSplitFieldName(dataSplitName)
-                .versioningComparator(VersioningComparator.ALWAYS)
+                .versionResolver(VersionResolver.DIGEST_BASED)
                 .performVersioning(false)
                 .build())
             .transactionMilestoning(BatchId.builder()