finos · rafaelbey · Oct 19, 2023 · Aug 14, 2023 · Aug 14, 2023 · Aug 16, 2023
@@ -19,5 +19,5 @@ public enum FileFormat
     CSV,
     JSON,
     AVRO,
-    PARQUET
+    PARQUET;
 }
@@ -12,20 +12,33 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package org.finos.legend.engine.persistence.components.logicalplan.values;
+package org.finos.legend.engine.persistence.components.common;
 
-import static org.immutables.value.Value.Immutable;
-import static org.immutables.value.Value.Style;
+import org.immutables.value.Value;
 
-@Immutable
-@Style(
+import java.util.Optional;
+
+@Value.Immutable
+@Value.Style(
     typeAbstract = "*Abstract",
     typeImmutable = "*",
     jdkOnly = true,
     optionalAcceptNullable = true,
     strictBuilder = true
 )
-public interface BulkLoadBatchIdValueAbstract extends Value
+public interface LoadOptionsAbstract
 {
-    BulkLoadBatchIdValue INSTANCE = BulkLoadBatchIdValue.builder().build();
+    Optional<String> fieldDelimiter();
+
+    Optional<String> encoding();
+
+    Optional<String> nullMarker();
+
+    Optional<String> quote();
+
+    Optional<Long> skipLeadingRows();
+
+    Optional<Long> maxBadRecords();
+
+    Optional<String> compression();
 }
@@ -149,7 +149,7 @@ public Dataset visitBulkLoad(BulkLoadAbstract bulkLoad)
         }
         Field batchIdField = Field.builder()
                 .name(bulkLoad.batchIdField())
-                .type(FieldType.of(DataType.VARCHAR, Optional.empty(), Optional.empty()))
+                .type(FieldType.of(DataType.INT, Optional.empty(), Optional.empty()))
                 .primaryKey(false)
                 .build();
         mainSchemaFields.add(batchIdField);

@@ -15,7 +15,8 @@
 package org.finos.legend.engine.persistence.components.logicalplan;
 
 import org.finos.legend.engine.persistence.components.common.Datasets;
-import org.finos.legend.engine.persistence.components.logicalplan.conditions.Condition;
+import org.finos.legend.engine.persistence.components.ingestmode.BulkLoad;
+import org.finos.legend.engine.persistence.components.ingestmode.IngestMode;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.CsvExternalDatasetReference;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.Selection;
@@ -32,12 +33,13 @@
 import org.finos.legend.engine.persistence.components.logicalplan.values.StringValue;
 import org.finos.legend.engine.persistence.components.logicalplan.values.TabularValues;
 import org.finos.legend.engine.persistence.components.logicalplan.values.Value;
+import org.finos.legend.engine.persistence.components.util.BulkLoadMetadataDataset;
+import org.finos.legend.engine.persistence.components.util.BulkLoadMetadataUtils;
 import org.finos.legend.engine.persistence.components.util.LogicalPlanUtils;
 import org.finos.legend.engine.persistence.components.util.MetadataDataset;
 import org.finos.legend.engine.persistence.components.util.MetadataUtils;
 
 import java.util.List;
-import java.util.Optional;
 
 public class LogicalPlanFactory
 {
@@ -91,14 +93,23 @@ public static LogicalPlan getLogicalPlanForConstantStats(String stats, Long valu
             .build();
     }
 
-    public static LogicalPlan getLogicalPlanForNextBatchId(Datasets datasets)
+    public static LogicalPlan getLogicalPlanForNextBatchId(Datasets datasets, IngestMode ingestMode)
     {
         StringValue mainTable = StringValue.of(datasets.mainDataset().datasetReference().name().orElseThrow(IllegalStateException::new));
-        MetadataDataset metadataDataset = datasets.metadataDataset().isPresent()
-            ? datasets.metadataDataset().get()
-            : MetadataDataset.builder().build();
-        MetadataUtils metadataUtils = new MetadataUtils(metadataDataset);
-        Selection selection = metadataUtils.getBatchId(mainTable).selection();
+        Selection selection;
+        if (ingestMode instanceof BulkLoad)
+        {
+            BulkLoadMetadataDataset bulkLoadMetadataDataset = datasets.bulkLoadMetadataDataset().orElse(BulkLoadMetadataDataset.builder().build());
+            BulkLoadMetadataUtils bulkLoadMetadataUtils = new BulkLoadMetadataUtils(bulkLoadMetadataDataset);
+            selection = bulkLoadMetadataUtils.getBatchId(mainTable).selection();
+        }
+        else
+        {
+            MetadataDataset metadataDataset = datasets.metadataDataset().orElse(MetadataDataset.builder().build());
+            MetadataUtils metadataUtils = new MetadataUtils(metadataDataset);
+            selection = metadataUtils.getBatchId(mainTable).selection();
+        }
+
         return LogicalPlan.builder().addOps(selection).build();
     }
 

@@ -14,7 +14,10 @@
 
 package org.finos.legend.engine.persistence.components.logicalplan.values;
 
+import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
+
 import java.util.List;
+import java.util.Optional;
 
 @org.immutables.value.Value.Immutable
 @org.immutables.value.Value.Style(
@@ -32,4 +35,6 @@ public interface DigestUdfAbstract extends Value
     List<String> fieldNames();
 
     List<Value> values();
+
+    Optional<Dataset> dataset();
 }
@@ -37,5 +37,6 @@ public enum FunctionName
     GENERATE_ARRAY,
     PARSE_DATETIME,
     OBJECT_CONSTRUCT,
-    TO_VARIANT;
+    TO_VARIANT,
+    TO_JSON;
 }
