[enhance](regression) Add check_runtime_filter_partition_prune_counter to support testing runtime filter partition pruning

be/src/vec/exec/scan/vfile_scanner.cpp

@@ -1358,6 +1358,17 @@ Status VFileScanner::close(RuntimeState* state) {
         RETURN_IF_ERROR(_cur_reader->close());
     }
 
+    auto expected_counter = _state->query_options().check_runtime_filter_partition_prune_counter;
+    auto actual_counter = _runtime_filter_partition_pruned_range_counter->value();
+    if (expected_counter >= 0 && actual_counter > 0 && expected_counter != actual_counter) {
+        // because the close is called in the destructor, the Status::InternalError will be not handled
+        // so we just core dump here
+        DCHECK(false)
+                << "The check_runtime_filter_partition_prune_counter is set but not equal to the "
+                   "actual value: expected: "
+                << expected_counter << " actual: " << actual_counter;
+    }
+
     RETURN_IF_ERROR(VScanner::close(state));
     return Status::OK();
 }

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -362,8 +362,10 @@ public class SessionVariable implements Serializable, Writable {
     public static final String ENABLE_RUNTIME_FILTER_PRUNE =
             "enable_runtime_filter_prune";
 
-    public static final String ENABLE_RUNTIME_FILTER_PARTITION_PRUNE =
-            "enable_runtime_filter_partition_prune";
+    public static final String ENABLE_RUNTIME_FILTER_PARTITION_PRUNE = "enable_runtime_filter_partition_prune";
+
+    public static final String CHECK_RUNTIME_FILTER_PARTITION_PRUNE_COUNTER =
+            "check_runtime_filter_partition_prune_counter";
 
     static final String SESSION_CONTEXT = "session_context";
 
@@ -1525,9 +1527,18 @@ public void setEnableLeftZigZag(boolean enableLeftZigZag) {
     @VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PRUNE, needForward = true, fuzzy = true)
     public boolean enableRuntimeFilterPrune = true;
 
-    @VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PARTITION_PRUNE, needForward = true, fuzzy = true)
+    @VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PARTITION_PRUNE, needForward = true, fuzzy = true, description = {
+            "是否开启运行时过滤分区裁剪",
+            "Whether to enable runtime filter partition pruning."
+    })
     public boolean enableRuntimeFilterPartitionPrune = true;
 
+    @VariableMgr.VarAttr(name = CHECK_RUNTIME_FILTER_PARTITION_PRUNE_COUNTER, needForward = true, description = {
+            "检查运行时过滤分区裁剪的计数器，用于测试",
+            "Check the counter of runtime filter partition pruning, used for testing."
+    })
+    public long checkRuntimeFilterPartitionPruneCounter = -1;
+
     /**
      * The client can pass some special information by setting this session variable in the format: "k1:v1;k2:v2".
      * For example, trace_id can be passed to trace the query request sent by the user.
@@ -3756,6 +3767,14 @@ public void setEnableRuntimeFilterPartitionPrune(boolean enableRuntimeFilterPart
         this.enableRuntimeFilterPartitionPrune = enableRuntimeFilterPartitionPrune;
     }
 
+    public long getCheckRuntimeFilterPartitionPruneCounter() {
+        return checkRuntimeFilterPartitionPruneCounter;
+    }
+
+    public void setCheckRuntimeFilterPartitionPruneCounter(long checkRuntimeFilterPartitionPruneCounter) {
+        this.checkRuntimeFilterPartitionPruneCounter = checkRuntimeFilterPartitionPruneCounter;
+    }
+
     public void setFragmentTransmissionCompressionCodec(String codec) {
         this.fragmentTransmissionCompressionCodec = codec;
     }
@@ -4084,8 +4103,8 @@ public TQueryOptions toThrift() {
         tResult.setEnableFixedLenToUint32V2(enableFixedLenToUint32V2);
         tResult.setProfileLevel(getProfileLevel());
         tResult.setEnableRuntimeFilterPartitionPrune(enableRuntimeFilterPartitionPrune);
-
         tResult.setMinimumOperatorMemoryRequiredKb(minimumOperatorMemoryRequiredKB);
+        tResult.setCheckRuntimeFilterPartitionPruneCounter(checkRuntimeFilterPartitionPruneCounter);
         return tResult;
     }
 

gensrc/thrift/PaloInternalService.thrift

@@ -391,6 +391,8 @@ struct TQueryOptions {
   161: optional i64 low_memory_mode_buffer_limit = 33554432
   162: optional bool dump_heap_profile_when_mem_limit_exceeded = false
 
+  163: optional i64 check_runtime_filter_partition_prune_counter = -1;
+
   // For cloud, to control if the content would be written into file cache
   // In write path, to control if the content would be written into file cache.
   // In read path, read from file cache or remote storage when execute query.

regression-test/suites/external_table_p0/hive/test_hive_runtime_filter_partition_pruning.groovy

@@ -17,60 +17,71 @@
 
 suite("test_hive_runtime_filter_partition_pruning", "p0,external,hive,external_docker,external_docker_hive") {
     def test_runtime_filter_partition_pruning = {
+        sql """set runtime_filter_wait_time_ms = 10000"""
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_decimal1 """
             select count(*) from decimal_partition_table where partition_col =
                 (select partition_col from decimal_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 1"""
         qt_runtime_filter_partition_pruning_decimal2 """
             select count(*) from decimal_partition_table where partition_col in
                 (select partition_col from decimal_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 2);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_decimal3 """
             select count(*) from decimal_partition_table where abs(partition_col) =
                 (select partition_col from decimal_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_int1 """
             select count(*) from int_partition_table where partition_col =
                 (select partition_col from int_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 1"""
         qt_runtime_filter_partition_pruning_int2 """
             select count(*) from int_partition_table where partition_col in
                 (select partition_col from int_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 2);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_int3 """
             select count(*) from int_partition_table where abs(partition_col) =
                 (select partition_col from int_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_string1 """
             select count(*) from string_partition_table where partition_col =
                 (select partition_col from string_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 1"""
         qt_runtime_filter_partition_pruning_string2 """
             select count(*) from string_partition_table where partition_col in
                 (select partition_col from string_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 2);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 2"""
         qt_runtime_filter_partition_pruning_date1 """
             select count(*) from date_partition_table where partition_col =
                 (select partition_col from date_partition_table
                 group by partition_col having count(*) > 0
                 order by partition_col desc limit 1);
         """
+        sql """set check_runtime_filter_partition_prune_counter = 1"""
         qt_runtime_filter_partition_pruning_decimal2 """
             select count(*) from date_partition_table where partition_col in
                 (select partition_col from date_partition_table
@@ -101,6 +112,8 @@ suite("test_hive_runtime_filter_partition_pruning", "p0,external,hive,external_d
             test_runtime_filter_partition_pruning()
 
         } finally {
+            sql """set runtime_filter_wait_time_ms = 1000"""
+            sql """set check_runtime_filter_partition_prune_counter = -1"""
         }
     }
 }