[fix](regression) wait for adaptive scan profile

regression-test/framework/src/main/groovy/org/apache/doris/regression/action/ProfileAction.groovy

@@ -26,6 +26,10 @@ import org.apache.doris.regression.util.JdbcUtils
 
 @Slf4j
 class ProfileAction implements SuiteAction {
+    private static final long DEFAULT_PROFILE_WAIT_TIMEOUT_MS = 10000
+    private static final long DEFAULT_PROFILE_WAIT_INTERVAL_MS = 500
+    private static final String PROFILE_COMPLETE = "Profile Completion State: COMPLETE"
+
     private String tag
     private Runnable runCallback
     private Closure<String> check
@@ -76,6 +80,26 @@ class ProfileAction implements SuiteAction {
         return profileData
     }
 
+    private String normalizeProfileText(String profileText) {
+        // Convert HTML entities and actual NBSPs to regular spaces,
+        // retain line breaks, and then compress multiple consecutive spaces into a single space.
+        profileText = profileText.replace("&nbsp;", " ")
+        profileText = profileText.replace("</br>", "\n")
+        profileText = profileText.replace('\u00A0' as char, ' ' as char)
+        return profileText.replaceAll(" {2,}", " ")
+    }
+
+    boolean isProfileReady(String profileText, List<String> requiredContents = []) {
+        if (profileText == null || profileText.isEmpty()) {
+            return false
+        }
+        if (requiredContents != null && !requiredContents.isEmpty()
+                && requiredContents.every { profileText.contains(it) }) {
+            return true
+        }
+        return profileText.contains(PROFILE_COMPLETE)
+    }
+
     String getProfile(String profileId) {
         def profileCli = new HttpCliAction(context)
         def addr = context.getFeHttpAddress()
@@ -97,20 +121,58 @@ class ProfileAction implements SuiteAction {
 
             def jsonSlurper2 = new JsonSlurper()
             def profileText = jsonSlurper2.parseText(profileResp).data
-            // Convert HTML entities and actual NBSPs to regular spaces, 
-            // retain line breaks, and then compress multiple consecutive spaces into a single space
-            profileText = profileText.replace("&nbsp;", " ")
-            profileText = profileText.replace("</br>", "\n")
-            // Replace the actual NBSP characters with regular spaces
-            profileText = profileText.replace('\u00A0' as char, ' ' as char)
-            // Compress two or more consecutive regular spaces into a single space (without affecting line breaks)
-            profileText = profileText.replaceAll(" {2,}", " ")
-            result.text = profileText
+            result.text = normalizeProfileText(profileText)
         }
         profileCli.run()
         return result.text
     }
 
+    String getProfile(String profileId, List<String> requiredContents, long timeoutMs = DEFAULT_PROFILE_WAIT_TIMEOUT_MS,
+            long intervalMs = DEFAULT_PROFILE_WAIT_INTERVAL_MS) {
+        String profileText = ""
+        long deadline = System.currentTimeMillis() + timeoutMs
+        while (System.currentTimeMillis() <= deadline) {
+            profileText = getProfile(profileId)
+            if (isProfileReady(profileText, requiredContents)) {
+                return profileText
+            }
+            log.info("Profile {} is not ready, required contents: {}", profileId, requiredContents)
+            Thread.sleep(intervalMs)
+        }
+        return profileText
+    }
+
+    String getProfileBySql(String sqlPattern, List<String> requiredContents = [],
+            long timeoutMs = DEFAULT_PROFILE_WAIT_TIMEOUT_MS, long intervalMs = DEFAULT_PROFILE_WAIT_INTERVAL_MS) {
+        String profileId = ""
+        String profileText = ""
+        long deadline = System.currentTimeMillis() + timeoutMs
+        while (System.currentTimeMillis() <= deadline) {
+            for (final def profileItem in getProfileList()) {
+                if (profileItem["Sql Statement"].toString().contains(sqlPattern)) {
+                    profileId = profileItem["Profile ID"].toString()
+                    long remainingMs = Math.max(1, deadline - System.currentTimeMillis())
+                    profileText = getProfile(profileId, requiredContents, remainingMs, intervalMs)
+                    if (isProfileReady(profileText, requiredContents)) {
+                        return profileText
+                    }
+                    break
+                }
+            }
+            if (profileId == "") {
+                log.info("Profile with sql pattern {} is not found yet", sqlPattern)
+            } else {
+                log.info("Profile {} with sql pattern {} is not ready", profileId, sqlPattern)
+            }
+            Thread.sleep(intervalMs)
+        }
+
+        if (profileId == "") {
+            throw new IllegalStateException("Missing profile with sql pattern: " + sqlPattern)
+        }
+        return profileText
+    }
+
     @Override
     void run() {
         if (runCallback.is(null)) {

regression-test/suites/query_profile/adaptive_pipeline_task_serial_read_on_limit.groovy

@@ -21,28 +21,13 @@ import groovy.json.StringEscapeUtils
 import org.apache.doris.regression.action.ProfileAction
 
 def verifyProfileContent = { suiteContext, stmt, serialReadOnLimit ->
-    // Sleep 1000ms to wait for the profile collection
-    Thread.sleep(1000)
-    // Get profile list by using getProfileList
     def profileAction = new ProfileAction(suiteContext)
-    List profileData = profileAction.getProfileList()
-    // Find the profile id for the query that we just emitted
-    String profileId = ""
-    for (def profileItem : profileData) {
-        if (profileItem["Sql Statement"].toString().contains(stmt)) {
-            profileId = profileItem["Profile ID"].toString()
-            logger.info("Profile ID of ${stmt} is ${profileId}")
-            break
-        }
-    }
-
-    if (profileId == "" || profileId == null) {
-        logger.error("Profile ID of ${stmt} is not found")
+    String profileContent = profileAction.getProfileBySql(stmt, ["MaxScanConcurrency"])
+    logger.info("Profile content of ${stmt} is\n${profileContent}")
+    if (!profileContent.contains("MaxScanConcurrency")) {
+        logger.error("Profile of ${stmt} does not contain MaxScanConcurrency, content:\n${profileContent}")
         return false
     }
-    // Get profile content by using getProfile
-    String profileContent = profileAction.getProfile(profileId)
-    logger.info("Profile content of ${stmt} is\n${profileContent}")
     // Check if the profile contains the expected content
     if (serialReadOnLimit) {
         return profileContent.contains("- MaxScanConcurrency: 1") == true

regression-test/suites/query_profile/scanner_profile.groovy

@@ -62,20 +62,11 @@ suite('scanner_profile') {
         select "${token}", * from scanner_profile limit 10;
     """
     def profileAction = new ProfileAction(context)
-    def getProfileByToken = { pattern ->
-        List profileData = profileAction.getProfileList()
-        def profileContent = ""
-        for (final def profileItem in profileData) {
-            if (profileItem["Sql Statement"].toString().contains(pattern)) {
-                profileContent = profileAction.getProfile(profileItem["Profile ID"].toString())
-                break
-            }
-        }
-        return profileContent        
+    def getProfileByToken = { pattern, requiredContents ->
+        return profileAction.getProfileBySql(pattern, requiredContents)
     }
 
-    List profileData = profileAction.getProfileList()
-    def profileWithLimit1 = getProfileByToken(token)
+    def profileWithLimit1 = getProfileByToken(token, ["TaskCpuTime", "MaxScanConcurrency"])
     logger.info("${token} Profile Data: ${profileWithLimit1}")
     assertTrue(profileWithLimit1.toString().contains("- TaskCpuTime:"))
     assertTrue(profileWithLimit1.toString().contains("- MaxScanConcurrency: 1"))
@@ -85,7 +76,7 @@ suite('scanner_profile') {
         select "${token}", * from scanner_profile where id < 10;
     """
 
-    String profileWithFilter = getProfileByToken(token)
+    String profileWithFilter = getProfileByToken(token, ["TaskCpuTime", "ScannerCpuTime"])
     logger.info("${token} Profile Data: ${profileWithFilter}")
     assertTrue(profileWithFilter.toString().contains("- TaskCpuTime:"))
     assertTrue(profileWithFilter.toString().contains("- ScannerCpuTime:"))