Merge pull request #1044 from atlanhq/FT-779

Adds full delta calculation logic for RAB and CAB

Author: cmgrote

Diff for: gradle/libs.versions.toml

@@ -1,15 +1,15 @@
 [versions]
 jackson = "2.18.1"
 slf4j = "2.0.16"
-elasticsearch = "8.16.0"
+elasticsearch = "8.16.1"
 freemarker = "2.3.33"
 classgraph = "4.8.179"
 testng = "7.10.2"
-log4j = "2.24.1"
+log4j = "2.24.2"
 wiremock = "3.9.2"
 jnanoid = "2.0.0"
 numaflow = "0.9.0"
-awssdk = "2.29.18"
+awssdk = "2.29.20"
 gcs = "26.50.0"
 system-stubs = "2.1.7"
 fastcsv = "3.4.0"

Diff for: package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/cache/ChecksumCache.kt

@@ -0,0 +1,35 @@
+/* SPDX-License-Identifier: Apache-2.0
+   Copyright 2024 Atlan Pte. Ltd. */
+package com.atlan.pkg.cache
+
+import com.atlan.cache.AbstractOffHeapCache
+import com.atlan.util.AssetBatch.AssetIdentity
+import java.nio.charset.StandardCharsets
+
+/**
+ * Generic class through which to cache any objects efficiently, off-heap, to avoid risking extreme
+ * memory usage.
+ */
+class ChecksumCache(
+    name: String?,
+) : AbstractOffHeapCache<AssetIdentity, String>(name) {
+    /** {@inheritDoc}  */
+    override fun serializeKey(key: AssetIdentity): ByteArray {
+        return key.toString().toByteArray(StandardCharsets.UTF_8)
+    }
+
+    /** {@inheritDoc}  */
+    override fun deserializeKey(bytes: ByteArray): AssetIdentity {
+        return AssetIdentity.fromString(String(bytes, StandardCharsets.UTF_8))
+    }
+
+    /** {@inheritDoc}  */
+    override fun serializeValue(value: String): ByteArray {
+        return value.toByteArray(StandardCharsets.UTF_8)
+    }
+
+    /** {@inheritDoc}  */
+    override fun deserializeValue(bytes: ByteArray): String {
+        return String(bytes, StandardCharsets.UTF_8)
+    }
+}

Diff for: package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/util/AssetResolver.kt

@@ -2,6 +2,10 @@
    Copyright 2024 Atlan Pte. Ltd. */
 package com.atlan.pkg.util
 
+import com.atlan.model.assets.Asset
+import com.atlan.util.AssetBatch.AssetIdentity
+import java.io.IOException
+
 /**
  * Interface for resolving asset identities entirely from CSV file input (no calls to Atlan).
  */
@@ -47,4 +51,36 @@ interface AssetResolver {
             return "$name/$type"
         }
     }
+
+    /**
+     * Resolve the asset represented by a row of values in a CSV to an asset identity.
+     *
+     * @param values row of values for that asset from the CSV
+     * @param header order of column names in the CSV file being processed
+     * @param connectionsMap cache of connection qualifiedNames, keyed by connection identity
+     * @return a unique asset identity for that row of the CSV
+     */
+    @Throws(IOException::class)
+    fun resolveAsset(
+        values: List<String>,
+        header: List<String>,
+        connectionsMap: Map<ConnectionIdentity, String>,
+    ): AssetIdentity? {
+        val typeIdx = header.indexOf(Asset.TYPE_NAME.atlanFieldName)
+        if (typeIdx < 0) {
+            throw IOException(
+                "Unable to find the column 'typeName'. This is a mandatory column in the input CSV.",
+            )
+        }
+        val typeName = values[typeIdx]
+        val qnDetails = getQualifiedNameDetails(values, header, typeName)
+        val agnosticQN = qnDetails.uniqueQN
+        val connectionIdentity = getConnectionIdentityFromQN(agnosticQN)
+        if (connectionIdentity != null && connectionsMap.containsKey(connectionIdentity)) {
+            val qualifiedName =
+                agnosticQN.replaceFirst(connectionIdentity.toString(), connectionsMap[connectionIdentity]!!)
+            return AssetIdentity(typeName, qualifiedName)
+        }
+        return null
+    }
 }

