diff --git a/samples/packages/lineage-builder/src/main/kotlin/com/atlan/pkg/lb/Loader.kt b/samples/packages/lineage-builder/src/main/kotlin/com/atlan/pkg/lb/Loader.kt index 75ac951d6f..702ec01a43 100644 --- a/samples/packages/lineage-builder/src/main/kotlin/com/atlan/pkg/lb/Loader.kt +++ b/samples/packages/lineage-builder/src/main/kotlin/com/atlan/pkg/lb/Loader.kt @@ -25,6 +25,13 @@ import kotlin.system.exitProcess object Loader { private val logger = Utils.getLogger(Loader.javaClass.name) + /** + * Object store prefix for the files this package saves after each run. It must not be the same as + * Asset Import's own prefix, or a Full Replacement import on the same connection compares its file + * against the lineage files saved here instead of its own previous file. + */ + const val PREVIOUS_FILES_PREFIX = "csa-lineage-builder" + @JvmStatic fun main(args: Array) { val outputDirectory = if (args.isEmpty()) "tmp" else args[0] @@ -93,6 +100,7 @@ object Loader { assetsCmHandling = CustomMetadataHandling.IGNORE.value, assetsTagHandling = AtlanTagHandling.IGNORE.value, relaxedCqn = ctx.config.relaxedCqn, + assetsPreviousFilePrefix = PREVIOUS_FILES_PREFIX, ) lateinit var qualifiedNameMap: Map Utils.initializeContext(importConfig, ctx).use { iCtx -> @@ -144,6 +152,7 @@ object Loader { assetsCmHandling = ctx.config.cmHandling, assetsTagHandling = ctx.config.tagHandling, relaxedCqn = ctx.config.relaxedCqn, + assetsPreviousFilePrefix = PREVIOUS_FILES_PREFIX, ) Utils.initializeContext(lineageConfig, ctx).use { iCtx -> Importer.import(iCtx, outputDirectory)?.close() diff --git a/samples/packages/lineage-builder/src/test/kotlin/SeparatePreviousFilesTest.kt b/samples/packages/lineage-builder/src/test/kotlin/SeparatePreviousFilesTest.kt new file mode 100644 index 0000000000..ad0b12b21c --- /dev/null +++ b/samples/packages/lineage-builder/src/test/kotlin/SeparatePreviousFilesTest.kt @@ -0,0 +1,156 @@ +/* SPDX-License-Identifier: Apache-2.0 + Copyright 2023 Atlan Pte. Ltd. */ +import com.atlan.model.assets.Connection +import com.atlan.model.assets.LineageProcess +import com.atlan.model.assets.Table +import com.atlan.model.enums.AtlanConnectorType +import com.atlan.model.enums.AtlanStatus +import com.atlan.pkg.PackageTest +import com.atlan.pkg.Utils +import com.atlan.pkg.aim.Importer +import com.atlan.pkg.lb.Loader +import java.nio.file.Paths +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * Test that a Lineage Builder run between two Full Replacement runs of Asset Import on the same + * connection does not stop Asset Import from removing an asset that drops out of its file. + */ +class SeparatePreviousFilesTest : PackageTest("spf") { + override val logger = Utils.getLogger(this.javaClass.name) + + private val conn1 = makeUnique("c1") + private val conn1Type = AtlanConnectorType.IBM_DB2 + private lateinit var connection: Connection + + private fun createConnection(): Connection { + val c1 = Connection.creator(client, conn1, conn1Type).build() + val response = c1.save(client).block() + return response.getResult(c1) + } + + private fun prepAssets( + filename: String, + tables: List, + ): String { + val cq = connection.qualifiedName + val ct = conn1Type.value + val output = Paths.get(testDirectory, filename).toFile() + output.writeText( + "qualifiedName,typeName,name,connectorType,connectionQualifiedName," + + "databaseQualifiedName,databaseName,schemaQualifiedName,schemaName,description,database,schema\n", + ) + output.appendText("$cq/DB,Database,DB,$ct,$cq,,,,,,,\n") + output.appendText("$cq/DB/SCH,Schema,SCH,$ct,$cq,$cq/DB,DB,,,,Database@$cq/DB,\n") + tables.forEach { tbl -> + output.appendText("$cq/DB/SCH/$tbl,Table,$tbl,$ct,$cq,$cq/DB,DB,$cq/DB/SCH,SCH,,,Schema@$cq/DB/SCH\n") + } + return output.path + } + + private fun prepLineage(filename: String): String { + val ct = conn1Type.value + val output = Paths.get(testDirectory, filename).toFile() + output.writeText( + "Source Type,Source Connector,Source Connection,Source Identity,Source Name," + + "Target Type,Target Connector,Target Connection,Target Identity,Target Name," + + "Transformation Connector,Transformation Connection,Transformation Identity,Transformation Name\n", + ) + output.appendText("Table,$ct,$conn1,DB/SCH/T1,T1,Table,$ct,$conn1,DB/SCH/T2,T2,$ct,$conn1,DB/SCH/T1_to_T2,T1 > T2\n") + return output.path + } + + override fun setup() { + connection = createConnection() + // 1. Full Replacement import of T1, T2 and T3 + runCustomPackage( + AssetImportCfg( + assetsFile = prepAssets("assets1.csv", listOf("T1", "T2", "T3")), + assetsUpsertSemantic = "upsert", + assetsFailOnErrors = true, + assetsDeltaSemantic = "full", + ), + Importer::main, + ) + // 2. Lineage Builder on the same connection, between the two imports + runCustomPackage( + LineageBuilderCfg( + lineageFile = prepLineage("lineage.csv"), + lineageUpsertSemantic = "upsert", + lineageFailOnErrors = false, + ), + Loader::main, + ) + // 3. Full Replacement import again, with T3 dropped from the file + runCustomPackage( + AssetImportCfg( + assetsFile = prepAssets("assets2.csv", listOf("T1", "T2")), + assetsUpsertSemantic = "upsert", + assetsFailOnErrors = true, + assetsDeltaSemantic = "full", + ), + Importer::main, + ) + } + + override fun teardown() { + removeConnection(conn1, conn1Type) + } + + @Test + fun droppedTableArchived() { + val request = + Table + .select(client, true) + .where(Table.QUALIFIED_NAME.eq("${connection.qualifiedName}/DB/SCH/T3")) + .where(Table.STATUS.eq(AtlanStatus.DELETED)) + .includeOnResults(Table.STATUS) + .toRequest() + val response = retrySearchUntil(request, 1, true) + assertEquals(1, response.assets.size) + assertEquals(AtlanStatus.DELETED, response.assets[0].status) + } + + @Test + fun keptTablesStillActive() { + val request = + Table + .select(client) + .where(Table.CONNECTION_QUALIFIED_NAME.eq(connection.qualifiedName)) + .includeOnResults(Table.NAME) + .toRequest() + val response = retrySearchUntil(request, 2) + assertEquals(setOf("T1", "T2"), response.assets.map { it.name }.toSet()) + } + + @Test + fun lineageStillActive() { + val request = + LineageProcess + .select(client) + .where(LineageProcess.CONNECTION_QUALIFIED_NAME.eq(connection.qualifiedName)) + .includeOnResults(LineageProcess.STATUS) + .toRequest() + val response = retrySearchUntil(request, 1) + assertEquals(1, response.assets.size) + assertEquals(AtlanStatus.ACTIVE, response.assets[0].status) + } + + private fun savedFiles(prefix: String): List = + Paths + .get(testDirectory, prefix, connection.qualifiedName) + .toFile() + .listFiles() + ?.map { it.name } + ?.filter { it.endsWith(".processed") } + ?: listOf() + + @Test + fun eachPackageSavesToItsOwnPrefix() { + // Two Asset Import runs, one saved file each + assertEquals(2, savedFiles(Importer.PREVIOUS_FILES_PREFIX).size) + // One Lineage Builder run, one saved file for its assets and one for its lineage + assertEquals(2, savedFiles("csa-lineage-builder").size) + } +}