finos · epsstan · Dec 4, 2023 · Aug 14, 2023 · Aug 14, 2023 · Aug 16, 2023
@@ -41,4 +41,8 @@ public interface LoadOptionsAbstract
     Optional<Long> maxBadRecords();
 
     Optional<String> compression();
+
+    Optional<Boolean> force();
+
+    Optional<String> onError();
 }
@@ -15,10 +15,10 @@
 package org.finos.legend.engine.persistence.components.ingestmode;
 
 import org.finos.legend.engine.persistence.components.ingestmode.audit.Auditing;
+import org.finos.legend.engine.persistence.components.ingestmode.digest.DigestGenStrategy;
+import org.finos.legend.engine.persistence.components.ingestmode.digest.NoDigestGenStrategy;
 import org.immutables.value.Value;
 
-import java.util.Optional;
-
 import static org.immutables.value.Value.Immutable;
 import static org.immutables.value.Value.Style;
 
@@ -32,7 +32,11 @@
 )
 public interface AppendOnlyAbstract extends IngestMode
 {
-    Optional<String> digestField();
+    @Value.Default
+    default DigestGenStrategy digestGenStrategy()
+    {
+        return NoDigestGenStrategy.builder().build();
+    }
 
     Auditing auditing();
 

@@ -70,10 +70,7 @@ public Dataset visitAppendOnly(AppendOnlyAbstract appendOnly)
     {
         boolean isAuditingFieldPK = doesDatasetContainsAnyPK(mainSchemaFields);
         appendOnly.auditing().accept(new EnrichSchemaWithAuditing(mainSchemaFields, isAuditingFieldPK));
-        if (appendOnly.digestField().isPresent())
-        {
-            addDigestField(mainSchemaFields, appendOnly.digestField().get());
-        }
+        appendOnly.digestGenStrategy().accept(IngestModeVisitors.EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY).ifPresent(digest -> addDigestField(mainSchemaFields, digest));
         removeDataSplitField(appendOnly.dataSplitField());
         return mainDatasetDefinitionBuilder.schema(mainSchemaDefinitionBuilder.addAllFields(mainSchemaFields).build()).build();
     }
@@ -138,11 +135,7 @@ public Dataset visitBitemporalDelta(BitemporalDeltaAbstract bitemporalDelta)
     @Override
     public Dataset visitBulkLoad(BulkLoadAbstract bulkLoad)
     {
-        Optional<String> digestField = bulkLoad.digestGenStrategy().accept(IngestModeVisitors.EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY);
-        if (digestField.isPresent())
-        {
-            addDigestField(mainSchemaFields, digestField.get());
-        }
+        bulkLoad.digestGenStrategy().accept(IngestModeVisitors.EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY).ifPresent(digest -> addDigestField(mainSchemaFields, digest));
         Field batchIdField = Field.builder()
                 .name(bulkLoad.batchIdField())
                 .type(FieldType.of(DataType.INT, Optional.empty(), Optional.empty()))

@@ -20,6 +20,8 @@
 import org.finos.legend.engine.persistence.components.ingestmode.audit.DateTimeAuditing;
 import org.finos.legend.engine.persistence.components.ingestmode.audit.NoAuditingAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.audit.AuditingVisitor;
+import org.finos.legend.engine.persistence.components.ingestmode.digest.UserProvidedDigestGenStrategy;
+import org.finos.legend.engine.persistence.components.ingestmode.digest.UserProvidedDigestGenStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.versioning.*;
 import org.finos.legend.engine.persistence.components.ingestmode.digest.DigestGenStrategy;
 import org.finos.legend.engine.persistence.components.ingestmode.digest.NoDigestGenStrategyAbstract;
@@ -75,7 +77,7 @@ public IngestMode visitAppendOnly(AppendOnlyAbstract appendOnly)
     {
         return AppendOnly
                 .builder()
-                .digestField(applyCase(appendOnly.digestField()))
+                .digestGenStrategy(appendOnly.digestGenStrategy().accept(new DigestGenStrategyCaseConverter()))
                 .auditing(appendOnly.auditing().accept(new AuditingCaseConverter()))
                 .deduplicationStrategy(appendOnly.deduplicationStrategy())
                 .versioningStrategy(appendOnly.versioningStrategy().accept(new VersionStrategyCaseConverter()))
@@ -244,6 +246,14 @@ public DigestGenStrategy visitUDFBasedDigestGenStrategy(UDFBasedDigestGenStrateg
                     .digestField(applyCase(udfBasedDigestGenStrategy.digestField()))
                     .build();
         }
+
+        @Override
+        public DigestGenStrategy visitUserProvidedDigestGenStrategy(UserProvidedDigestGenStrategyAbstract userProvidedDigestGenStrategy)
+        {
+            return UserProvidedDigestGenStrategy.builder()
+                .digestField(applyCase(userProvidedDigestGenStrategy.digestField()))
+                .build();
+        }
     }
 
 

@@ -18,6 +18,7 @@
 import org.finos.legend.engine.persistence.components.ingestmode.digest.DigestGenStrategyVisitor;
 import org.finos.legend.engine.persistence.components.ingestmode.digest.NoDigestGenStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.digest.UDFBasedDigestGenStrategyAbstract;
+import org.finos.legend.engine.persistence.components.ingestmode.digest.UserProvidedDigestGenStrategyAbstract;
 import org.finos.legend.engine.persistence.components.ingestmode.merge.MergeStrategyVisitors;
 
 import java.util.Collections;
@@ -37,7 +38,7 @@ private IngestModeVisitors()
         @Override
         public Boolean visitAppendOnly(AppendOnlyAbstract appendOnly)
         {
-            return appendOnly.filterExistingRecords();
+            return appendOnly.filterExistingRecords() || appendOnly.digestGenStrategy().accept(DIGEST_GEN_STRATEGY_DIGEST_REQUIRED);
         }
 
         @Override
@@ -88,7 +89,7 @@ public Boolean visitBulkLoad(BulkLoadAbstract bulkLoad)
         @Override
         public Optional<String> visitAppendOnly(AppendOnlyAbstract appendOnly)
         {
-            return appendOnly.digestField();
+            return appendOnly.digestGenStrategy().accept(EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY);
         }
 
         @Override
@@ -140,7 +141,7 @@ public Optional<String> visitBulkLoad(BulkLoadAbstract bulkLoad)
         public Set<String> visitAppendOnly(AppendOnlyAbstract appendOnly)
         {
             Set<String> metaFields = new HashSet<>();
-            appendOnly.digestField().ifPresent(metaFields::add);
+            appendOnly.digestGenStrategy().accept(EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY).ifPresent(metaFields::add);
             appendOnly.dataSplitField().ifPresent(metaFields::add);
             return metaFields;
         }
@@ -374,6 +375,12 @@ public Boolean visitUDFBasedDigestGenStrategy(UDFBasedDigestGenStrategyAbstract
         {
             return true;
         }
+
+        @Override
+        public Boolean visitUserProvidedDigestGenStrategy(UserProvidedDigestGenStrategyAbstract userProvidedDigestGenStrategy)
+        {
+            return true;
+        }
     };
 
     public static final DigestGenStrategyVisitor<Optional<String>> EXTRACT_DIGEST_FIELD_FROM_DIGEST_GEN_STRATEGY = new DigestGenStrategyVisitor<Optional<String>>()
@@ -389,6 +396,12 @@ public Optional<String> visitUDFBasedDigestGenStrategy(UDFBasedDigestGenStrategy
         {
             return Optional.of(udfBasedDigestGenStrategy.digestField());
         }
+
+        @Override
+        public Optional<String> visitUserProvidedDigestGenStrategy(UserProvidedDigestGenStrategyAbstract userProvidedDigestGenStrategy)
+        {
+            return Optional.of(userProvidedDigestGenStrategy.digestField());
+        }
     };
 
 }
@@ -19,4 +19,6 @@ public interface DigestGenStrategyVisitor<T>
     T visitNoDigestGenStrategy(NoDigestGenStrategyAbstract noDigestGenStrategy);
 
     T visitUDFBasedDigestGenStrategy(UDFBasedDigestGenStrategyAbstract udfBasedDigestGenStrategy);
+
+    T visitUserProvidedDigestGenStrategy(UserProvidedDigestGenStrategyAbstract userProvidedDigestGenStrategy);
 }