Diff for: package-toolkit/runtime/src/main/kotlin/com/atlan/pkg/util/DeltaProcessor.kt

@@ -8,8 +8,10 @@ import com.atlan.pkg.cache.ConnectionCache
 import com.atlan.pkg.objectstore.ObjectStorageSyncer
 import com.atlan.pkg.serde.csv.CSVPreprocessor
 import com.atlan.pkg.serde.csv.RowPreprocessor
+import com.atlan.util.AssetBatch.AssetIdentity
 import mu.KLogger
 import java.io.File.separator
+import java.io.IOException
 import java.nio.file.Paths
 import java.time.Instant
 import java.time.ZoneId
@@ -27,6 +29,7 @@ import java.time.format.DateTimeFormatter
  * @param preprocessedDetails details retrieved from pre-processing the input CSV file
  * @param typesToRemove limit the asset types that will be removed to this collection of type names
  * @param logger for logging
+ * @param reloadSemantic the type of reload to do for assets (default {@code all} will reload any assets that are not deleted, whether changed or not)
  * @param previousFilePreprocessor responsible for pre-processing the previous CSV file (if provided directly)
  * @param outputDirectory local directory where files can be written and compared
  * @param previousFileProcessedExtension extension to use in the object store for files that have been processed
@@ -40,59 +43,110 @@ class DeltaProcessor(
     val preprocessedDetails: Results,
     val typesToRemove: Collection<String>,
     private val logger: KLogger,
+    val reloadSemantic: String = "all",
     val previousFilePreprocessor: CSVPreprocessor? = null,
     val outputDirectory: String = Paths.get(separator, "tmp").toString(),
     private val previousFileProcessedExtension: String = ".processed",
-) {
+) : AutoCloseable {
+    private val objectStore = Utils.getBackingStore(outputDirectory)
+    private var initialLoad: Boolean = true
+    private var delta: FileBasedDelta? = null
+    private var deletedAssets: OffHeapAssetCache? = null
+    private val reloadAll = reloadSemantic == "all"
+
     /**
-     * Run the delta detection.
-     * This includes: determining which assets should be deleted, deleting those assets, and updating
-     * the persistent connection cache by removing any deleted assets and creating / updating any assets
-     * that were created or modified.
-     *
-     * @param modifiedAssets list of assets that were modified by the processing up to the point of this delta detection
+     * Calculate any delta from the provided file context.
      */
-    fun run(modifiedAssets: OffHeapAssetCache? = null) {
-        var deletedAssets: OffHeapAssetCache? = null
+    fun calculate() {
         if (semantic == "full") {
             if (qualifiedNamePrefix.isNullOrBlank()) {
-                logger.warn { "Unable to determine qualifiedName prefix, will not delete any assets." }
+                logger.warn { "Unable to determine qualifiedName prefix, cannot calculate any delta." }
             } else {
                 val purgeAssets = removalType == "purge"
                 val assetRootName = preprocessedDetails.assetRootName
                 val previousFileLocation = "$previousFilesPrefix/$qualifiedNamePrefix"
-                val objectStore = Utils.getBackingStore(outputDirectory)
                 val previousFile =
                     if (previousFilePreprocessor != null && previousFilePreprocessor.filename.isNotBlank()) {
                         transformPreviousRaw(assetRootName, previousFilePreprocessor)
                     } else {
                         objectStore.copyLatestFrom(previousFileLocation, previousFileProcessedExtension, outputDirectory)
                     }
                 if (previousFile.isNotBlank()) {
-                    // If there was a previous file, calculate the delta to see what we need
-                    // to delete
-                    val assetRemover =
-                        AssetRemover(
+                    // If there was a previous file, calculate the delta (changes + deletions)
+                    initialLoad = false
+                    delta =
+                        FileBasedDelta(
                             ConnectionCache.getIdentityMap(),
                             resolver,
                             logger,
                             typesToRemove.toList(),
                             qualifiedNamePrefix,
                             purgeAssets,
+                            !reloadAll,
                             outputDirectory,
                         )
-                    assetRemover.calculateDeletions(preprocessedDetails.preprocessedFile, previousFile)
-                    if (assetRemover.hasAnythingToDelete()) {
-                        // Note: this will update the persistent connection cache for both adds and deletes
-                        deletedAssets = assetRemover.deleteAssets()
-                    }
+                    delta!!.calculateDelta(preprocessedDetails.preprocessedFile, previousFile)
                 } else {
                     logger.info { "No previous file found, treated it as an initial load." }
                 }
-                // Copy processed files to specified location in object storage for future comparison purposes
-                uploadToBackingStore(objectStore, preprocessedDetails.preprocessedFile, qualifiedNamePrefix, previousFileProcessedExtension)
             }
         }
+    }
+
+    /**
+     * Resolve the asset represented by a row of values in a CSV to an asset identity.
+     *
+     * @param values row of values for that asset from the CSV
+     * @param header order of column names in the CSV file being processed
+     * @return a unique asset identity for that row of the CSV
+     */
+    @Throws(IOException::class)
+    fun resolveAsset(
+        values: List<String>,
+        header: List<String>,
+    ): AssetIdentity? {
+        return delta?.resolveAsset(values, header)
+    }
+
+    /**
+     * Determine whether the provided asset identity should be processed (true) or skipped (false).
+     *
+     * @param identity of the asset to check whether reloading should occur
+     * @return true if the asset with this identity should be reloaded, otherwise false
+     */
+    fun reloadAsset(identity: AssetIdentity): Boolean {
+        if (!reloadAll) {
+            return delta?.assetsToReload?.containsKey(identity) ?: true
+        }
+        return true
+    }
+
+    /**
+     * Delete any assets that were detected by the delta to be deleted.
+     */
+    fun processDeletions() {
+        if (!initialLoad && delta!!.hasAnythingToDelete()) {
+            // Note: this will update the persistent connection cache for both adds and deletes
+            deletedAssets = delta!!.deleteAssets()
+        }
+    }
+
+    /**
+     * Upload the latest processed file to the backing store, to persist the state for the next run.
+     */
+    fun uploadStateToBackingStore() {
+        if (!qualifiedNamePrefix.isNullOrBlank()) {
+            // Copy processed files to specified location in object storage for future comparison purposes
+            uploadToBackingStore(objectStore, preprocessedDetails.preprocessedFile, qualifiedNamePrefix, previousFileProcessedExtension)
+        }
+    }
+
+    /**
+     * Update the persistent connection cache with details of any assets that were added or removed.
+     *
+     * @param modifiedAssets cache of assets that were created or modified (whether by initial processing or reloading)
+     */
+    fun updateConnectionCache(modifiedAssets: OffHeapAssetCache? = null) {
         // Update the connection cache with any changes (added and / or removed assets)
         Utils.updateConnectionCache(
             added = modifiedAssets,
@@ -101,6 +155,33 @@ class DeltaProcessor(
         )
     }
 
+    /** {@inheritDoc} */
+    override fun close() {
+        uploadStateToBackingStore()
+        var exception: Exception? = null
+        if (delta != null) {
+            try {
+                delta!!.close()
+            } catch (e: Exception) {
+                exception = e
+            }
+        }
+        if (deletedAssets != null) {
+            try {
+                deletedAssets!!.close()
+            } catch (e: Exception) {
+                if (exception != null) {
+                    exception.addSuppressed(e)
+                } else {
+                    exception = e
+                }
+            }
+        }
+        if (exception != null) {
+            throw exception
+        }
+    }
+
     /**
      * Upload the file used to load the assets to Atlan backing store.
      *