Enable Concurrent safety on Bulk Load (#2744)

finos · Apr 3, 2024 · 907b2a9 · 907b2a9
1 parent 5d2754a
commit 907b2a9
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 0 deletions.
diff --git a/...src/main/java/org/finos/legend/engine/persistence/components/planner/BulkLoadPlanner.java b/...src/main/java/org/finos/legend/engine/persistence/components/planner/BulkLoadPlanner.java
@@ -338,6 +338,10 @@ public LogicalPlan buildLogicalPlanForPreActions(Resources resources)
         {
             operations.add(Create.of(false, externalDataset));
         }
+        if (options().enableConcurrentSafety())
+        {
+            operations.add(Create.of(true, lockInfoDataset().orElseThrow(IllegalStateException::new).get()));
+        }
         return LogicalPlan.of(operations);
     }
 

diff --git a/...finos/legend/engine/persistence/components/relational/api/RelationalIngestorAbstract.java b/...finos/legend/engine/persistence/components/relational/api/RelationalIngestorAbstract.java
@@ -744,6 +744,7 @@ private List<IngestorResult> performBulkLoad(Datasets datasets, Transformer<SqlG
                                                  Executor<SqlGen, TabularData, SqlPlan> executor, GeneratorResult generatorResult,
                                                  IngestMode ingestMode, SchemaEvolutionResult schemaEvolutionResult)
     {
+        acquireLock();
         List<IngestorResult> results = new ArrayList<>();
         Map<String, PlaceholderValue> placeHolderKeyValues = extractPlaceHolderKeyValues(datasets, executor, planner, transformer, ingestMode, Optional.empty());