diff --git a/...ava/org/finos/legend/engine/persistence/components/ingestmode/digest/DigestGenerator.java b/...ava/org/finos/legend/engine/persistence/components/ingestmode/digest/DigestGenerator.java
@@ -0,0 +1,69 @@
+// Copyright 2023 Goldman Sachs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package org.finos.legend.engine.persistence.components.ingestmode.digest;
+
+import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
+import org.finos.legend.engine.persistence.components.logicalplan.values.DigestUdf;
+import org.finos.legend.engine.persistence.components.logicalplan.values.FieldValue;
+import org.finos.legend.engine.persistence.components.logicalplan.values.Value;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DigestGenerator implements DigestGenStrategyVisitor<Void>
+{
+    private List<Value> fieldsToSelect;
+    private List<Value> fieldsToInsert;
+    private List<Value> fieldsForDigest;
+    private Dataset stagingDataset;
+    private Dataset mainDataset;
+
+    public DigestGenerator(Dataset mainDataset, Dataset stagingDataset, List<Value> fieldsToSelect, List<Value> fieldsToInsert, List<Value> fieldsForDigest)
+    {
+        this.mainDataset = mainDataset;
+        this.stagingDataset = stagingDataset;
+        this.fieldsToSelect = fieldsToSelect;
+        this.fieldsToInsert = fieldsToInsert;
+        this.fieldsForDigest = fieldsForDigest;
+    }
+
+    @Override
+    public Void visitNoDigestGenStrategy(NoDigestGenStrategyAbstract noDigestGenStrategy)
+    {
+        return null;
+    }
+
+    @Override
+    public Void visitUDFBasedDigestGenStrategy(UDFBasedDigestGenStrategyAbstract udfBasedDigestGenStrategy)
+    {
+        Value digestValue = DigestUdf
+            .builder()
+            .udfName(udfBasedDigestGenStrategy.digestUdfName())
+            .addAllFieldNames(stagingDataset.schemaReference().fieldValues().stream().map(fieldValue -> fieldValue.fieldName()).collect(Collectors.toList()))
+            .addAllValues(fieldsForDigest)
+            .dataset(stagingDataset)
+            .build();
+        String digestField = udfBasedDigestGenStrategy.digestField();
+        fieldsToInsert.add(FieldValue.builder().datasetRef(mainDataset.datasetReference()).fieldName(digestField).build());
+        fieldsToSelect.add(digestValue);
+        return null;
+    }
+
+    @Override
+    public Void visitUserProvidedDigestGenStrategy(UserProvidedDigestGenStrategyAbstract userProvidedDigestGenStrategy)
+    {
+        return null;
+    }
+}
@@ -12,22 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package org.finos.legend.engine.persistence.components.relational.snowflake.sqldom.common;
+package org.finos.legend.engine.persistence.components.ingestmode.digest;
 
