I need your help in designing the solution for databricks verification. My stack is Kotlin, TestNG, Gradle, Appium. In my test I do some actions which lead to the events saving in databricks. In current case there is 4 events and I need to find each of them, verify there is no duplicates and validate the data.
My current solution builds a long query which should be resulted to an exact one event and in case if one of criteria is invalid - we will not find any events. My concern is how good this solution in terms of optimization? Maybe it will be better to have a shorter query which will give us the same result (one event) and after that separately validate the field values and not rely on select filter to validate the actual field values of found event.

It is my test:

@Test(dataProvider = "firstBootFlowWithMmpPartners", dataProviderClass = DatabricksAttributionsEventsDataProvider::class)
    fun verifyAttributionEventsFbFlowAppsFlyerContentUriOff(testCaseId: String, packageName: String, appList: List<Map<String, String>>) {
        val configEdits = mapOf(
            CoreServiceData().clientConfigExtPath to
                mapOf(
                    "data.u_event_one_call_limit" to 5,
                ),
            LegacyAppServiceData().getAppsPath to mapOf(
                "Applications" to appList,
            ),
        )
        pushMocksStartReignite(configEdits, CoreServiceData().updateUEventPath)

        val events = listOf(
            Event.REFERRAL_COMPLETED_VALID_REFERRAL,
            Event.MMP_CLICKURL_FIRED,
            Event.DT_REVENUE_CLICKURL_FIRED,
            Event.DT_REVENUE_CLICK_URL_FIRED_RESPONSE_OK,
        )

        val dtid = LogCatReader.getDtid()

        val trackingUrl = getTrackingUrlFromAppList(appList, packageName)
        val campaignId = getCampaignIdFromTrackingUrl(trackingUrl ?: "")

        val eventCriteria = mapOf(
            "eventstring" to "MMP_ClickURL_Fired",
            "triggersource" to "legacy_flow_1",
            "lifecycle" to "OOB",
            "mechanism" to "boot",
            "experience" to "Dynamic Delivery",
            "recovery" to false,
            "discovery" to false,
            "trackingurl" to trackingUrl,
            "packagename" to packageName,
            "callingpackagename" to AppPackage.REIGNITE,
            "campaignid" to campaignId,
            "dtid" to dtid,
//            "metadata" to emptyMap<String, String>(),
        )


        DatabricksVerifier.verifyEventsExist(
            dtid,
            events,
            eventCriteria
        )
    }

verifyEventsExist method:

fun verifyEventsExist(
        dtid: String,
        events: List<Event>,
        additionalCriteria: Map<String, Any> = emptyMap(),
        checkDuplicates: Boolean = true,
        timeoutInSeconds: Long = 60,
    ) {
        events.forEach { event ->
            try {
                verifyEventExists(dtid, event, additionalCriteria, checkDuplicates, timeoutInSeconds)
            } catch (e: Exception) {
                logger.error(e) { "Failed to verify event ${event.name} for dtid = $dtid" }
                throw e
            }
        }
    }

verifyEventExists method:

/**
     * Polls Databricks until a specific event is found, verifying its field values.
     * It builds a query using all provided criteria to ensure the found event matches expectations.
     *
     * @param dtid The device identifier to filter by.
     * @param event The Event enum entry, used primarily for the event's name and code.
     * @param criteria A map where keys are column names and values are the exact expected values for verification.
     * @param checkDuplicates If true, the test will fail if more than one event matching the criteria is found.
     * @param timeoutInSeconds The maximum time to wait for the event.
     */
    @Step("Verify event '{event.name}' with criteria exists. Duplicate check: {checkDuplicates}")
    fun verifyEventExists(
        dtid: String,
        event: Event,
        criteria: Map<String, Any> = emptyMap(),
        checkDuplicates: Boolean = true,
        timeoutInSeconds: Long = 30,
    ) {
        // 1. Build a highly specific query from all provided criteria.
        val queryBuilder = DatabricksQueryBuilder(EVENTS_TABLE)
            .withDtid(dtid)
            .withEventCode(event.eventCode)
            .withDateAfter(LocalDate.now().format(DateTimeFormatter.ISO_LOCAL_DATE))
            .orderBy("receivedtimestamp", "desc")

        criteria.forEach { (key, value) ->
            val formattedValue = if (value is String) "'$value'" else value.toString()
            queryBuilder.withCustomClause("$key = $formattedValue")
        }

        // Select all columns to get the full event record.
        val query = queryBuilder.build(columns = "*")
        logger.info { "Executing verification query: $query" }

        // 2. Poll using Awaitility until the specific event is found.
        val foundEvents = await()
            .atMost(timeoutInSeconds, TimeUnit.SECONDS)
            .pollInterval(3, TimeUnit.SECONDS)
            .ignoreExceptions()
            .until({ DatabricksClient.executeQuery(query) }, { it.isNotEmpty() })

        // 3. Assert that at least one event matching all criteria was found.
        val fullCriteria = criteria + ("eventcode" to event.eventCode)
        logger.info { "Verifying the number of events matching criteria $fullCriteria for dtid = $dtid" }
        Assert.assertTrue(
            foundEvents.isNotEmpty(),
            "Event '${event.name}' with criteria $fullCriteria for dtid = $dtid was NOT FOUND.",
        )

        // 4. Optionally, check for duplicates by ensuring exactly one event was found.
        if (checkDuplicates) {
            Assert.assertEquals(
                foundEvents.size,
                1,
                "Duplicate Check Failed: Found ${foundEvents.size} events matching criteria $fullCriteria. Expected exactly 1.",
            )
            logger.info { "Uniqueness confirmed for event '${event.name}' for dtid=$dtid." }
        } else {
            logger.info { "Found ${foundEvents.size} event(s) matching criteria for '${event.name}' for dtid=$dtid." }
        }
    }