[fix](regression) fix index change wait timeout

regression-test/plugins/plugin_index_change.groovy

@@ -23,22 +23,44 @@ import java.util.regex.Pattern;
 
 def delta_time = 1000
 
-Suite.metaClass.wait_for_last_build_index_finish = {table_name, OpTimeout ->
-    def useTime = 0
-    for(int t = delta_time; t <= OpTimeout; t += delta_time){
-        def alter_res = sql """SHOW BUILD INDEX WHERE TableName = "${table_name}" ORDER BY CreateTime DESC LIMIT 1;"""
-        alter_res = alter_res.toString()
-        if(alter_res.contains("FINISHED")) {
-            logger.info(table_name + " latest alter job finished, detail: " + alter_res)
+Suite.metaClass.snapshot_build_index_job_ids = { table_name ->
+    def alter_res = sql """SHOW BUILD INDEX WHERE TableName = "${table_name}";"""
+    return alter_res.collect { it[0].toString() }.toSet()
+}
+
+Suite.metaClass.wait_for_new_build_index_jobs_finish = { table_name, OpTimeout, previous_job_ids ->
+    assertTrue(previous_job_ids != null, "previous build index job ids must be snapshotted before the operation")
+    def finished = false
+    def alter_res = []
+    for (int t = 0; t <= OpTimeout; t += delta_time) {
+        def all_jobs = sql """SHOW BUILD INDEX WHERE TableName = "${table_name}"
+                ORDER BY CreateTime DESC;"""
+        alter_res = all_jobs.findAll { !previous_job_ids.contains(it[0].toString()) }
+
+        if (!alter_res.isEmpty() && alter_res.any { it[7] == "CANCELLED" }) {
+            logger.info(table_name + " build index job failed, detail: " + alter_res)
+            assertTrue(false, "build index job cancelled, result: ${alter_res}")
+        }
+        if (!alter_res.isEmpty() && alter_res.every { it[7] == "FINISHED" }) {
+            logger.info(table_name + " build index jobs finished, detail: " + alter_res)
+            finished = true
+            break
+        }
+        if (t >= OpTimeout) {
             break
-        } else if (alter_res.contains("CANCELLED")) {
-            logger.info(table_name + " latest alter job failed, detail: " + alter_res)
-            assertTrue(false)
         }
-        useTime = t
         sleep(delta_time)
     }
-    assertTrue(useTime <= OpTimeout, "wait for last build index finish timeout")
+    assertTrue(finished, "wait for build index finish timeout, latest result: ${alter_res}")
+}
+
+// Run an operation that must create one or more IndexChangeJobs and wait for
+// every job created by that operation. SHOW BUILD INDEX history is retained, so
+// the pre-operation snapshot is required to isolate the current operation.
+Suite.metaClass.run_index_change_job_and_wait = { table_name, OpTimeout, Closure operation ->
+    def previous_job_ids = snapshot_build_index_job_ids(table_name)
+    operation.call()
+    wait_for_new_build_index_jobs_finish(table_name, OpTimeout, previous_job_ids)
 }
 
 Suite.metaClass.build_index_on_table = {index_name, table_name ->
@@ -51,25 +73,28 @@ Suite.metaClass.build_index_on_table = {index_name, table_name ->
 }
 
 Suite.metaClass.wait_for_last_col_change_finish = { table_name, OpTimeout ->
-    def useTime = 0
+    def finished = false
+    def alter_res = ""
 
-    for (int t = delta_time; t <= OpTimeout; t += delta_time) {
-        def alter_res = sql """SHOW ALTER TABLE COLUMN WHERE TableName = "${table_name}" ORDER BY CreateTime DESC LIMIT 1;"""
+    for (int t = 0; t <= OpTimeout; t += delta_time) {
+        alter_res = sql """SHOW ALTER TABLE COLUMN WHERE TableName = "${table_name}" ORDER BY CreateTime DESC LIMIT 1;"""
         alter_res = alter_res.toString()
         if (alter_res.contains("FINISHED")) {
             sleep(3000) // wait change table state to normal
             logger.info(table_name + " latest alter job finished, detail: " + alter_res)
+            finished = true
+            break
+        }
+        if (t >= OpTimeout) {
             break
         }
-        useTime = t
         sleep(delta_time)
     }
-    assertTrue(useTime <= OpTimeout, "wait_for_last_col_change_finish timeout")
+    assertTrue(finished, "wait_for_last_col_change_finish timeout, latest result: ${alter_res}")
 }
 
 Suite.metaClass.wait_for_last_schema_change_finish = {table_name, OpTimeout ->
     wait_for_last_col_change_finish(table_name, OpTimeout)
-    wait_for_last_build_index_finish(table_name, OpTimeout)
 }
 
 
@@ -137,4 +162,4 @@ Suite.metaClass.check_bf_index_filter_rows = { sql, expectedRowsBfFiltered ->
         log.info("filter number:{}", number)
         assertEquals(expectedRowsBfFiltered, number)
     }
-}
+}

regression-test/suites/index_p0/test_ngram_bloomfilter_index_change.groovy

@@ -19,9 +19,6 @@ import groovy.json.JsonSlurper
 suite("test_ngram_bloomfilter_index_change") {
     def tableName = 'test_ngram_bloomfilter_index_change'
     def timeout = 60000
-    def delta_time = 1000
-    def alter_res = "null"
-    def useTime = 0
 
     // Helper functions to fetch profile via HTTP API
     def getProfileList = {
@@ -81,21 +78,6 @@ suite("test_ngram_bloomfilter_index_change") {
         return null
     }
 
-    def wait_for_latest_op_on_table_finish = { table_name, OpTimeout ->
-        for(int t = delta_time; t <= OpTimeout; t += delta_time){
-            alter_res = sql """SHOW ALTER TABLE COLUMN WHERE TableName = "${table_name}" ORDER BY CreateTime DESC LIMIT 1;"""
-            alter_res = alter_res.toString()
-            if(alter_res.contains("FINISHED")) {
-                sleep(3000) // wait change table state to normal
-                logger.info(table_name + " latest alter job finished, detail: " + alter_res)
-                break
-            }
-            useTime = t
-            sleep(delta_time)
-        }
-        assertTrue(useTime <= OpTimeout, "wait_for_latest_op_on_table_finish timeout")
-    }
-
     // Function to insert test data batch
     def insertTestData = { ->
         // insert 10 records
@@ -181,9 +163,12 @@ suite("test_ngram_bloomfilter_index_change") {
     assertTrue(filtered2 != null && filtered2 == 10, "Expected RowsBloomFilterFiltered = 10, but got ${filtered2}")
 
     // Drop index
+    def previousJobIds = isCloudMode() ? snapshot_build_index_job_ids(tableName) : null
     sql "DROP INDEX idx_ngram_customer_name ON ${tableName};"
-    wait_for_latest_op_on_table_finish(tableName, timeout)
-    wait_for_last_build_index_finish(tableName, timeout)
+    wait_for_last_col_change_finish(tableName, timeout)
+    if (isCloudMode()) {
+        wait_for_new_build_index_jobs_finish(tableName, timeout, previousJobIds)
+    }
 
     // Test after dropping index
     def token3 = UUID.randomUUID().toString()
@@ -237,8 +222,7 @@ suite("test_ngram_bloomfilter_index_change") {
 
     // Add NGRAM Bloom Filter index (will trigger schema change in this mode)
     sql "ALTER TABLE ${tableName} ADD INDEX idx_ngram_customer_name(customer_name) USING NGRAM_BF PROPERTIES('bf_size' = '1024', 'gram_size' = '3');"
-    wait_for_latest_op_on_table_finish(tableName, timeout)
-    wait_for_last_build_index_finish(tableName, timeout)
+    wait_for_last_col_change_finish(tableName, timeout)
 
     // Test after adding NGRAM Bloom Filter index (should filter existing data)
     def token5 = UUID.randomUUID().toString()
@@ -263,8 +247,7 @@ suite("test_ngram_bloomfilter_index_change") {
 
     // Drop index
     sql "DROP INDEX idx_ngram_customer_name ON ${tableName};"
-    wait_for_latest_op_on_table_finish(tableName, timeout)
-    wait_for_last_build_index_finish(tableName, timeout)
+    wait_for_last_col_change_finish(tableName, timeout)
 
     // Test after dropping index
     def token7 = UUID.randomUUID().toString()
@@ -306,8 +289,7 @@ suite("test_ngram_bloomfilter_index_change") {
 
     // Add ngram bf index before data insertion
     sql "ALTER TABLE ${tableName} ADD INDEX idx_ngram_customer_name(customer_name) USING NGRAM_BF PROPERTIES('bf_size' = '1024', 'gram_size' = '3');"
-    wait_for_latest_op_on_table_finish(tableName, timeout)
-    wait_for_last_build_index_finish(tableName, timeout)
+    wait_for_last_col_change_finish(tableName, timeout)
 
     // Insert data after index creation
     insertTestData()

regression-test/suites/inverted_index_p0/analyzer/test_custom_analyzer.groovy

@@ -461,9 +461,9 @@ suite("test_custom_analyzer", "p0") {
     qt_sql """ select tokenize("BAR", '"analyzer"="lowercase_delimited"'); """
 
     sql """ alter table ${indexTbName1} add index idx_ch_default(`ch`)  using inverted; """
-    wait_for_last_build_index_finish("${indexTbName1}", 60000)
+    wait_for_last_col_change_finish("${indexTbName1}", 60000)
     sql """ alter table ${indexTbName1} add index idx_ch(`ch`) using inverted properties("support_phrase" = "true", "analyzer" = "lowercase_delimited"); """
-    wait_for_last_build_index_finish("${indexTbName1}", 60000)
+    wait_for_last_col_change_finish("${indexTbName1}", 60000)
 
     qt_sql """ select * from ${indexTbName1} where ch match_all 'FOO'; """
     qt_sql """ select * from ${indexTbName1} where ch match_all 'BAR'; """
@@ -883,4 +883,4 @@ suite("test_custom_analyzer", "p0") {
     } finally {
         sql "DROP TABLE IF EXISTS ${indexTbName7}"
     }
-}
+}

regression-test/suites/inverted_index_p0/cloud_build_index/test_cloud_build_index_basic.groovy

@@ -57,25 +57,29 @@ suite("test_cloud_build_index_basic"){
     sql """create index idx1 on test_cloud_build_index_basic_table(address) using inverted;"""
     check_inverted_index_filter_rows("select * from test_cloud_build_index_basic_table where address='dddd'", 0)
 
-    build_index_on_table("idx1", tabName1)
-    wait_for_last_build_index_finish(tabName1, timeout)
+    run_index_change_job_and_wait(tabName1, timeout) {
+        build_index_on_table("idx1", tabName1)
+    }
     check_inverted_index_filter_rows("select * from test_cloud_build_index_basic_table where address='dddd'", 4)
 
-    sql "drop index idx1 on test_cloud_build_index_basic_table"
-    wait_for_last_build_index_finish(tabName1, timeout)
+    run_index_change_job_and_wait(tabName1, timeout) {
+        sql "drop index idx1 on test_cloud_build_index_basic_table"
+    }
     check_inverted_index_filter_rows("select * from test_cloud_build_index_basic_table where address='dddd'", 0)
 
     // 2 string type, ngram index
     sql "set enable_function_pushdown = true;"
     sql """create index idx2 on test_cloud_build_index_basic_table(address) using ngram_bf properties("gram_size"="3", "bf_size"="256");"""
     check_bf_index_filter_rows("select * from test_cloud_build_index_basic_table where address like '%sdf%'", 0)
 
-    build_index_on_table("idx2", tabName1)
-    wait_for_last_build_index_finish(tabName1, timeout)
+    run_index_change_job_and_wait(tabName1, timeout) {
+        build_index_on_table("idx2", tabName1)
+    }
     check_bf_index_filter_rows("select * from test_cloud_build_index_basic_table where address like '%sdf%'", 5)
 
-    sql "drop index idx2 on test_cloud_build_index_basic_table"
-    wait_for_last_build_index_finish(tabName1, timeout)
+    run_index_change_job_and_wait(tabName1, timeout) {
+        sql "drop index idx2 on test_cloud_build_index_basic_table"
+    }
     check_bf_index_filter_rows("select * from test_cloud_build_index_basic_table where address like '%sdf%'", 0)
 
     // test add column/reset varchar len
@@ -121,18 +125,21 @@ suite("test_cloud_build_index_basic"){
     check_inverted_index_filter_rows("select * from test_cloud_light_sc_table where content='jj1'", 0)
 
     sql """alter table test_cloud_light_sc_table modify column address varchar(400);"""
-    sql """build index on test_cloud_light_sc_table;"""
-    wait_for_last_build_index_finish("test_cloud_light_sc_table", timeout)
+    run_index_change_job_and_wait("test_cloud_light_sc_table", timeout) {
+        sql """build index on test_cloud_light_sc_table;"""
+    }
     check_inverted_index_filter_rows("select * from test_cloud_light_sc_table where address='eeeee'", 4)
     check_inverted_index_filter_rows("select * from test_cloud_light_sc_table where content='jj1'", 4)
 
-    sql """drop index idx1 on test_cloud_light_sc_table;"""
-    wait_for_last_build_index_finish("test_cloud_light_sc_table", timeout)
+    run_index_change_job_and_wait("test_cloud_light_sc_table", timeout) {
+        sql """drop index idx1 on test_cloud_light_sc_table;"""
+    }
     check_inverted_index_filter_rows("select * from test_cloud_light_sc_table where address='eeeee'", 0)
-    sql """drop index idx2 on test_cloud_light_sc_table;"""
-    wait_for_last_build_index_finish("test_cloud_light_sc_table", timeout)
+    run_index_change_job_and_wait("test_cloud_light_sc_table", timeout) {
+        sql """drop index idx2 on test_cloud_light_sc_table;"""
+    }
     check_inverted_index_filter_rows("select * from test_cloud_light_sc_table where content='jj1'", 0)
 
     qt_select_sc_output "select * from test_cloud_light_sc_table order by user_id,username,age,address,content"
 
-}
+}

regression-test/suites/inverted_index_p0/cloud_build_index/test_cloud_build_index_update.groovy

@@ -88,13 +88,15 @@ suite("test_cloud_build_index_update") {
     check_inverted_index_filter_rows("select * from test_cloud_build_idx_uq_table where address='hhhhh'" +
             " order by user_id,username,age,address", 0)
 
-    sql """build index on test_cloud_build_idx_uq_table"""
-    wait_for_last_build_index_finish("test_cloud_build_idx_uq_table", timeout)
+    run_index_change_job_and_wait("test_cloud_build_idx_uq_table", timeout) {
+        sql """build index on test_cloud_build_idx_uq_table"""
+    }
     check_inverted_index_filter_rows("select * from test_cloud_build_idx_uq_table where address='hhhhh'" +
             " order by user_id,username,age,address", 12)
 
-    sql """drop index idx1 on test_cloud_build_idx_uq_table"""
-    wait_for_last_build_index_finish("test_cloud_build_idx_uq_table", timeout)
+    run_index_change_job_and_wait("test_cloud_build_idx_uq_table", timeout) {
+        sql """drop index idx1 on test_cloud_build_idx_uq_table"""
+    }
 
     check_inverted_index_filter_rows("select * from test_cloud_build_idx_uq_table where address='hhhhh'" +
             " order by user_id,username,age,address", 0)
@@ -135,22 +137,24 @@ suite("test_cloud_build_index_update") {
 
     sql """create index idx1 on test_cloud_build_idx_del_tab(address) using inverted;"""
 
-    build_index_on_table("idx1", "test_cloud_build_idx_del_tab")
-    wait_for_last_build_index_finish("test_cloud_build_idx_del_tab", timeout)
+    run_index_change_job_and_wait("test_cloud_build_idx_del_tab", timeout) {
+        build_index_on_table("idx1", "test_cloud_build_idx_del_tab")
+    }
 
     check_inverted_index_filter_rows("select * from test_cloud_build_idx_del_tab" +
             " where address='eeeee' order by user_id,username,age,address", 12)
 
     qt_select_del_tab_ret3 """select count(1) from test_cloud_build_idx_del_tab where age=23;"""
     qt_select_del_tab_ret4 """select count(1) from test_cloud_build_idx_del_tab where age!=23;"""
 
-    sql """drop index idx1 on test_cloud_build_idx_del_tab;"""
-    wait_for_last_build_index_finish("test_cloud_build_idx_del_tab", timeout)
+    run_index_change_job_and_wait("test_cloud_build_idx_del_tab", timeout) {
+        sql """drop index idx1 on test_cloud_build_idx_del_tab;"""
+    }
 
     check_inverted_index_filter_rows("select * from test_cloud_build_idx_del_tab" +
             " where address='eeeee' order by user_id,username,age,address", 0)
 
     qt_select_del_tab_ret5 """select count(1) from test_cloud_build_idx_del_tab where age=23;"""
     qt_select_del_tab_ret6 """select count(1) from test_cloud_build_idx_del_tab where age!=23;"""
 
-}
+}