This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new e89e3acde08 [fix](test) stabilize external file cache cases (#65261)
e89e3acde08 is described below
commit e89e3acde082cf7e1ad3478358070fcf5d9a60b7
Author: shuke <[email protected]>
AuthorDate: Tue Jul 7 18:34:12 2026 +0800
[fix](test) stabilize external file cache cases (#65261)
- Stabilize external file cache regression cases under
`external_table_p0/cache`.
- Move file-cache metric/config cases that mutate or assert shared state
to `nonConcurrent`.
- Aggregate `information_schema.file_cache_statistics` metrics across
cache paths instead of sampling arbitrary rows.
- Replace fixed metric-refresh sleeps with polling, remove global
`enable_file_cache` writes, and strengthen the query-limit assertion to
compare against the computed query limit bytes.
## Problem
`information_schema.file_cache_statistics` reports one row per
`(CACHE_PATH, METRIC_NAME)`, while
the affected cases assert shared file-cache state. Running them in the
normal concurrent group means
another case can populate or clear the same file cache while these cases
are checking counters.
The previous implementation also had several case-local sources of
flakiness:
- `test_file_cache_query_limit.groovy` sampled metric rows with `limit
1` and checked the limited query against the baseline query footprint
instead of the computed query limit.
- `test_file_cache_statistics.groovy` used fixed sleeps before reading
async file-cache statistics.
- `test_file_cache_features.groovy` sampled boolean metrics with `limit
1`; its closure could return `false` without failing the test.
- `test_file_cache_statistics.groovy` and
`test_file_cache_features.groovy` changed global `enable_file_cache`.
---
.../cache/test_file_cache_features.groovy | 185 +++++-----------
.../cache/test_file_cache_query_limit.groovy | 193 +++++++----------
.../test_file_cache_query_limit_config.groovy | 8 +-
.../cache/test_file_cache_statistics.groovy | 238 ++++++---------------
4 files changed, 197 insertions(+), 427 deletions(-)
diff --git
a/regression-test/suites/external_table_p0/cache/test_file_cache_features.groovy
b/regression-test/suites/external_table_p0/cache/test_file_cache_features.groovy
index b6a8e2dc11e..45a0e6b4a4d 100644
---
a/regression-test/suites/external_table_p0/cache/test_file_cache_features.groovy
+++
b/regression-test/suites/external_table_p0/cache/test_file_cache_features.groovy
@@ -15,9 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-import java.util.concurrent.TimeUnit;
-import org.awaitility.Awaitility;
-
// Constants for file cache configuration
final String BACKEND_CONFIG_CHECK_FAILED_PREFIX = "Backend configuration check
failed: "
final String FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX = "File cache features
check failed: "
@@ -26,43 +23,41 @@ final String ENABLE_FILE_CACHE_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PR
final String FILE_CACHE_PATH_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "file_cache_path is empty or not
configured"
final String INITIAL_DISK_RESOURCE_LIMIT_MODE_CHECK_FAILED_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "initial disk_resource_limit_mode
does not exist"
final String INITIAL_NEED_EVICT_CACHE_IN_ADVANCE_CHECK_FAILED_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "initial need_evict_cache_in_advance
does not exist"
-final String INITIAL_VALUES_NOT_ZERO_CHECK_FAILED_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "initial values are not both 0 - "
-final String DISK_RESOURCE_LIMIT_MODE_TEST_FAILED_MSG = "Disk resource limit
mode test failed"
-final String NEED_EVICT_CACHE_IN_ADVANCE_TEST_FAILED_MSG = "Need evict cache
in advance test failed"
-suite("test_file_cache_features", "p0,external") {
+suite("test_file_cache_features", "p0,external,nonConcurrent") {
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
logger.info("diable Hive test.")
return;
}
+ sql """set enable_file_cache=true"""
+ sql """set disable_file_cache=false"""
+
// Check backend configuration prerequisites
// Note: This test case assumes a single backend scenario. Testing with
single backend is logically equivalent
// to testing with multiple backends having identical configurations, but
simpler in logic.
def enableFileCacheResult = sql """show backend config like
'enable_file_cache';"""
logger.info("enable_file_cache configuration: " + enableFileCacheResult)
-
- if (enableFileCacheResult.size() == 0 ||
!enableFileCacheResult[0][3].equalsIgnoreCase("true")) {
- logger.info(ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
- assertTrue(false, ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
- }
+ assertFalse(enableFileCacheResult.size() == 0 ||
!enableFileCacheResult[0][3].equalsIgnoreCase("true"),
+ ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
def fileCachePathResult = sql """show backend config like
'file_cache_path';"""
logger.info("file_cache_path configuration: " + fileCachePathResult)
-
- if (fileCachePathResult.size() == 0 || fileCachePathResult[0][3] == null
|| fileCachePathResult[0][3].trim().isEmpty()) {
- logger.info(FILE_CACHE_PATH_CHECK_FAILED_MSG)
- assertTrue(false, FILE_CACHE_PATH_CHECK_FAILED_MSG)
- }
+ assertFalse(fileCachePathResult.size() == 0 || fileCachePathResult[0][3]
== null || fileCachePathResult[0][3].trim().isEmpty(),
+ FILE_CACHE_PATH_CHECK_FAILED_MSG)
String catalog_name = "test_file_cache_features"
String ex_db_name = "`tpch1_parquet`"
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
String hms_port = context.config.otherConfigs.get(hivePrefix + "HmsPort")
- String hdfs_port = context.config.otherConfigs.get(hivePrefix + "HdfsPort")
- sql """set global enable_file_cache=true"""
+ def cacheMetricMax = { String metricName ->
+ def r = sql """select max(cast(METRIC_VALUE as double)) from
information_schema.file_cache_statistics
+ where METRIC_NAME = '${metricName}';"""
+ return (r.size() == 0 || r[0][0] == null) ? null :
Double.valueOf(r[0][0].toString())
+ }
+
sql """drop catalog if exists ${catalog_name} """
sql """CREATE CATALOG ${catalog_name} PROPERTIES (
@@ -86,54 +81,29 @@ suite("test_file_cache_features", "p0,external") {
group by l_returnflag, l_linestatus
order by l_returnflag, l_linestatus;"""
- // Check file cache features
- // ===== File Cache Features Metrics Check =====
- // Get initial values for disk resource limit mode and cache eviction
advance
- def initialDiskResourceLimitModeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'disk_resource_limit_mode' limit 1;"""
- logger.info("Initial disk_resource_limit_mode result: " +
initialDiskResourceLimitModeResult)
-
- def initialNeedEvictCacheInAdvanceResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'need_evict_cache_in_advance' limit 1;"""
- logger.info("Initial need_evict_cache_in_advance result: " +
initialNeedEvictCacheInAdvanceResult)
-
- // Check if initial values exist
- if (initialDiskResourceLimitModeResult.size() == 0) {
- logger.info(INITIAL_DISK_RESOURCE_LIMIT_MODE_CHECK_FAILED_MSG)
- assertTrue(false, INITIAL_DISK_RESOURCE_LIMIT_MODE_CHECK_FAILED_MSG)
- }
- if (initialNeedEvictCacheInAdvanceResult.size() == 0) {
- logger.info(INITIAL_NEED_EVICT_CACHE_IN_ADVANCE_CHECK_FAILED_MSG)
- assertTrue(false, INITIAL_NEED_EVICT_CACHE_IN_ADVANCE_CHECK_FAILED_MSG)
- }
-
- // Store initial values
- double initialDiskResourceLimitMode =
Double.valueOf(initialDiskResourceLimitModeResult[0][0])
- double initialNeedEvictCacheInAdvance =
Double.valueOf(initialNeedEvictCacheInAdvanceResult[0][0])
-
- logger.info("Initial file cache features values -
disk_resource_limit_mode: ${initialDiskResourceLimitMode}, " +
- "need_evict_cache_in_advance: ${initialNeedEvictCacheInAdvance}")
-
- // Check if initial values are both 0
- if (initialDiskResourceLimitMode != 0.0 || initialNeedEvictCacheInAdvance
!= 0.0) {
- logger.info(INITIAL_VALUES_NOT_ZERO_CHECK_FAILED_MSG +
- "disk_resource_limit_mode: ${initialDiskResourceLimitMode},
need_evict_cache_in_advance: ${initialNeedEvictCacheInAdvance}")
- assertTrue(false, INITIAL_VALUES_NOT_ZERO_CHECK_FAILED_MSG +
- "disk_resource_limit_mode: ${initialDiskResourceLimitMode},
need_evict_cache_in_advance: ${initialNeedEvictCacheInAdvance}")
- }
-
def fileCacheBackgroundMonitorIntervalMsResult = sql """show backend
config like 'file_cache_background_monitor_interval_ms';"""
logger.info("file_cache_background_monitor_interval_ms configuration: " +
fileCacheBackgroundMonitorIntervalMsResult)
assertFalse(fileCacheBackgroundMonitorIntervalMsResult.size() == 0 ||
fileCacheBackgroundMonitorIntervalMsResult[0][3] == null ||
fileCacheBackgroundMonitorIntervalMsResult[0][3].trim().isEmpty(),
"file_cache_background_monitor_interval_ms is empty or not set to true")
- // brpc metrics will be updated at most 5 seconds
def totalWaitTime =
(fileCacheBackgroundMonitorIntervalMsResult[0][3].toInteger() / 1000) as int
- def interval = 1
- def iterations = totalWaitTime / interval
+ int pollTimeoutSeconds = Math.max(30, totalWaitTime * 6)
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('disk_resource_limit_mode') != null &&
+ cacheMetricMax('need_evict_cache_in_advance') != null &&
+ cacheMetricMax('disk_resource_limit_mode') == 0.0 &&
+ cacheMetricMax('need_evict_cache_in_advance') == 0.0
+ }
+
+ double initialDiskResourceLimitMode =
cacheMetricMax('disk_resource_limit_mode')
+ double initialNeedEvictCacheInAdvance =
cacheMetricMax('need_evict_cache_in_advance')
+ logger.info("Initial file cache features values -
disk_resource_limit_mode: ${initialDiskResourceLimitMode}, " +
+ "need_evict_cache_in_advance: ${initialNeedEvictCacheInAdvance}")
+ assertTrue(initialDiskResourceLimitMode == 0.0,
INITIAL_DISK_RESOURCE_LIMIT_MODE_CHECK_FAILED_MSG)
+ assertTrue(initialNeedEvictCacheInAdvance == 0.0,
INITIAL_NEED_EVICT_CACHE_IN_ADVANCE_CHECK_FAILED_MSG)
// Set backend configuration parameters for testing
- boolean diskResourceLimitModeTestPassed = true
setBeConfigTemporary([
"file_cache_enter_disk_resource_limit_mode_percent": "2",
"file_cache_exit_disk_resource_limit_mode_percent": "1"
@@ -141,49 +111,20 @@ suite("test_file_cache_features", "p0,external") {
// Execute test logic with modified configuration
logger.info("Backend configuration set -
file_cache_enter_disk_resource_limit_mode_percent: 2, " +
"file_cache_exit_disk_resource_limit_mode_percent: 1")
-
- // Wait for disk_resource_limit_mode metric to change to 1
- try {
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for backend configuration update
${elapsedSeconds} seconds, ${remainingSeconds} seconds remaining")
- }
-
- def updatedDiskResourceLimitModeResult = sql """select
METRIC_VALUE from information_schema.file_cache_statistics
- where METRIC_NAME = 'disk_resource_limit_mode' limit 1;"""
- logger.info("Checking disk_resource_limit_mode result: " +
updatedDiskResourceLimitModeResult)
-
- if (updatedDiskResourceLimitModeResult.size() > 0) {
- double updatedDiskResourceLimitMode =
Double.valueOf(updatedDiskResourceLimitModeResult[0][0])
- logger.info("Current disk_resource_limit_mode value:
${updatedDiskResourceLimitMode}")
-
- if (updatedDiskResourceLimitMode == 1.0) {
- logger.info("Disk resource limit mode is now active (value
= 1)")
- return true
- } else {
- logger.info("Disk resource limit mode is not yet active
(value = ${updatedDiskResourceLimitMode}), waiting...")
- return false
- }
- } else {
- logger.info("Failed to get disk_resource_limit_mode metric,
waiting...")
- return false
- }
- } catch (Exception e) {
- logger.info(DISK_RESOURCE_LIMIT_MODE_TEST_FAILED_MSG +
e.getMessage())
- diskResourceLimitModeTestPassed = false
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('disk_resource_limit_mode') != null &&
+ cacheMetricMax('disk_resource_limit_mode') == 1.0
}
+ logger.info("Disk resource limit mode is now active (value =
${cacheMetricMax('disk_resource_limit_mode')})")
}
-
- // Check disk resource limit mode test result
- if (!diskResourceLimitModeTestPassed) {
- logger.info(DISK_RESOURCE_LIMIT_MODE_TEST_FAILED_MSG)
- assertTrue(false, DISK_RESOURCE_LIMIT_MODE_TEST_FAILED_MSG)
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('disk_resource_limit_mode') != null &&
+ cacheMetricMax('disk_resource_limit_mode') == 0.0
}
// Set backend configuration parameters for need_evict_cache_in_advance
testing
- boolean needEvictCacheInAdvanceTestPassed = true
setBeConfigTemporary([
"enable_evict_file_cache_in_advance": "true",
"file_cache_enter_need_evict_cache_in_advance_percent": "2",
@@ -194,48 +135,18 @@ suite("test_file_cache_features", "p0,external") {
"enable_evict_file_cache_in_advance: true, " +
"file_cache_enter_need_evict_cache_in_advance_percent: 2, " +
"file_cache_exit_need_evict_cache_in_advance_percent: 1")
-
- // Wait for need_evict_cache_in_advance metric to change to 1
- try {
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for backend configuration update
${elapsedSeconds} seconds, ${remainingSeconds} seconds remaining")
- }
-
- def updatedNeedEvictCacheInAdvanceResult = sql """select
METRIC_VALUE from information_schema.file_cache_statistics
- where METRIC_NAME = 'need_evict_cache_in_advance' limit 1;"""
- logger.info("Checking need_evict_cache_in_advance result: " +
updatedNeedEvictCacheInAdvanceResult)
-
- if (updatedNeedEvictCacheInAdvanceResult.size() > 0) {
- double updatedNeedEvictCacheInAdvance =
Double.valueOf(updatedNeedEvictCacheInAdvanceResult[0][0])
- logger.info("Current need_evict_cache_in_advance value:
${updatedNeedEvictCacheInAdvance}")
-
- if (updatedNeedEvictCacheInAdvance == 1.0) {
- logger.info("Need evict cache in advance mode is now
active (value = 1)")
- return true
- } else {
- logger.info("Need evict cache in advance mode is not yet
active (value = ${updatedNeedEvictCacheInAdvance}), waiting...")
- return false
- }
- } else {
- logger.info("Failed to get need_evict_cache_in_advance metric,
waiting...")
- return false
- }
- } catch (Exception e) {
- logger.info(NEED_EVICT_CACHE_IN_ADVANCE_TEST_FAILED_MSG +
e.getMessage())
- needEvictCacheInAdvanceTestPassed = false
- }
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('need_evict_cache_in_advance') != null &&
+ cacheMetricMax('need_evict_cache_in_advance') == 1.0
+ }
+ logger.info("Need evict cache in advance mode is now active (value =
${cacheMetricMax('need_evict_cache_in_advance')})")
}
-
- // Check need evict cache in advance test result
- if (!needEvictCacheInAdvanceTestPassed) {
- logger.info(NEED_EVICT_CACHE_IN_ADVANCE_TEST_FAILED_MSG)
- assertTrue(false, NEED_EVICT_CACHE_IN_ADVANCE_TEST_FAILED_MSG)
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('need_evict_cache_in_advance') != null &&
+ cacheMetricMax('need_evict_cache_in_advance') == 0.0
}
- // ===== End File Cache Features Metrics Check =====
- sql """set global enable_file_cache=false"""
return true
}
diff --git
a/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit.groovy
b/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit.groovy
index a69d2eefb25..28896bed3b9 100644
---
a/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit.groovy
+++
b/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit.groovy
@@ -15,8 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-import org.codehaus.groovy.runtime.dgmimpl.arrays.LongArrayGetAtMetaMethod
-
import java.util.concurrent.TimeUnit;
import org.awaitility.Awaitility;
@@ -29,7 +27,7 @@ final String WEB_SERVER_PORT_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREF
final String BRPC_PORT_CHECK_FAILED_MSG = BACKEND_CONFIG_CHECK_FAILED_PREFIX +
"brpc_port is empty or not configured"
final String ENABLE_FILE_CACHE_QUERY_LIMIT_CHECK_FALSE_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "enable_file_cache_query_limit is empty or
not set to false"
final String ENABLE_FILE_CACHE_QUERY_LIMIT_CHECK_TRUE_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "enable_file_cache_query_limit is empty or
not set to true"
-final String FILE_CACHE_QUERY_LIMIT_BYTES_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "file_cache_query_limit_bytes is empty or
not configured"
+final String FILE_CACHE_QUERY_LIMIT_BYTES_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "query limit bytes should be positive and
smaller than baseline query cache capacity"
// Constants for cache query features check
final String FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX = "File cache features
check failed: "
final String BASE_NORMAL_QUEUE_CURR_SIZE_IS_ZERO_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "base normal_queue_curr_size is 0"
@@ -41,9 +39,9 @@ final String INITIAL_NORMAL_QUEUE_MAX_SIZE_IS_ZERO_MSG =
FILE_CACHE_FEATURES_CHE
final String INITIAL_NORMAL_QUEUE_MAX_ELEMENTS_IS_ZERO_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "initial normal_queue_max_elements is
0"
final String NORMAL_QUEUE_CURR_SIZE_NOT_GREATER_THAN_ZERO_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "normal_queue_curr_size is not
greater than 0 after cache operation"
final String NORMAL_QUEUE_CURR_ELEMENTS_NOT_GREATER_THAN_ZERO_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "normal_queue_curr_elements is not
greater than 0 after cache operation"
-final String NORMAL_QUEUE_CURR_SIZE_GREATER_THAN_QUERY_CACHE_CAPACITY_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "normal_queue_curr_size is greater
than query cache capacity"
+final String NORMAL_QUEUE_CURR_SIZE_GREATER_THAN_QUERY_CACHE_LIMIT_MSG =
FILE_CACHE_FEATURES_CHECK_FAILED_PREFIX + "normal_queue_curr_size is greater
than query cache limit"
-suite("test_file_cache_query_limit", "p0,external") {
+suite("test_file_cache_query_limit", "p0,external,nonConcurrent") {
String enableHiveTest = context.config.otherConfigs.get("enableHiveTest")
if (enableHiveTest == null || !enableHiveTest.equalsIgnoreCase("true")) {
logger.info("disable hive test.")
@@ -51,6 +49,7 @@ suite("test_file_cache_query_limit", "p0,external") {
}
sql """set enable_file_cache=true"""
+ sql """set disable_file_cache=false"""
// Check backend configuration prerequisites
// Note: This test case assumes a single backend scenario. Testing with
single backend is logically equivalent
@@ -69,7 +68,17 @@ suite("test_file_cache_query_limit", "p0,external") {
String ex_db_name = "tpch1_parquet"
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
String hms_port = context.config.otherConfigs.get(hivePrefix + "HmsPort")
- int queryCacheCapacity
+ long queryCacheCapacity
+ long queryLimitBytes
+
+ // Sum a file_cache_statistics metric across all cache paths.
file_cache_statistics reports
+ // one row per (cache_path, metric_name), so "limit 1" can observe an
arbitrary path instead
+ // of the whole file cache state.
+ def cacheMetricSum = { String metricName ->
+ def r = sql """select sum(cast(METRIC_VALUE as double)) from
information_schema.file_cache_statistics
+ where METRIC_NAME = '${metricName}';"""
+ return (r.size() == 0 || r[0][0] == null) ? null :
Double.valueOf(r[0][0].toString())
+ }
// Poll a file_cache_statistics metric until predicate holds, or until
timeout.
// file_cache_statistics is refreshed by the background monitor on its own
cadence,
@@ -82,9 +91,8 @@ suite("test_file_cache_query_limit", "p0,external") {
.atMost(timeoutSeconds, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.until {
- def r = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = '${metricName}' limit 1;"""
- return r.size() > 0 &&
predicate(Double.valueOf(r[0][0]))
+ Double metricValue = cacheMetricSum(metricName)
+ return metricValue != null && predicate(metricValue)
}
} catch (org.awaitility.core.ConditionTimeoutException ignored) {
// fall through; the caller's assert will surface the precise
failure
@@ -139,11 +147,15 @@ suite("test_file_cache_query_limit", "p0,external") {
// Wait for process completion and check exit status
def exitCode = process.waitFor()
- def fileCacheCapacityResult = output.toString().split("\n").find {
it.contains("file_cache_capacity") }?.split(":")?.last()?.trim()
+ def fileCacheCapacityResults = output.toString().split("\n")
+ .findAll { it.contains("file_cache_capacity") }
+ .collect { it.split(":")?.last()?.trim() }
+ .findAll { it != null && !it.isEmpty() }
+ .collect { Long.valueOf(it) }
- logger.info("File cache capacity: ${fileCacheCapacityResult}")
- assertTrue(fileCacheCapacityResult != null, "Failed to find
file_cache_capacity in brpc metrics")
- def fileCacheCapacity = Long.valueOf(fileCacheCapacityResult)
+ logger.info("File cache capacities: ${fileCacheCapacityResults}")
+ assertTrue(!fileCacheCapacityResults.isEmpty(), "Failed to find
file_cache_capacity in brpc metrics")
+ long fileCacheCapacity = fileCacheCapacityResults.sum() as Long
// Run file cache base test for setting the parameter
file_cache_query_limit_bytes
logger.info("========================= Start running file cache base test
========================")
@@ -165,31 +177,34 @@ suite("test_file_cache_query_limit", "p0,external") {
// brpc metrics will be updated at most 5 seconds
def totalWaitTime =
(fileCacheBackgroundMonitorIntervalMsResult[0][3].toLong() / 1000) as int
- def interval = 1
- def iterations = totalWaitTime / interval
long pollTimeoutSeconds = Math.max(30L, (long) totalWaitTime * 6L)
+ def waitBackendConfig = { String configName, String expectedValue ->
+ Awaitility.await()
+ .atMost(pollTimeoutSeconds, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until {
+ def r = sql """SHOW BACKEND CONFIG LIKE '${configName}';"""
+ return r.size() > 0 && r[0][3] != null &&
r[0][3].toString().equalsIgnoreCase(expectedValue)
+ }
+ }
+
// Poll until the cache clear has drained the LRU queue. The HTTP clear
endpoint with sync=true
// deletes blocks synchronously, but the queue counters are republished by
the background monitor
// thread on its own cadence — so a single fixed-time wait can race the
refresh.
pollFileCacheMetric('normal_queue_curr_size', { it == 0.0 },
pollTimeoutSeconds)
pollFileCacheMetric('normal_queue_curr_elements', { it == 0.0 },
pollTimeoutSeconds)
- def initialNormalQueueCurrSizeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_size' limit 1;"""
- logger.info("normal_queue_curr_size result: " +
initialNormalQueueCurrSizeResult)
- assertFalse(initialNormalQueueCurrSizeResult.size() == 0 ||
Double.valueOf(initialNormalQueueCurrSizeResult[0][0]) != 0.0,
+ Double initialNormalQueueCurrSize =
cacheMetricSum('normal_queue_curr_size')
+ logger.info("normal_queue_curr_size sum result: " +
initialNormalQueueCurrSize)
+ assertFalse(initialNormalQueueCurrSize == null ||
initialNormalQueueCurrSize != 0.0,
INITIAL_NORMAL_QUEUE_CURR_SIZE_NOT_ZERO_MSG)
- def initialNormalQueueCurrElementsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_elements' limit 1;"""
- logger.info("normal_queue_curr_elements result: " +
initialNormalQueueCurrElementsResult)
- assertFalse(initialNormalQueueCurrElementsResult.size() == 0 ||
Double.valueOf(initialNormalQueueCurrElementsResult[0][0]) != 0.0,
+ Double initialNormalQueueCurrElements =
cacheMetricSum('normal_queue_curr_elements')
+ logger.info("normal_queue_curr_elements sum result: " +
initialNormalQueueCurrElements)
+ assertFalse(initialNormalQueueCurrElements == null ||
initialNormalQueueCurrElements != 0.0,
INITIAL_NORMAL_QUEUE_CURR_ELEMENTS_NOT_ZERO_MSG)
- double initialNormalQueueCurrSize =
Double.valueOf(initialNormalQueueCurrSizeResult[0][0])
- double initialNormalQueueCurrElements =
Double.valueOf(initialNormalQueueCurrElementsResult[0][0])
-
logger.info("Initial normal queue curr size and elements - size:
${initialNormalQueueCurrSize} , " +
"elements: ${initialNormalQueueCurrElements}")
@@ -199,13 +214,7 @@ suite("test_file_cache_query_limit", "p0,external") {
// Execute test logic with modified configuration for
file_cache_query_limit
logger.info("Backend configuration set -
enable_file_cache_query_limit: false")
- // Waiting for backend configuration update
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for backend configuration update
${elapsedSeconds} seconds, ${remainingSeconds} seconds remaining")
- }
+ waitBackendConfig("enable_file_cache_query_limit", "false")
// Check if the configuration is modified
def enableFileCacheQueryLimitResult = sql """SHOW BACKEND CONFIG LIKE
'enable_file_cache_query_limit';"""
@@ -221,20 +230,17 @@ suite("test_file_cache_query_limit", "p0,external") {
pollFileCacheMetric('normal_queue_curr_elements', { it > 0.0 },
pollTimeoutSeconds)
pollFileCacheMetric('normal_queue_curr_size', { it > 0.0 },
pollTimeoutSeconds)
- def baseNormalQueueCurrElementsResult = sql """select METRIC_VALUE
from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_elements' limit 1;"""
- logger.info("normal_queue_curr_elements result: " +
baseNormalQueueCurrElementsResult)
- assertFalse(baseNormalQueueCurrElementsResult.size() == 0 ||
Double.valueOf(baseNormalQueueCurrElementsResult[0][0]) == 0.0,
+ Double baseNormalQueueCurrElements =
cacheMetricSum('normal_queue_curr_elements')
+ logger.info("normal_queue_curr_elements sum result: " +
baseNormalQueueCurrElements)
+ assertFalse(baseNormalQueueCurrElements == null ||
baseNormalQueueCurrElements == 0.0,
BASE_NORMAL_QUEUE_CURR_ELEMENTS_IS_ZERO_MSG)
- def baseNormalQueueCurrSizeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_size' limit 1;"""
- logger.info("normal_queue_curr_size result: " +
baseNormalQueueCurrSizeResult)
- assertFalse(baseNormalQueueCurrSizeResult.size() == 0 ||
Double.valueOf(baseNormalQueueCurrSizeResult[0][0]) == 0.0,
+ Double baseNormalQueueCurrSize =
cacheMetricSum('normal_queue_curr_size')
+ logger.info("normal_queue_curr_size sum result: " +
baseNormalQueueCurrSize)
+ assertFalse(baseNormalQueueCurrSize == null || baseNormalQueueCurrSize
== 0.0,
BASE_NORMAL_QUEUE_CURR_SIZE_IS_ZERO_MSG)
- int baseNormalQueueCurrElements =
Double.valueOf(baseNormalQueueCurrElementsResult[0][0]) as Long
- queryCacheCapacity =
Double.valueOf(baseNormalQueueCurrSizeResult[0][0]) as Long
+ queryCacheCapacity = baseNormalQueueCurrSize as Long
}
// The parameter file_cache_query_limit_percent must be set smaller than
the cache capacity required by the query
@@ -245,8 +251,12 @@ suite("test_file_cache_query_limit", "p0,external") {
logger.info("==================== Start running file cache query limit
test 1 ====================")
- def fileCacheQueryLimitPercentTest1 = (fileCacheQueryLimitPercent / 2) as
Long
+ long fileCacheQueryLimitPercentTest1 = Math.max(1L,
(fileCacheQueryLimitPercent / 2) as Long)
+ queryLimitBytes = (fileCacheCapacity * fileCacheQueryLimitPercentTest1 /
100) as Long
logger.info("file_cache_query_limit_percent_test1: " +
fileCacheQueryLimitPercentTest1)
+ logger.info("query_limit_bytes: ${queryLimitBytes}, query_cache_capacity:
${queryCacheCapacity}")
+ assertTrue(queryLimitBytes > 0 && queryLimitBytes < queryCacheCapacity,
+ FILE_CACHE_QUERY_LIMIT_BYTES_CHECK_FAILED_MSG)
// Clear file cache
process = new ProcessBuilder(stringCommand as
String[]).redirectErrorStream(true).start()
@@ -267,38 +277,29 @@ suite("test_file_cache_query_limit", "p0,external") {
// ===== Normal Queue Metrics Check =====
// Check normal queue current size
- initialNormalQueueCurrSizeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_size' limit 1;"""
- logger.info("normal_queue_curr_size result: " +
initialNormalQueueCurrSizeResult)
- assertFalse(initialNormalQueueCurrSizeResult.size() == 0 ||
Double.valueOf(initialNormalQueueCurrSizeResult[0][0]) != 0.0,
+ initialNormalQueueCurrSize = cacheMetricSum('normal_queue_curr_size')
+ logger.info("normal_queue_curr_size sum result: " +
initialNormalQueueCurrSize)
+ assertFalse(initialNormalQueueCurrSize == null ||
initialNormalQueueCurrSize != 0.0,
INITIAL_NORMAL_QUEUE_CURR_SIZE_NOT_ZERO_MSG)
// Check normal queue current elements
- initialNormalQueueCurrElementsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_elements' limit 1;"""
- logger.info("normal_queue_curr_elements result: " +
initialNormalQueueCurrElementsResult)
- assertFalse(initialNormalQueueCurrElementsResult.size() == 0 ||
Double.valueOf(initialNormalQueueCurrElementsResult[0][0]) != 0.0,
+ initialNormalQueueCurrElements =
cacheMetricSum('normal_queue_curr_elements')
+ logger.info("normal_queue_curr_elements sum result: " +
initialNormalQueueCurrElements)
+ assertFalse(initialNormalQueueCurrElements == null ||
initialNormalQueueCurrElements != 0.0,
INITIAL_NORMAL_QUEUE_CURR_ELEMENTS_NOT_ZERO_MSG)
// Check normal queue max size
- def initialNormalQueueMaxSizeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_max_size' limit 1;"""
- logger.info("normal_queue_max_size result: " +
initialNormalQueueMaxSizeResult)
- assertFalse(initialNormalQueueMaxSizeResult.size() == 0 ||
Double.valueOf(initialNormalQueueMaxSizeResult[0][0]) == 0.0,
+ Double initialNormalQueueMaxSize = cacheMetricSum('normal_queue_max_size')
+ logger.info("normal_queue_max_size sum result: " +
initialNormalQueueMaxSize)
+ assertFalse(initialNormalQueueMaxSize == null || initialNormalQueueMaxSize
== 0.0,
INITIAL_NORMAL_QUEUE_MAX_SIZE_IS_ZERO_MSG)
// Check normal queue max elements
- def initialNormalQueueMaxElementsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_max_elements' limit 1;"""
- logger.info("normal_queue_max_elements result: " +
initialNormalQueueMaxElementsResult)
- assertFalse(initialNormalQueueMaxElementsResult.size() == 0 ||
Double.valueOf(initialNormalQueueMaxElementsResult[0][0]) == 0.0,
+ Double initialNormalQueueMaxElements =
cacheMetricSum('normal_queue_max_elements')
+ logger.info("normal_queue_max_elements sum result: " +
initialNormalQueueMaxElements)
+ assertFalse(initialNormalQueueMaxElements == null ||
initialNormalQueueMaxElements == 0.0,
INITIAL_NORMAL_QUEUE_MAX_ELEMENTS_IS_ZERO_MSG)
- initialNormalQueueCurrSize =
Double.valueOf(initialNormalQueueCurrSizeResult[0][0])
- initialNormalQueueCurrElements =
Double.valueOf(initialNormalQueueCurrElementsResult[0][0])
- double initialNormalQueueMaxSize =
Double.valueOf(initialNormalQueueMaxSizeResult[0][0])
- double initialNormalQueueMaxElements =
Double.valueOf(initialNormalQueueMaxElementsResult[0][0])
-
logger.info("Initial normal queue curr size and elements - size:
${initialNormalQueueCurrSize} , " +
"elements: ${initialNormalQueueCurrElements}")
@@ -306,18 +307,10 @@ suite("test_file_cache_query_limit", "p0,external") {
"elements: ${initialNormalQueueMaxElements}")
// ===== Hit And Read Counts Metrics Check =====
- // Get initial values for hit and read counts
- def initialTotalHitCountsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'total_hit_counts' limit 1;"""
- logger.info("Initial total_hit_counts result: " +
initialTotalHitCountsResult)
-
- def initialTotalReadCountsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'total_read_counts' limit 1;"""
- logger.info("Initial total_read_counts result: " +
initialTotalReadCountsResult)
-
- // Store initial values
- double initialTotalHitCounts =
Double.valueOf(initialTotalHitCountsResult[0][0])
- double initialTotalReadCounts =
Double.valueOf(initialTotalReadCountsResult[0][0])
+ // total_hit_counts / total_read_counts are also reported per cache path.
+ Double initialTotalHitCounts = cacheMetricSum('total_hit_counts')
+ Double initialTotalReadCounts = cacheMetricSum('total_read_counts')
+ logger.info("Initial total_hit_counts (sum): ${initialTotalHitCounts},
total_read_counts (sum): ${initialTotalReadCounts}")
// Set backend configuration parameters for file_cache_query_limit test 1
setBeConfigTemporary([
@@ -328,13 +321,7 @@ suite("test_file_cache_query_limit", "p0,external") {
sql """set file_cache_query_limit_percent =
${fileCacheQueryLimitPercentTest1}"""
- // Waiting for backend configuration update
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for backend configuration update
${elapsedSeconds} seconds, ${remainingSeconds} seconds remaining")
- }
+ waitBackendConfig("enable_file_cache_query_limit", "true")
// Check if the configuration is modified
def enableFileCacheQueryLimitResult = sql """SHOW BACKEND CONFIG LIKE
'enable_file_cache_query_limit';"""
@@ -352,42 +339,24 @@ suite("test_file_cache_query_limit", "p0,external") {
pollFileCacheMetric('normal_queue_curr_elements', { it > 0.0 },
pollTimeoutSeconds)
// Get updated value of normal queue current elements and max elements
after cache operations
- def updatedNormalQueueCurrSizeResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_size' limit 1;"""
- logger.info("normal_queue_curr_size result: " +
updatedNormalQueueCurrSizeResult)
-
- def updatedNormalQueueCurrElementsResult = sql """select METRIC_VALUE
from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_elements' limit 1;"""
- logger.info("normal_queue_curr_elements result: " +
updatedNormalQueueCurrElementsResult)
-
// Check if updated values are greater than initial values
- double updatedNormalQueueCurrSize =
Double.valueOf(updatedNormalQueueCurrSizeResult[0][0])
- double updatedNormalQueueCurrElements =
Double.valueOf(updatedNormalQueueCurrElementsResult[0][0])
+ Double updatedNormalQueueCurrSize =
cacheMetricSum('normal_queue_curr_size')
+ Double updatedNormalQueueCurrElements =
cacheMetricSum('normal_queue_curr_elements')
- logger.info("Updated normal queue curr size and elements - size:
${updatedNormalQueueCurrSize} , " +
+ logger.info("Updated normal queue curr size and elements sum - size:
${updatedNormalQueueCurrSize} , " +
"elements: ${updatedNormalQueueCurrElements}")
assertTrue(updatedNormalQueueCurrSize > 0.0,
NORMAL_QUEUE_CURR_SIZE_NOT_GREATER_THAN_ZERO_MSG)
assertTrue(updatedNormalQueueCurrElements > 0.0,
NORMAL_QUEUE_CURR_ELEMENTS_NOT_GREATER_THAN_ZERO_MSG)
- logger.info("Normal queue curr size and query cache capacity
comparison - normal queue curr size: ${updatedNormalQueueCurrSize as Long} , " +
- "query cache capacity: ${fileCacheCapacity}")
+ logger.info("Normal queue curr size and query cache limit comparison -
normal queue curr size: ${updatedNormalQueueCurrSize as Long} , " +
+ "query cache limit: ${queryLimitBytes}, baseline query cache
capacity: ${queryCacheCapacity}")
- assertTrue((updatedNormalQueueCurrSize as Long) <= queryCacheCapacity,
- NORMAL_QUEUE_CURR_SIZE_GREATER_THAN_QUERY_CACHE_CAPACITY_MSG)
+ assertTrue((updatedNormalQueueCurrSize as Long) <= queryLimitBytes,
+ NORMAL_QUEUE_CURR_SIZE_GREATER_THAN_QUERY_CACHE_LIMIT_MSG)
- // Get updated values for hit and read counts after cache operations
- def updatedTotalHitCountsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'total_hit_counts' limit 1;"""
- logger.info("Updated total_hit_counts result: " +
updatedTotalHitCountsResult)
-
- def updatedTotalReadCountsResult = sql """select METRIC_VALUE from
information_schema.file_cache_statistics
- where METRIC_NAME = 'total_read_counts' limit 1;"""
- logger.info("Updated total_read_counts result: " +
updatedTotalReadCountsResult)
-
- // Check if updated values are greater than initial values
- double updatedTotalHitCounts =
Double.valueOf(updatedTotalHitCountsResult[0][0])
- double updatedTotalReadCounts =
Double.valueOf(updatedTotalReadCountsResult[0][0])
+ Double updatedTotalHitCounts = cacheMetricSum('total_hit_counts')
+ Double updatedTotalReadCounts = cacheMetricSum('total_read_counts')
logger.info("Total hit and read counts comparison - hit counts:
${initialTotalHitCounts} -> " +
"${updatedTotalHitCounts} , read counts:
${initialTotalReadCounts} -> ${updatedTotalReadCounts}")
@@ -398,4 +367,4 @@ suite("test_file_cache_query_limit", "p0,external") {
logger.info("===================== End running file cache query limit test
1 =====================")
return true;
-}
\ No newline at end of file
+}
diff --git
a/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit_config.groovy
b/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit_config.groovy
index 4c722a2c4b5..3bdd8a25d9a 100644
---
a/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit_config.groovy
+++
b/regression-test/suites/external_table_p0/cache/test_file_cache_query_limit_config.groovy
@@ -15,15 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-import org.codehaus.groovy.runtime.dgmimpl.arrays.LongArrayGetAtMetaMethod
-
-import java.util.concurrent.TimeUnit;
-import org.awaitility.Awaitility;
-
final String ERROR_SQL_SUCCEED_MSG = "SQL should have failed but succeeded"
final String SET_SESSION_VARIABLE_FAILED_MSG = "SQL set session variable
failed"
-suite("test_file_cache_query_limit_config", "p0,external") {
+suite("test_file_cache_query_limit_config", "p0,external,nonConcurrent") {
sql """set file_cache_query_limit_percent = 1"""
def fileCacheQueryLimitPercentResult = sql """show variables like
'file_cache_query_limit_percent';"""
@@ -121,4 +116,3 @@ suite("test_file_cache_query_limit_config", "p0,external") {
return true;
}
-
diff --git
a/regression-test/suites/external_table_p0/cache/test_file_cache_statistics.groovy
b/regression-test/suites/external_table_p0/cache/test_file_cache_statistics.groovy
index da6476cdc48..ff4354f8dec 100644
---
a/regression-test/suites/external_table_p0/cache/test_file_cache_statistics.groovy
+++
b/regression-test/suites/external_table_p0/cache/test_file_cache_statistics.groovy
@@ -15,9 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-import java.util.concurrent.TimeUnit;
-import org.awaitility.Awaitility;
-
// Constants for backend configuration check
final String BACKEND_CONFIG_CHECK_FAILED_PREFIX = "Backend configuration check
failed: "
final String ENABLE_FILE_CACHE_CHECK_FAILED_MSG =
BACKEND_CONFIG_CHECK_FAILED_PREFIX + "enable_file_cache is not set to true"
@@ -36,44 +33,45 @@ final String NORMAL_QUEUE_ELEMENTS_VALIDATION_FAILED_MSG =
NORMAL_QUEUE_CHECK_FA
// Constants for hit and read counts check
final String HIT_AND_READ_COUNTS_CHECK_FAILED_PREFIX = "Hit and read counts
check failed: "
-final String INITIAL_TOTAL_HIT_COUNTS_NOT_GREATER_THAN_0_MSG =
HIT_AND_READ_COUNTS_CHECK_FAILED_PREFIX + "initial total_hit_counts is not
greater than 0"
-final String INITIAL_TOTAL_READ_COUNTS_NOT_GREATER_THAN_0_MSG =
HIT_AND_READ_COUNTS_CHECK_FAILED_PREFIX + "initial total_read_counts is not
greater than 0"
final String TOTAL_HIT_COUNTS_DID_NOT_INCREASE_MSG =
HIT_AND_READ_COUNTS_CHECK_FAILED_PREFIX + "total_hit_counts did not increase
after cache operation"
final String TOTAL_READ_COUNTS_DID_NOT_INCREASE_MSG =
HIT_AND_READ_COUNTS_CHECK_FAILED_PREFIX + "total_read_counts did not increase
after cache operation"
-suite("test_file_cache_statistics", "p0,external") {
+suite("test_file_cache_statistics", "p0,external,nonConcurrent") {
String enabled = context.config.otherConfigs.get("enableHiveTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
logger.info("diable Hive test.")
return;
}
+ sql """set enable_file_cache=true"""
+ sql """set disable_file_cache=false"""
+
// Check backend configuration prerequisites
// Note: This test case assumes a single backend scenario. Testing with
single backend is logically equivalent
// to testing with multiple backends having identical configurations, but
simpler in logic.
def enableFileCacheResult = sql """show backend config like
'enable_file_cache';"""
logger.info("enable_file_cache configuration: " + enableFileCacheResult)
-
- if (enableFileCacheResult.size() == 0 ||
!enableFileCacheResult[0][3].equalsIgnoreCase("true")) {
- logger.info(ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
- assertTrue(false, ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
- }
+ assertFalse(enableFileCacheResult.size() == 0 ||
!enableFileCacheResult[0][3].equalsIgnoreCase("true"),
+ ENABLE_FILE_CACHE_CHECK_FAILED_MSG)
def fileCachePathResult = sql """show backend config like
'file_cache_path';"""
logger.info("file_cache_path configuration: " + fileCachePathResult)
-
- if (fileCachePathResult.size() == 0 || fileCachePathResult[0][3] == null
|| fileCachePathResult[0][3].trim().isEmpty()) {
- logger.info(FILE_CACHE_PATH_CHECK_FAILED_MSG)
- assertTrue(false, FILE_CACHE_PATH_CHECK_FAILED_MSG)
- }
+ assertFalse(fileCachePathResult.size() == 0 || fileCachePathResult[0][3]
== null || fileCachePathResult[0][3].trim().isEmpty(),
+ FILE_CACHE_PATH_CHECK_FAILED_MSG)
String catalog_name = "test_file_cache_statistics"
String ex_db_name = "`default`"
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
String hms_port = context.config.otherConfigs.get(hivePrefix + "HmsPort")
- String hdfs_port = context.config.otherConfigs.get(hivePrefix + "HdfsPort")
- sql """set global enable_file_cache=true"""
+ def cacheMetric = { String metricName, String aggregateFunc ->
+ def r = sql """select ${aggregateFunc}(cast(METRIC_VALUE as double))
from information_schema.file_cache_statistics
+ where METRIC_NAME = '${metricName}';"""
+ return (r.size() == 0 || r[0][0] == null) ? null :
Double.valueOf(r[0][0].toString())
+ }
+ def cacheMetricSum = { String metricName -> cacheMetric(metricName, "sum")
}
+ def cacheMetricMax = { String metricName -> cacheMetric(metricName, "max")
}
+
sql """drop catalog if exists ${catalog_name} """
sql """CREATE CATALOG ${catalog_name} PROPERTIES (
@@ -84,176 +82,74 @@ suite("test_file_cache_statistics", "p0,external") {
sql """switch ${catalog_name}"""
+ String querySql = """select * from
${catalog_name}.${ex_db_name}.parquet_partition_table
+ where l_orderkey=1 and l_partkey=1534 limit 1;"""
+
// load the table into file cache
- sql """select * from ${catalog_name}.${ex_db_name}.parquet_partition_table
where l_orderkey=1 and l_partkey=1534 limit 1;"""
+ sql querySql
// do it twice to make sure the table block could hit the cache
- order_qt_1 """select * from
${catalog_name}.${ex_db_name}.parquet_partition_table where l_orderkey=1 and
l_partkey=1534 limit 1;"""
+ order_qt_1 querySql
def fileCacheBackgroundMonitorIntervalMsResult = sql """show backend
config like 'file_cache_background_monitor_interval_ms';"""
logger.info("file_cache_background_monitor_interval_ms configuration: " +
fileCacheBackgroundMonitorIntervalMsResult)
assertFalse(fileCacheBackgroundMonitorIntervalMsResult.size() == 0 ||
fileCacheBackgroundMonitorIntervalMsResult[0][3] == null ||
fileCacheBackgroundMonitorIntervalMsResult[0][3].trim().isEmpty(),
"file_cache_background_monitor_interval_ms is empty or not set to true")
- // brpc metrics will be updated at most 5 seconds
def totalWaitTime =
(fileCacheBackgroundMonitorIntervalMsResult[0][3].toInteger() / 1000) as int
- def interval = 1
- def iterations = totalWaitTime / interval
-
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for file cache statistics update ${elapsedSeconds}
seconds, ${remainingSeconds} seconds remaining")
- }
-
- // ===== Hit Ratio Metrics Check =====
- // Check overall hit ratio hits_ratio
- def hitsRatioResult = sql """select MAX(CAST(METRIC_VALUE AS DOUBLE)) from
information_schema.file_cache_statistics where METRIC_NAME = 'hits_ratio';"""
- logger.info("hits_ratio result: " + hitsRatioResult)
-
- // Check 1-hour hit ratio hits_ratio_1h
- def hitsRatio1hResult = sql """select MAX(CAST(METRIC_VALUE AS DOUBLE))
from information_schema.file_cache_statistics where METRIC_NAME =
'hits_ratio_1h';"""
- logger.info("hits_ratio_1h result: " + hitsRatio1hResult)
-
- // Check 5-minute hit ratio hits_ratio_5m
- def hitsRatio5mResult = sql """select MAX(CAST(METRIC_VALUE AS DOUBLE))
from information_schema.file_cache_statistics where METRIC_NAME =
'hits_ratio_5m';"""
- logger.info("hits_ratio_5m result: " + hitsRatio5mResult)
-
- // Check if all three metrics exist and are greater than 0
- boolean hasHitsRatio = hitsRatioResult.size() > 0 &&
Double.valueOf(hitsRatioResult[0][0]) > 0
- boolean hasHitsRatio1h = hitsRatio1hResult.size() > 0 &&
Double.valueOf(hitsRatio1hResult[0][0]) > 0
- boolean hasHitsRatio5m = hitsRatio5mResult.size() > 0 &&
Double.valueOf(hitsRatio5mResult[0][0]) > 0
-
- logger.info("Hit ratio metrics check result - hits_ratio: ${hasHitsRatio},
hits_ratio_1h: ${hasHitsRatio1h}, hits_ratio_5m: ${hasHitsRatio5m}")
-
- // Return false if any metric is false, otherwise return true
- if (!hasHitsRatio) {
- logger.info(HIT_RATIO_METRIC_FALSE_MSG)
- assertTrue(false, HIT_RATIO_METRIC_FALSE_MSG)
- }
- if (!hasHitsRatio1h) {
- logger.info(HIT_RATIO_1H_METRIC_FALSE_MSG)
- assertTrue(false, HIT_RATIO_1H_METRIC_FALSE_MSG)
- }
- if (!hasHitsRatio5m) {
- logger.info(HIT_RATIO_5M_METRIC_FALSE_MSG)
- assertTrue(false, HIT_RATIO_5M_METRIC_FALSE_MSG)
- }
- // ===== End Hit Ratio Metrics Check =====
-
- // ===== Normal Queue Metrics Check =====
- // Check normal queue current size and max size
- def normalQueueCurrSizeResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_size';"""
- logger.info("normal_queue_curr_size result: " + normalQueueCurrSizeResult)
-
- def normalQueueMaxSizeResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_max_size';"""
- logger.info("normal_queue_max_size result: " + normalQueueMaxSizeResult)
-
- // Check normal queue current elements and max elements
- def normalQueueCurrElementsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_curr_elements';"""
- logger.info("normal_queue_curr_elements result: " +
normalQueueCurrElementsResult)
-
- def normalQueueMaxElementsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'normal_queue_max_elements';"""
- logger.info("normal_queue_max_elements result: " +
normalQueueMaxElementsResult)
-
- // Check normal queue size metrics
- boolean hasNormalQueueCurrSize = normalQueueCurrSizeResult.size() > 0 &&
- Double.valueOf(normalQueueCurrSizeResult[0][0]) > 0
- boolean hasNormalQueueMaxSize = normalQueueMaxSizeResult.size() > 0 &&
- Double.valueOf(normalQueueMaxSizeResult[0][0]) > 0
- boolean hasNormalQueueCurrElements = normalQueueCurrElementsResult.size()
> 0 &&
- Double.valueOf(normalQueueCurrElementsResult[0][0]) > 0
- boolean hasNormalQueueMaxElements = normalQueueMaxElementsResult.size() >
0 &&
- Double.valueOf(normalQueueMaxElementsResult[0][0]) > 0
-
- // Check if current size is less than max size and current elements is
less than max elements
- boolean normalQueueSizeValid = hasNormalQueueCurrSize &&
hasNormalQueueMaxSize &&
- Double.valueOf(normalQueueCurrSizeResult[0][0]) <
Double.valueOf(normalQueueMaxSizeResult[0][0])
- boolean normalQueueElementsValid = hasNormalQueueCurrElements &&
hasNormalQueueMaxElements &&
- Double.valueOf(normalQueueCurrElementsResult[0][0]) <
Double.valueOf(normalQueueMaxElementsResult[0][0])
-
- logger.info("Normal queue metrics check result - size valid:
${normalQueueSizeValid}, " +
- "elements valid: ${normalQueueElementsValid}")
-
- if (!normalQueueSizeValid) {
- logger.info(NORMAL_QUEUE_SIZE_VALIDATION_FAILED_MSG)
- assertTrue(false, NORMAL_QUEUE_SIZE_VALIDATION_FAILED_MSG)
- }
- if (!normalQueueElementsValid) {
- logger.info(NORMAL_QUEUE_ELEMENTS_VALIDATION_FAILED_MSG)
- assertTrue(false, NORMAL_QUEUE_ELEMENTS_VALIDATION_FAILED_MSG)
+ int pollTimeoutSeconds = Math.max(30, totalWaitTime * 6)
+
+ awaitUntil(pollTimeoutSeconds, 1) {
+ return cacheMetricMax('hits_ratio') != null &&
cacheMetricMax('hits_ratio') > 0.0 &&
+ cacheMetricMax('hits_ratio_1h') != null &&
cacheMetricMax('hits_ratio_1h') > 0.0 &&
+ cacheMetricMax('hits_ratio_5m') != null &&
cacheMetricMax('hits_ratio_5m') > 0.0 &&
+ cacheMetricSum('normal_queue_curr_size') != null &&
cacheMetricSum('normal_queue_curr_size') > 0.0 &&
+ cacheMetricSum('normal_queue_max_size') != null &&
cacheMetricSum('normal_queue_max_size') > 0.0 &&
+ cacheMetricSum('normal_queue_curr_elements') != null &&
cacheMetricSum('normal_queue_curr_elements') > 0.0 &&
+ cacheMetricSum('normal_queue_max_elements') != null &&
cacheMetricSum('normal_queue_max_elements') > 0.0 &&
+ cacheMetricSum('total_hit_counts') != null &&
cacheMetricSum('total_hit_counts') > 0.0 &&
+ cacheMetricSum('total_read_counts') != null &&
cacheMetricSum('total_read_counts') > 0.0
}
- // ===== End Normal Queue Metrics Check =====
-
- // ===== Hit and Read Counts Metrics Check =====
- // Get initial values for hit and read counts
- def initialHitCountsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'total_hit_counts';"""
- logger.info("Initial total_hit_counts result: " + initialHitCountsResult)
- def initialReadCountsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'total_read_counts';"""
- logger.info("Initial total_read_counts result: " + initialReadCountsResult)
-
- // Check if initial values exist and are greater than 0
- if (initialHitCountsResult.size() == 0 ||
Double.valueOf(initialHitCountsResult[0][0]) <= 0) {
- logger.info(INITIAL_TOTAL_HIT_COUNTS_NOT_GREATER_THAN_0_MSG)
- assertTrue(false, INITIAL_TOTAL_HIT_COUNTS_NOT_GREATER_THAN_0_MSG)
- }
- if (initialReadCountsResult.size() == 0 ||
Double.valueOf(initialReadCountsResult[0][0]) <= 0) {
- logger.info(INITIAL_TOTAL_READ_COUNTS_NOT_GREATER_THAN_0_MSG)
- assertTrue(false, INITIAL_TOTAL_READ_COUNTS_NOT_GREATER_THAN_0_MSG)
- }
-
- // Store initial values
- double initialHitCounts = Double.valueOf(initialHitCountsResult[0][0])
- double initialReadCounts = Double.valueOf(initialReadCountsResult[0][0])
-
- (1..iterations).each { count ->
- Thread.sleep(interval * 1000)
- def elapsedSeconds = count * interval
- def remainingSeconds = totalWaitTime - elapsedSeconds
- logger.info("Waited for file cache statistics update ${elapsedSeconds}
seconds, ${remainingSeconds} seconds remaining")
- }
+ Double hitsRatio = cacheMetricMax('hits_ratio')
+ Double hitsRatio1h = cacheMetricMax('hits_ratio_1h')
+ Double hitsRatio5m = cacheMetricMax('hits_ratio_5m')
+ logger.info("Hit ratio metrics - hits_ratio: ${hitsRatio}, hits_ratio_1h:
${hitsRatio1h}, hits_ratio_5m: ${hitsRatio5m}")
+ assertTrue(hitsRatio > 0.0, HIT_RATIO_METRIC_FALSE_MSG)
+ assertTrue(hitsRatio1h > 0.0, HIT_RATIO_1H_METRIC_FALSE_MSG)
+ assertTrue(hitsRatio5m > 0.0, HIT_RATIO_5M_METRIC_FALSE_MSG)
+
+ Double normalQueueCurrSize = cacheMetricSum('normal_queue_curr_size')
+ Double normalQueueMaxSize = cacheMetricSum('normal_queue_max_size')
+ Double normalQueueCurrElements =
cacheMetricSum('normal_queue_curr_elements')
+ Double normalQueueMaxElements = cacheMetricSum('normal_queue_max_elements')
+ logger.info("Normal queue metrics - curr_size: ${normalQueueCurrSize},
max_size: ${normalQueueMaxSize}, " +
+ "curr_elements: ${normalQueueCurrElements}, max_elements:
${normalQueueMaxElements}")
+ assertTrue(normalQueueCurrSize > 0.0 && normalQueueCurrSize <
normalQueueMaxSize,
+ NORMAL_QUEUE_SIZE_VALIDATION_FAILED_MSG)
+ assertTrue(normalQueueCurrElements > 0.0 && normalQueueCurrElements <
normalQueueMaxElements,
+ NORMAL_QUEUE_ELEMENTS_VALIDATION_FAILED_MSG)
+
+ Double initialHitCounts = cacheMetricSum('total_hit_counts')
+ Double initialReadCounts = cacheMetricSum('total_read_counts')
+ logger.info("Initial hit/read counts - hit_counts: ${initialHitCounts},
read_counts: ${initialReadCounts}")
// Execute the same query to trigger cache operations
- order_qt_2 """select * from
${catalog_name}.${ex_db_name}.parquet_partition_table
- where l_orderkey=1 and l_partkey=1534 limit 1;"""
-
- // Get updated values after cache operations
- def updatedHitCountsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'total_hit_counts';"""
- logger.info("Updated total_hit_counts result: " + updatedHitCountsResult)
+ order_qt_2 querySql
- def updatedReadCountsResult = sql """select SUM(CAST(METRIC_VALUE AS
DOUBLE)) from information_schema.file_cache_statistics
- where METRIC_NAME = 'total_read_counts';"""
- logger.info("Updated total_read_counts result: " + updatedReadCountsResult)
-
- // Check if updated values are greater than initial values
- double updatedHitCounts = Double.valueOf(updatedHitCountsResult[0][0])
- double updatedReadCounts = Double.valueOf(updatedReadCountsResult[0][0])
+ awaitUntil(pollTimeoutSeconds, 1) {
+ Double updatedHitCounts = cacheMetricSum('total_hit_counts')
+ Double updatedReadCounts = cacheMetricSum('total_read_counts')
+ return updatedHitCounts != null && updatedHitCounts > initialHitCounts
&&
+ updatedReadCounts != null && updatedReadCounts >
initialReadCounts
+ }
- boolean hitCountsIncreased = updatedHitCounts > initialHitCounts
- boolean readCountsIncreased = updatedReadCounts > initialReadCounts
+ Double updatedHitCounts = cacheMetricSum('total_hit_counts')
+ Double updatedReadCounts = cacheMetricSum('total_read_counts')
logger.info("Hit and read counts comparison - hit_counts:
${initialHitCounts} -> " +
- "${updatedHitCounts} (increased: ${hitCountsIncreased}), read_counts:
${initialReadCounts} -> " +
- "${updatedReadCounts} (increased: ${readCountsIncreased})")
+ "${updatedHitCounts}, read_counts: ${initialReadCounts} ->
${updatedReadCounts}")
- if (!hitCountsIncreased) {
- logger.info(TOTAL_HIT_COUNTS_DID_NOT_INCREASE_MSG)
- assertTrue(false, TOTAL_HIT_COUNTS_DID_NOT_INCREASE_MSG)
- }
- if (!readCountsIncreased) {
- logger.info(TOTAL_READ_COUNTS_DID_NOT_INCREASE_MSG)
- assertTrue(false, TOTAL_READ_COUNTS_DID_NOT_INCREASE_MSG)
- }
- // ===== End Hit and Read Counts Metrics Check =====
- sql """set global enable_file_cache=false"""
+ assertTrue(updatedHitCounts > initialHitCounts,
TOTAL_HIT_COUNTS_DID_NOT_INCREASE_MSG)
+ assertTrue(updatedReadCounts > initialReadCounts,
TOTAL_READ_COUNTS_DID_NOT_INCREASE_MSG)
return true
}
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