-import org.finos.legend.engine.persistence.components.relational.sqldom.SqlGen;
+import org.immutables.value.Value;
 
-public class ExternalVolume implements SqlGen
+@Value.Immutable
+@Value.Style(
+        typeAbstract = "*Abstract",
+        typeImmutable = "*",
+        jdkOnly = true,
+        optionalAcceptNullable = true,
+        strictBuilder = true
+)
+public interface UserProvidedDigestGenStrategyAbstract extends DigestGenStrategy
 {
-    private String externalVolume;
-
-    public ExternalVolume(String externalVolume)
-    {
-        this.externalVolume = externalVolume;
-    }
+    String digestField();
 
     @Override
-    public void genSql(StringBuilder builder)
+    default <T> T accept(DigestGenStrategyVisitor<T> visitor)
     {
-        builder.append(String.format("with EXTERNAL_VOLUME = '%s'", externalVolume));
+        return visitor.visitUserProvidedDigestGenStrategy(this);
     }
 }
@@ -35,7 +35,7 @@ public interface DatasetAdditionalPropertiesAbstract extends LogicalPlanNode
 
     Optional<TableOrigin> tableOrigin();
 
-    Optional<String> externalVolume();
+    Optional<IcebergProperties> icebergProperties();
 
     Map<String, String> tags();
 }
@@ -0,0 +1,36 @@
+// Copyright 2023 Goldman Sachs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package org.finos.legend.engine.persistence.components.logicalplan.datasets;
+
+import org.finos.legend.engine.persistence.components.logicalplan.LogicalPlanNode;
+import org.immutables.value.Value.Immutable;
+import org.immutables.value.Value.Style;
+
+@Immutable
+@Style(
+    typeAbstract = "*Abstract",
+    typeImmutable = "*",
+    jdkOnly = true,
+    optionalAcceptNullable = true,
+    strictBuilder = true
+)
+public interface IcebergPropertiesAbstract extends LogicalPlanNode
+{
+    String catalog();
+
+    String externalVolume();
+
+    String baseLocation();
+}
@@ -14,9 +14,14 @@
 
 package org.finos.legend.engine.persistence.components.logicalplan.datasets;
 
+import org.finos.legend.engine.persistence.components.common.LoadOptions;
+
 import java.util.List;
+import java.util.Optional;
 
 public interface StagedFilesDatasetProperties
 {
     List<String> files();
+
+    Optional<LoadOptions> loadOptions();
 }
@@ -14,10 +14,13 @@
 
 package org.finos.legend.engine.persistence.components.logicalplan.operations;
 
+import org.finos.legend.engine.persistence.components.common.LoadOptions;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
 import org.finos.legend.engine.persistence.components.logicalplan.values.Value;
 
 import java.util.List;
+import java.util.Optional;
+
 import static org.immutables.value.Value.Immutable;
 import static org.immutables.value.Value.Parameter;
 import static org.immutables.value.Value.Style;
@@ -40,4 +43,7 @@ public interface CopyAbstract extends Operation
 
     @Parameter(order = 2)
     List<Value> fields();
+
+    @Parameter(order = 3)
+    Optional<LoadOptions> loadOptions();
 }
@@ -40,5 +40,6 @@ public enum FunctionName
     PARSE_DATETIME,
     OBJECT_CONSTRUCT,
     TO_VARIANT,
-    TO_JSON;
+    TO_JSON,
+    CONVERT;
 }