diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/pom.xml b/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/pom.xml new file mode 100644 index 00000000000..22911a6d785 --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/pom.xml @@ -0,0 +1,89 @@ + + + 4.0.0 + + org.finos.legend.engine + legend-engine-xts-ingest + 4.67.1-SNAPSHOT + + + legend-engine-xt-ingest-protocol + + + 11 + 11 + UTF-8 + + + + com.fasterxml.jackson.core + jackson-annotations + + + org.finos.legend.engine + legend-engine-executionPlan-dependencies + + + + + + org.codehaus.mojo + exec-maven-plugin + + true + + + + generate metamodel + generate-sources + + java + + + true + org.finos.legend.engine.protocol.generation.GenerateMetaClasses + + core_ingest_metamodel.protocol.json + ${project.build.directory}/generated-sources/ + + test + + + + + + org.finos.legend.engine + legend-engine-protocol-generation + ${project.version} + + + org.finos.legend.engine + legend-engine-xt-ingest-pure + ${project.version} + + + + + org.codehaus.mojo + build-helper-maven-plugin + + + add-source + generate-sources + + add-source + + + + ${project.basedir}/target/generated-sources/ + + + + + + + + + diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/src/main/resources/core_ingest_metamodel.protocol.json b/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/src/main/resources/core_ingest_metamodel.protocol.json new file mode 100644 index 00000000000..10c47469bf5 --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-protocol/src/main/resources/core_ingest_metamodel.protocol.json @@ -0,0 +1,7 @@ +{ + "purePackage": "meta::external::ingest::metamodel", + "javaPackage": "org.finos.legend.engine.protocol.ingest.metamodel", + "elementsToBeExcluded": [ + "meta::external::ingest::metamodel::IngestConfig" + ] +} \ No newline at end of file diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/pom.xml b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/pom.xml new file mode 100644 index 00000000000..4c9e354ec4a --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/pom.xml @@ -0,0 +1,168 @@ + + + 4.0.0 + + org.finos.legend.engine + legend-engine-xts-ingest + 4.67.1-SNAPSHOT + + + legend-engine-xt-ingest-pure + + + 11 + 11 + UTF-8 + + + jar + Legend Engine - XT - INGEST - PAR/JAVA + + + + + org.finos.legend.pure + legend-pure-maven-generation-par + + src/main/resources + ${legend.pure.version} + + core_ingest + + + ${project.basedir}/src/main/resources/core_ingest.definition.json + + + + + generate-sources + + build-pure-jar + + + + + + org.finos.legend.engine + legend-engine-xt-analytics-lineage-pure + ${project.version} + + + org.finos.legend.engine + legend-engine-pure-functions-unclassified-pure + ${project.version} + + + org.finos.legend.engine + legend-engine-xt-functionActivator-pure + ${project.version} + + + org.finos.legend.pure + legend-pure-m2-dsl-diagram-grammar + ${legend.pure.version} + + + org.finos.legend.engine + legend-engine-xt-relationalStore-core-pure + ${project.version} + + + + + org.finos.legend.pure + legend-pure-maven-generation-java + + + compile + + build-pure-compiled-jar + + + true + true + modular + true + + core_ingest + + + + + + + org.finos.legend.engine + legend-engine-pure-functions-unclassified-pure + ${project.version} + + + org.finos.legend.engine + legend-engine-xt-functionActivator-pure + ${project.version} + + + org.finos.legend.pure + legend-pure-m2-dsl-diagram-grammar + ${legend.pure.version} + + + org.finos.legend.engine + legend-engine-xt-relationalStore-core-pure + ${project.version} + + + + + + + + + org.finos.legend.pure + legend-pure-m3-core + + + org.finos.legend.pure + legend-pure-m4 + + + org.finos.legend.pure + legend-pure-runtime-java-engine-compiled + + + org.finos.legend.engine + legend-engine-pure-platform-java + + + + org.eclipse.collections + eclipse-collections-api + + + org.eclipse.collections + eclipse-collections + + + org.finos.legend.engine + legend-engine-pure-platform-dsl-store-java + + + org.finos.legend.engine + legend-engine-pure-platform-store-relational-java + + + org.finos.legend.engine + legend-engine-xt-relationalStore-core-pure + + + org.finos.legend.pure + legend-pure-m2-store-relational-pure + + + org.finos.legend.engine + legend-engine-pure-code-compiled-core + + + + diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/java/org/finos/legend/pure/code/core/CoreIngestCodeRepositoryProvider.java b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/java/org/finos/legend/pure/code/core/CoreIngestCodeRepositoryProvider.java new file mode 100644 index 00000000000..c50e0846e2c --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/java/org/finos/legend/pure/code/core/CoreIngestCodeRepositoryProvider.java @@ -0,0 +1,29 @@ +// 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.pure.code.core; + +import org.finos.legend.pure.m3.serialization.filesystem.repository.CodeRepository; +import org.finos.legend.pure.m3.serialization.filesystem.repository.CodeRepositoryProvider; +import org.finos.legend.pure.m3.serialization.filesystem.repository.GenericCodeRepository; + +public class CoreIngestCodeRepositoryProvider implements CodeRepositoryProvider +{ + @Override + public CodeRepository repository() + { + return GenericCodeRepository.build("core_ingest.definition.json"); + } +} + diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/META-INF/services/org.finos.legend.pure.m3.serialization.filesystem.repository.CodeRepositoryProvider b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/META-INF/services/org.finos.legend.pure.m3.serialization.filesystem.repository.CodeRepositoryProvider new file mode 100644 index 00000000000..c49e886cffa --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/META-INF/services/org.finos.legend.pure.m3.serialization.filesystem.repository.CodeRepositoryProvider @@ -0,0 +1 @@ +org.finos.legend.pure.code.core.CoreIngestCodeRepositoryProvider \ No newline at end of file diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest.definition.json b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest.definition.json new file mode 100644 index 00000000000..fa34287c0b1 --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest.definition.json @@ -0,0 +1,12 @@ +{ + "name": "core_ingest", + "pattern": "(meta::external::ingest)(::.*)?", + "dependencies": [ + "core", + "platform", + "platform_dsl_store", + "platform_store_relational", + "core_functions_json", + "core_relational" + ] +} \ No newline at end of file diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/barrier.pure b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/barrier.pure new file mode 100644 index 00000000000..0c0e6dfe0de --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/barrier.pure @@ -0,0 +1,30 @@ +import meta::external::ingest::barrier::*; +import meta::external::ingest::metamodel::*; + +function meta::external::ingest::barrier::isDatasetAvailable(ingestConfig: IngestConfig[1], prevPublishedWatermark: Watermark[1], currentStateWatermark: Watermark[1]):Boolean[1] +{ + let prevHighBatchId = getBatchId($ingestConfig.datasetName, $prevPublishedWatermark); + let currentBatchId = getBatchId($ingestConfig.datasetName, $currentStateWatermark); + ($currentBatchId > $prevHighBatchId) ; +} + +function meta::external::ingest::barrier::isDatasetAvailable(ingestConfig: IngestConfig[1],filterFunc:Function<{T[1]->Boolean[1]}>[1], prevPublishedWatermark: Watermark[1], currentStateWatermark: Watermark[1]):Boolean[1] +{ + let prevHighBatchId = getBatchId($ingestConfig.datasetName, $prevPublishedWatermark); + let currentBatchId = getBatchId($ingestConfig.datasetName, $currentStateWatermark); + +//TODO : Change to T , likely need to uplift getAll to handle T properly + meta::pure::functions::collection::getAll(BatchMetadata) + ->filter(x | $x.datasetName == $ingestConfig.datasetName) + ->filter(y | ($y.batchId > $prevHighBatchId) && ($y.batchId <= $currentBatchId)) + ->cast(@T) + ->filter($filterFunc) + ->isNotEmpty(); +} + + +function <> meta::external::ingest::barrier::getBatchId(datasetName: String[1], watermark: Watermark[1]):Integer[1] +{ + $watermark.datasetMarks->filter(w | $w.datasetName == $datasetName)->toOne().batchId; +} + diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/metamodel.pure b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/metamodel.pure new file mode 100644 index 00000000000..d054535cadb --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/metamodel.pure @@ -0,0 +1,37 @@ +import meta::external::ingest::metamodel::*; + +Class meta::external::ingest::metamodel::DatasetMark +{ + datasetName: String[1]; + batchId: Integer[1]; +} + +Class meta::external::ingest::metamodel::Watermark +{ + watermarkId: String[1]; + datasetMarks: DatasetMark[*]; +} + +Class meta::external::ingest::metamodel::IngestConfig +{ + datasetName: String[1]; + metadataType: Class[1]; +} + +Class meta::external::ingest::metamodel::BatchMetadata +{ + datasetName: String[1]; + batchId: Integer[1]; + ingest_request_id: String[1]; + ingest_definition_urn: String[1]; +} + +Class meta::external::ingest::metamodel::EodBatchMetadata extends meta::external::ingest::metamodel::BatchMetadata +{ + endOfDay: Boolean[1]; +} + +Class meta::external::ingest::metamodel::BusinessDateBatchMetadata extends meta::external::ingest::metamodel::EodBatchMetadata +{ + businessDate: StrictDate[1]; +} \ No newline at end of file diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/barrierTests.pure b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/barrierTests.pure new file mode 100644 index 00000000000..6ce4006d1ac --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/barrierTests.pure @@ -0,0 +1,100 @@ + +import meta::legend::*; +import meta::external::ingest::barrier::*; +import meta::external::ingest::metamodel::*; + +function <> meta::external::ingest::tests::testAvailabilityForSingleDataset() : Boolean[1] +{ + fail('Needs fix in plan generation and clustering'); + + let prevPublishedWatermark = ^Watermark(watermarkId='watermark1', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=1)]); + + let currentWM = ^Watermark(watermarkId='watermark2', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=3)]); + + let ingestConfig = ^IngestConfig(datasetName = 'Dataset1', metadataType = EodBatchMetadata); + + let barrierQuery = { + + prevPublishedWatermark: Watermark[1] , currentWatermark: Watermark[1] | isDatasetAvailable($ingestConfig, $prevPublishedWatermark, $currentWatermark); + + }; + + let result = executeLegendQuery($barrierQuery,[^Pair(first='prevPublishedWatermark',second=$prevPublishedWatermark),^Pair(first='currentWM',second=$currentWM)],^meta::pure::runtime::ExecutionContext(),meta::relational::extension::relationalExtensions()); + + assertEquals(true,$result); +} + +function <> meta::external::ingest::tests::testAvailabilityForMultipleDatasets() : Boolean[1] +{ + fail('Needs fix in plan generation and clustering'); + + let prevPublishedWatermark = ^Watermark(watermarkId='watermark1', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=1)]); + + let currentWM = ^Watermark(watermarkId='watermark2', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=3)]); + + let ingestConfig1 = ^IngestConfig(datasetName = 'Dataset1', metadataType = EodBatchMetadata); + + let ingestConfig2 = ^IngestConfig(datasetName = 'Dataset2', metadataType = EodBatchMetadata); + + let barrierQuery = { + + prevPublishedWatermark: Watermark[1] , currentWatermark: Watermark[1] | isDatasetAvailable($ingestConfig1, $prevPublishedWatermark, $currentWatermark) + && isDatasetAvailable($ingestConfig2, $prevPublishedWatermark, $currentWatermark); + }; + + let result = executeLegendQuery($barrierQuery,[^Pair(first='prevPublishedWatermark',second=$prevPublishedWatermark),^Pair(first='currentWM',second=$currentWM)],^meta::pure::runtime::ExecutionContext(),meta::relational::extension::relationalExtensions()); + + assertEquals(true,$result); +} + +function <> meta::external::ingest::tests::testAvailabilityForSingleDatasetWithFilter() : Boolean[1] +{ + fail('Needs fix in plan generation and clustering'); + + let prevPublishedWatermark = ^Watermark(watermarkId='watermark1', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=1)]); + + let currentWM = ^Watermark(watermarkId='watermark2', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=3)]); + + let ingestConfig = ^IngestConfig(datasetName = 'Dataset1', metadataType = EodBatchMetadata); + + let barrierQuery = { + + prevPublishedWatermark: Watermark[1] , currentWatermark: Watermark[1] | isDatasetAvailable($ingestConfig,x | $x.endOfDay, $prevPublishedWatermark, $currentWatermark); + + }; + + let result = executeLegendQuery($barrierQuery,[^Pair(first='prevPublishedWatermark',second=$prevPublishedWatermark),^Pair(first='currentWM',second=$currentWM)],^meta::pure::runtime::ExecutionContext(),meta::relational::extension::relationalExtensions()); + + assertEquals(true,$result); +} + +function <> meta::external::ingest::tests::testAvailabilityForMultipleDatasetsWithFilter() : Boolean[1] +{ + fail('Needs fix in plan generation and clustering'); + + let prevPublishedWatermark = ^Watermark(watermarkId='watermark1', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=1)]); + + let currentWM = ^Watermark(watermarkId='watermark2', + datasetMarks = [^DatasetMark(datasetName='Dataset1',batchId=3)]); + + let ingestConfig1 = ^IngestConfig(datasetName = 'Dataset1', metadataType = EodBatchMetadata); + + let ingestConfig2 = ^IngestConfig(datasetName = 'Dataset2', metadataType = EodBatchMetadata); + + let barrierQuery = { + + prevPublishedWatermark: Watermark[1] , currentWatermark: Watermark[1] | isDatasetAvailable($ingestConfig1,x | $x.endOfDay, $prevPublishedWatermark, $currentWatermark) + && isDatasetAvailable($ingestConfig2,x | $x.endOfDay, $prevPublishedWatermark, $currentWatermark); + }; + + let result = executeLegendQuery($barrierQuery,[^Pair(first='prevPublishedWatermark',second=$prevPublishedWatermark),^Pair(first='currentWM',second=$currentWM)],^meta::pure::runtime::ExecutionContext(),meta::relational::extension::relationalExtensions()); + + assertEquals(true,$result); +} \ No newline at end of file diff --git a/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/testSetup.pure b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/testSetup.pure new file mode 100644 index 00000000000..3cfe06452f8 --- /dev/null +++ b/legend-engine-xts-ingest/legend-engine-xt-ingest-pure/src/main/resources/core_ingest/tests/testSetup.pure @@ -0,0 +1,79 @@ + +function <> meta::external::ingest::tests::createTablesAndFillDb():Boolean[1] +{ + let connection = meta::external::ingest::tests::runtime::testRuntime().connectionByElement(meta::external::ingest::tests::store::BarrierTestStore)->cast(@meta::external::store::relational::runtime::TestDatabaseConnection); + meta::relational::metamodel::execute::executeInDb('Drop table if exists BatchMetadata;', $connection); + meta::relational::metamodel::execute::executeInDb('Create Table BatchMetadata(PKEY INT PRIMARY KEY,BATCH_ID INT ,DATASET_NAME VARCHAR(200),EOD BOOLEAN,INGEST_REQUEST_ID VARCHAR(200),INGEST_DEFINITION_ARN VARCHAR(200));', $connection); + meta::relational::metamodel::execute::executeInDb('insert into BatchMetadata (PKEY,BATCH_ID, DATASET_NAME, EOD, INGEST_REQUEST_ID,INGEST_DEFINITION_ARN ) values (1,1, \'Dataset1\',True, \'R1 \',\' \');', $connection); + meta::relational::metamodel::execute::executeInDb('insert into BatchMetadata (PKEY,BATCH_ID, DATASET_NAME, EOD, INGEST_REQUEST_ID,INGEST_DEFINITION_ARN ) values (4,2, \'Dataset1\',True, \'R2 \',\' \');', $connection); + meta::relational::metamodel::execute::executeInDb('insert into BatchMetadata (PKEY,BATCH_ID, DATASET_NAME, EOD, INGEST_REQUEST_ID,INGEST_DEFINITION_ARN ) values (5,3, \'Dataset1\',True, \'R3 \',\' \');', $connection); + meta::relational::metamodel::execute::executeInDb('insert into BatchMetadata (PKEY,BATCH_ID, DATASET_NAME, EOD, INGEST_REQUEST_ID,INGEST_DEFINITION_ARN ) values (6,3, \'Dataset1\',True, \'R4 \',\' \');', $connection); + meta::relational::metamodel::execute::executeInDb('insert into BatchMetadata (PKEY,BATCH_ID, DATASET_NAME, EOD, INGEST_REQUEST_ID,INGEST_DEFINITION_ARN ) values (8,3, \'Dataset2\',True, \'R5 \',\' \');', $connection); + + // let result = meta::relational::metamodel::execute::executeInDb('Select * from BatchMetadata ;', $connection); + // $result.rows->map(x | println($x.values)); + true; +} + +###Relational +Database meta::external::ingest::tests::store::BarrierTestStore +( + Table BatchMetadata ( + PKEY INT PRIMARY KEY, + BATCH_ID INT , + DATASET_NAME VARCHAR(200) , + EOD BIT, + INGEST_REQUEST_ID VARCHAR(200), + INGEST_DEFINITION_ARN VARCHAR(200) + + ) +) + +###Mapping +import meta::relational::tests::*; +import meta::external::store::relational::tests::*; +import meta::relational::tests::model::simple::*; + +Mapping meta::external::ingest::tests::mapping::BatchMetadataMapping +( + meta::external::ingest::metamodel::EodBatchMetadata: Relational{ + ~primaryKey + ( + [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.PKEY + ) + ~mainTable [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata + endOfDay:[meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.EOD, + batchId: [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.BATCH_ID, + datasetName: [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.DATASET_NAME, + ingest_request_id: [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.INGEST_REQUEST_ID, + ingest_definition_urn: [meta::external::ingest::tests::store::BarrierTestStore]BatchMetadata.INGEST_DEFINITION_ARN + } +) + +###Pure +import meta::relational::metamodel::*; +import meta::json::*; +import meta::json::tests::*; +import meta::relational::tests::*; +import meta::external::store::relational::tests::*; +import meta::pure::profiles::*; +import meta::relational::metamodel::execute::*; +import meta::core::runtime::*; +import meta::external::store::relational::runtime::*; +import meta::relational::runtime::*; + + +function meta::external::ingest::tests::runtime::testRuntime():Runtime[1] +{ + ^Runtime(connectionStores= meta::external::ingest::tests::testDatabaseConnection(meta::external::ingest::tests::store::BarrierTestStore,'GMT')) +} + +function meta::external::ingest::tests::testDatabaseConnection(db:Database[1], timeZone:String[0..1]):ConnectionStore[1] +{ + ^ConnectionStore( + connection=^meta::external::store::relational::runtime::TestDatabaseConnection( + type = DatabaseType.H2, + timeZone = if($timeZone->isEmpty(), |'GMT', |$timeZone) + ), + element = $db); +} \ No newline at end of file diff --git a/legend-engine-xts-ingest/pom.xml b/legend-engine-xts-ingest/pom.xml new file mode 100644 index 00000000000..8fc62e18ba0 --- /dev/null +++ b/legend-engine-xts-ingest/pom.xml @@ -0,0 +1,25 @@ + + + 4.0.0 + + org.finos.legend.engine + legend-engine + 4.67.1-SNAPSHOT + + + legend-engine-xts-ingest + pom + + legend-engine-xt-ingest-pure + legend-engine-xt-ingest-protocol + + + + 11 + 11 + UTF-8 + + + diff --git a/pom.xml b/pom.xml index 5920fdb4d20..d7ac3e4c52a 100644 --- a/pom.xml +++ b/pom.xml @@ -103,6 +103,7 @@ legend-engine-application-query legend-engine-xts-identity + legend-engine-xts-ingest