@@ -60,9 +60,9 @@ class AppendOnlyPlanner extends Planner
 {
     private final Optional<Condition> dataSplitInRangeCondition;
 
-    AppendOnlyPlanner(Datasets datasets, AppendOnly ingestMode, PlannerOptions plannerOptions)
+    AppendOnlyPlanner(Datasets datasets, AppendOnly ingestMode, PlannerOptions plannerOptions, Set<Capability> capabilities)
     {
-        super(datasets, ingestMode, plannerOptions);
+        super(datasets, ingestMode, plannerOptions, capabilities);
 
         // validate
         ingestMode.deduplicationStrategy().accept(new ValidatePrimaryKeys(primaryKeys, this::validatePrimaryKeysIsEmpty,
@@ -83,7 +83,7 @@ protected AppendOnly ingestMode()
     }
 
     @Override
-    public LogicalPlan buildLogicalPlanForIngest(Resources resources, Set<Capability> capabilities)
+    public LogicalPlan buildLogicalPlanForIngest(Resources resources)
     {
         List<Value> fieldsToSelect = new ArrayList<>(stagingDataset().schemaReference().fieldValues());
         List<Value> fieldsToInsert = new ArrayList<>(stagingDataset().schemaReference().fieldValues());

@@ -105,9 +105,9 @@ class BitemporalDeltaPlanner extends BitemporalPlanner
     private List<FieldValue> primaryKeyFieldsAndFromFieldForSelection;
     private List<FieldValue> dataFields;
 
-    BitemporalDeltaPlanner(Datasets datasets, BitemporalDelta ingestMode, PlannerOptions plannerOptions)
+    BitemporalDeltaPlanner(Datasets datasets, BitemporalDelta ingestMode, PlannerOptions plannerOptions, Set<Capability> capabilities)
     {
-        super(datasets, ingestMode, plannerOptions);
+        super(datasets, ingestMode, plannerOptions, capabilities);
 
         if (ingestMode().validityMilestoning().validityDerivation() instanceof SourceSpecifiesFromDateTime && ingestMode().deduplicationStrategy() instanceof FilterDuplicates)
         {
@@ -201,7 +201,7 @@ protected BitemporalDelta ingestMode()
     }
 
     @Override
-    public LogicalPlan buildLogicalPlanForIngest(Resources resources, Set<Capability> capabilities)
+    public LogicalPlan buildLogicalPlanForIngest(Resources resources)
     {
         List<Operation> operations = new ArrayList<>();
 

@@ -25,17 +25,19 @@
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.Dataset;
 import org.finos.legend.engine.persistence.components.logicalplan.values.FieldValue;
 import org.finos.legend.engine.persistence.components.logicalplan.values.Value;
+import org.finos.legend.engine.persistence.components.util.Capability;
 import org.finos.legend.engine.persistence.components.util.LogicalPlanUtils;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 abstract class BitemporalPlanner extends UnitemporalPlanner
 {
-    BitemporalPlanner(Datasets datasets, BitemporalMilestoned bitemporalMilestoned, PlannerOptions plannerOptions)
+    BitemporalPlanner(Datasets datasets, BitemporalMilestoned bitemporalMilestoned, PlannerOptions plannerOptions, Set<Capability> capabilities)
     {
-        super(datasets, bitemporalMilestoned, plannerOptions);
+        super(datasets, bitemporalMilestoned, plannerOptions, capabilities);
 
         // validate
         String targetValidDateTimeFrom = bitemporalMilestoned.validityMilestoning().accept(EXTRACT_TARGET_VALID_DATE_TIME_FROM);

@@ -45,9 +45,9 @@
 
 class BitemporalSnapshotPlanner extends BitemporalPlanner
 {
-    BitemporalSnapshotPlanner(Datasets datasets, BitemporalSnapshot ingestMode, PlannerOptions plannerOptions)
+    BitemporalSnapshotPlanner(Datasets datasets, BitemporalSnapshot ingestMode, PlannerOptions plannerOptions, Set<Capability> capabilities)
     {
-        super(datasets, ingestMode, plannerOptions);
+        super(datasets, ingestMode, plannerOptions, capabilities);
 
         // validate
 
@@ -67,7 +67,7 @@ protected BitemporalSnapshot ingestMode()
     }
 
     @Override
-    public LogicalPlan buildLogicalPlanForIngest(Resources resources, Set<Capability> capabilities)
+    public LogicalPlan buildLogicalPlanForIngest(Resources resources)
     {
         List<Pair<FieldValue, Value>> keyValuePairs = keyValuesForMilestoningUpdate();
-Original file line number
+Diff line change
@@ Expand Up / @@ -19,5 +19,5 @@ public enum FileFormat @@
         CSV,
         JSON,
         AVRO,
-        PARQUET
+        PARQUET;
     }