github-actions[bot] commented on code in PR #65153: URL: https://github.com/apache/doris/pull/65153#discussion_r3575724549
########## regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/on_tables/test_warm_up_event_on_tables_overlap_semantics.groovy: ########## @@ -0,0 +1,125 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_warm_up_event_on_tables_overlap_semantics', 'docker') { + def options = new ClusterOptions() + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'cloud_warm_up_job_scheduler_interval_millisecond=100', + 'cloud_warm_up_table_filter_refresh_interval_ms=1000', + ] + options.beConfigs += [ + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'file_cache_background_monitor_interval_ms=1000', + ] + options.cloudMode = true + options.beNum = 1 + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + + def srcCluster = "warmup_overlap_src" + def dstCluster = "warmup_overlap_dst" + def dbName = "test_warmup_overlap_semantics_db" + def tableName = "overlap_tbl" + def otherTable = "overlap_other_tbl" + def jobIds = [] + + cluster.addBackend(1, srcCluster) + cluster.addBackend(1, dstCluster) + + sql """use @${srcCluster}""" + sql """CREATE DATABASE IF NOT EXISTS ${dbName}""" + sql """use ${dbName}""" + sql """CREATE TABLE IF NOT EXISTS ${tableName} (id INT, val STRING) + DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + sql """CREATE TABLE IF NOT EXISTS ${otherTable} (id INT, val STRING) + DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + + def preciseJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES (INCLUDE '${dbName}.${tableName}') + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + def overlapJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES ( + INCLUDE '${dbName}.*', + EXCLUDE '${dbName}.${otherTable}' + ) + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + def containerJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES (INCLUDE '${dbName}.*') + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + jobIds.addAll([preciseJobId, overlapJobId, containerJobId]) + + assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, preciseJobId, + ["${dbName}.${tableName}".toString()] as Set, + ["${dbName}.${otherTable}".toString()] as Set) == + ["${dbName}.${tableName}".toString()] as Set + assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, overlapJobId, Review Comment: `waitForMatchedTables()` can still return a timeout snapshot that violates `expectedNotContains`, and this caller only checks that the target table is present. If the overlap job keeps matching both `overlap_tbl` and `overlap_other_tbl` until timeout, the returned set still satisfies `.contains(overlap_tbl)`, so the test passes without proving the EXCLUDE rule. Please either make `waitForMatchedTables()` fail on timeout or assert here that the returned set excludes `otherTable` as well. ########## regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_once_queue_semantics.groovy: ########## @@ -0,0 +1,111 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import groovy.json.JsonOutput +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_vcg_warmup_with_manual_once_queue_semantics', 'docker') { + def options = new ClusterOptions() + options.feNum = 3 + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'sys_log_verbose_modules=org', + 'fetch_cluster_cache_hotspot_interval_ms=1000', + ] + options.cloudMode = true + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + + def ms = cluster.getAllMetaservices().get(0) + def msHttpPort = ms.host + ":" + ms.httpPort + def instanceId = "default_instance_id" + def vcgSrc = "vcg_queue_src" + def sharedDst = "vcg_queue_dst" + def manualSrc = "vcg_queue_manual_src" + def vcgName = "vcgWarmupManualOnceQueue" + def vcgId = "vcgWarmupManualOnceQueueId" + def createdJobIds = [] + + cluster.addBackend(1, vcgSrc) + cluster.addBackend(1, sharedDst) + cluster.addBackend(1, manualSrc) + + def addClusterApi = { requestBody, Closure checkFunc -> + httpTest { + endpoint msHttpPort + uri "/MetaService/http/add_cluster?token=$token" + body requestBody + check checkFunc + } + } + + def vcgBody = JsonOutput.toJson([ + instance_id: instanceId, + cluster: [ + cluster_name : vcgName, + cluster_id : vcgId, + type : "VIRTUAL", + cluster_names : [vcgSrc, sharedDst], + cluster_policy : [type: "ActiveStandby", active_cluster_name: vcgSrc, + standby_cluster_names: [sharedDst]] + ] + ]) + + try { + addClusterApi(vcgBody) { respCode, body -> + def json = parseJson(body) + assertTrue(json.code.equalsIgnoreCase("OK")) + } + + def autoRows = WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, vcgSrc, sharedDst, 2, 120000) + createdJobIds.addAll(autoRows*.jobId) + assertTrue(autoRows.any { it.syncMode.startsWith("PERIODIC") }) + assertTrue(autoRows.any { it.syncMode.startsWith("EVENT_DRIVEN") }) + + def manualOnceId = sql("""WARM UP CLUSTER ${sharedDst} WITH CLUSTER ${manualSrc}""")[0][0] + createdJobIds << manualOnceId.toString() + + def rowsByDst = WarmupMetricsUtils.showWarmupJobsByDst(sqlRunner, sharedDst) + assertTrue(rowsByDst*.jobId.contains(manualOnceId.toString())) + + def expectedNormalJobIds = (autoRows.findAll { WarmupMetricsUtils.isNormalWarmupJob(it) }*.jobId + + [manualOnceId.toString()]) + WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner, sharedDst, 30000, expectedNormalJobIds) Review Comment: Because `expectedNormalJobIds` includes the auto VCG normal job IDs, this wait can be satisfied by the auto periodic job alone. The assertions below only prove the manual once job row exists and targets `sharedDst`; they do not prove `manualOnceId` ever left PENDING, so starvation behind the auto VCG job would still pass. Please wait for the manual job ID specifically to reach RUNNING or FINISHED, and apply the same fix to the manual-periodic queue suite. ########## regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_master_fe.groovy: ########## @@ -0,0 +1,140 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import groovy.json.JsonOutput +import groovy.json.JsonSlurper +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_vcg_warmup_restart_master_fe', 'docker') { + def options = new ClusterOptions() + options.feNum = 3 + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'sys_log_verbose_modules=org', + 'fetch_cluster_cache_hotspot_interval_ms=1000', + ] + options.cloudMode = true + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + + def ms = cluster.getAllMetaservices().get(0) + def msHttpPort = ms.host + ":" + ms.httpPort + def instanceId = "default_instance_id" + def srcCluster = "vcg_restart_src" + def dstCluster = "vcg_restart_dst" + def vcgName = "vcgWarmupRestartMasterFe" + def vcgId = "vcgWarmupRestartMasterFeId" + def knownJobIds = [] + def jsonSlurper = new JsonSlurper() + + cluster.addBackend(1, srcCluster) + cluster.addBackend(1, dstCluster) + + def addClusterApi = { requestBody, Closure checkFunc -> + httpTest { + endpoint msHttpPort + uri "/MetaService/http/add_cluster?token=$token" + body requestBody + check checkFunc + } + } + + def restartMasterFe = { + def oldMasterFe = cluster.getMasterFe() + cluster.restartFrontends(oldMasterFe.index) + boolean restarted = false + for (int i = 0; i < 30; i++) { + if (cluster.getFeByIndex(oldMasterFe.index).alive) { + restarted = true + break + } + sleep(1000) + } + assertTrue(restarted) + context.reconnectFe() + } + + def getVcgWarmupJobIds = { + def groups = sql_return_maparray """SHOW COMPUTE GROUPS""" + def vcg = groups.find { it.Name == vcgName } + if (vcg == null) { + return [] + } + def policy = jsonSlurper.parseText(vcg.Policy) + def rawJobIds = policy.cacheWarmupJobIds ?: "[]" + if (rawJobIds instanceof Collection) { + return rawJobIds.collect { it.toString() } + } + return jsonSlurper.parseText(rawJobIds.toString()).collect { it.toString() } + } + + def waitForVcgWarmupJobIds = { Collection<String> excludedJobIds -> + List<String> jobIds = [] + awaitUntil(120, 1) { + jobIds = getVcgWarmupJobIds().findAll { + !excludedJobIds.contains(it) + } + jobIds.size() >= 2 + } + return jobIds + } + + def vcgBody = JsonOutput.toJson([ + instance_id: instanceId, + cluster: [ + cluster_name : vcgName, + cluster_id : vcgId, + type : "VIRTUAL", + cluster_names : [srcCluster, dstCluster], + cluster_policy : [type: "ActiveStandby", active_cluster_name: srcCluster, + standby_cluster_names: [dstCluster]] + ] + ]) + + try { + addClusterApi(vcgBody) { respCode, body -> + def json = parseJson(body) + assertTrue(json.code.equalsIgnoreCase("OK")) + } + + def beforeRestartRows = WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster, 2, 120000) + knownJobIds.addAll(beforeRestartRows*.jobId) + assertTrue(beforeRestartRows.any { it.syncMode.startsWith("PERIODIC") }) + assertTrue(beforeRestartRows.any { it.syncMode.startsWith("EVENT_DRIVEN") }) + assertTrue(getVcgWarmupJobIds().containsAll(knownJobIds)) + + restartMasterFe() + + WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner, knownJobIds, 120000) Review Comment: This assertion expects a plain master FE restart to cancel the old VCG warm-up jobs and publish new job IDs, but I do not see that path in the implementation. For an existing VCG, `diffAndUpdateComputeGroup()` only copies `cacheWarmupJobIds` from MS when the policy is otherwise unchanged, and `checkNeedRebuildFileCache()` rebuilds only when a referenced job is missing, cancelled, loses src/dst, or no longer matches active/standby. If the replayed jobs are valid, `needRebuildFileCache` stays false and this cancellation never happens. Please either change the test to assert the existing IDs remain active after restart, or add the intended restart-triggered rebuild path. ########## regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_add_new_be.groovy: ########## @@ -193,19 +293,35 @@ suite('test_warm_up_cluster_periodic_add_new_be', 'docker') { sql """truncate table __internal_schema.cloud_cache_hotspot;""" clearFileCacheOnAllBackends() - for (int i = 0; i < 1000; i++) { + def beforeWarmupMetrics = getPeriodicWarmupMetrics(clusterName2) + logFileCacheQueueMetrics(clusterName1, "after clear cache before warmup") + logFileCacheQueueMetrics(clusterName2, "after clear cache before warmup") + for (int i = 0; i < 500; i++) { sql """SELECT * FROM customer""" } - waitUntil({ getClusterTTLCacheSizeSum(clusterName1) > 0 }, 60000) - waitUntil({ getClusterTTLCacheSizeSum(clusterName1) == getClusterTTLCacheSizeSum(clusterName2) }, 60000) + waitUntil({ + def metrics = getPeriodicWarmupMetrics(clusterName2) + metrics.submitted > beforeWarmupMetrics.submitted && metrics.finished > beforeWarmupMetrics.finished + }, 120000, "periodic warmup metrics to advance from ${beforeWarmupMetrics}") - def hotspot = sql """select * from __internal_schema.cloud_cache_hotspot;""" - logger.info("hotspot: {}", hotspot) + waitUntil({ getClusterFileCacheSizeSum(clusterName1) > 0 }, + 60000, "${clusterName1} file cache size > 0") + waitUntil({ + logFileCacheQueueMetrics(clusterName1, "wait for warmup") + logFileCacheQueueMetrics(clusterName2, "wait for warmup") + getClusterFileCacheSizeSum(clusterName1) == getClusterFileCacheSizeSum(clusterName2) Review Comment: This still only proves the target cluster aggregate warmed, not that the two newly added target BEs warmed anything. Since `cluster.addBackend(2, clusterName2)` returns the new BE indexes but the test ignores them, a regression that keeps all warmed data on the original target BE can still satisfy the aggregate `file_cache_cache_size` equality here. Please capture the added BE indexes and assert each new target BE has nonzero cache size or advanced once/periodic warm-up metrics after the periodic round. ########## regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_scheduler_semantics.groovy: ########## @@ -0,0 +1,137 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_warm_up_cluster_periodic_scheduler_semantics', 'docker') { + def options = new ClusterOptions() + options.feNum = 3 + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'fetch_cluster_cache_hotspot_interval_ms=1000', + ] + options.beConfigs += [ + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'file_cache_background_monitor_interval_ms=1000', + ] + options.cloudMode = true + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + def srcCluster = "warmup_periodic_src" + def dstCluster = "warmup_periodic_dst" + def dbName = "test_warmup_periodic_scheduler_db" + def tableName = "periodic_tbl" + + cluster.addBackend(1, srcCluster) + cluster.addBackend(1, dstCluster) + + def restartMasterFe = { + def oldMasterFe = cluster.getMasterFe() + cluster.restartFrontends(oldMasterFe.index) + boolean restarted = false + for (int i = 0; i < 30; i++) { + if (cluster.getFeByIndex(oldMasterFe.index).alive) { + restarted = true + break + } + sleep(1000) + } + assertTrue(restarted) + context.reconnectFe() + } + + def getPeriodicWarmupMetrics = { + [ + submitted: WarmupMetricsUtils.getClusterMetricSum(sqlRunner, dstCluster, + "file_cache_once_or_periodic_warm_up_submitted_segment_num"), + finished : WarmupMetricsUtils.getClusterMetricSum(sqlRunner, dstCluster, + "file_cache_once_or_periodic_warm_up_finished_segment_num"), + ] + } + + def waitForPeriodicWarmupFinish = { Map before, long timeoutMs -> + long deadline = System.currentTimeMillis() + timeoutMs + Map latest = getPeriodicWarmupMetrics() + while (System.currentTimeMillis() < deadline) { + latest = getPeriodicWarmupMetrics() + if (latest.submitted > before.submitted && latest.finished > before.finished + && latest.finished >= latest.submitted) { + return latest + } + sleep(2000) + } + logger.warn("periodic warmup metrics did not advance after ${timeoutMs}ms, " + + "before=${before}, latest=${latest}") + return latest + } + + sql """use @${srcCluster}""" + sql """CREATE DATABASE IF NOT EXISTS ${dbName}""" + sql """use ${dbName}""" + sql """CREATE TABLE IF NOT EXISTS ${tableName} ( + id INT, + value STRING + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + for (int i = 0; i < 12; i++) { + sql """INSERT INTO ${tableName} VALUES (${i}, 'before_${i}')""" + } + + def periodicJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + PROPERTIES ( + "sync_mode" = "periodic", + "sync_interval_sec" = "1" + ) + """)[0][0] + + def cycles = WarmupMetricsUtils.waitForPeriodicCycles(sqlRunner, periodicJobId, 1, 120000) + assertTrue(cycles.states.contains("RUNNING")) + assertTrue(cycles.states.any { it in ["PENDING", "WAITING"] }) + + def timelineBeforeRestart = WarmupMetricsUtils.sampleJobTimeline(sqlRunner, periodicJobId, 5000, 1000) + assertTrue(timelineBeforeRestart*.status.any { it in ["RUNNING", "PENDING", "WAITING"] }) + + restartMasterFe() + sql """use @${srcCluster}""" + sql """use ${dbName}""" + + def timelineAfterRestart = WarmupMetricsUtils.sampleJobTimeline(sqlRunner, periodicJobId, 8000, 1000) + assertTrue(timelineAfterRestart*.status.any { it in ["RUNNING", "PENDING", "WAITING"] }, + "periodic job should continue after FE restart, timeline=${timelineAfterRestart}") + + def beforeMetrics = getPeriodicWarmupMetrics() + for (int i = 12; i < 16; i++) { + sql """INSERT INTO ${tableName} VALUES (${i}, 'after_restart_${i}')""" + } + for (int i = 0; i < 1000; i++) { + sql """SELECT * FROM ${tableName}""" + } + def afterMetrics = waitForPeriodicWarmupFinish(beforeMetrics, 120000) + assertTrue(afterMetrics.finished > beforeMetrics.finished, Review Comment: This wait helper can time out and still return `latest`, but the assertion below only checks `finished > before.finished`. That allows the test to pass if a pre-existing in-flight segment finishes after `beforeMetrics`, or if a post-restart round submits work but never drains. Please make the helper fail on timeout and keep the final assertion aligned with the waited condition (`submitted` advanced and `finished >= submitted`). ########## regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_normal_queue_semantics.groovy: ########## @@ -0,0 +1,146 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_warm_up_normal_queue_semantics', 'docker') { + def options = new ClusterOptions() + options.feNum = 3 + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'fetch_cluster_cache_hotspot_interval_ms=1000', + ] + options.beConfigs += [ + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'file_cache_background_monitor_interval_ms=1000', + ] + options.cloudMode = true + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + + def srcCluster = "warmup_queue_src" + def dstCluster = "warmup_queue_dst" + def auxCluster = "warmup_queue_aux" + def dbName = "test_warmup_queue_semantics_db" + def tableName = "queue_tbl" + + cluster.addBackend(1, srcCluster) + cluster.addBackend(1, dstCluster) + cluster.addBackend(1, auxCluster) + + def restartMasterFe = { + def oldMasterFe = cluster.getMasterFe() + cluster.restartFrontends(oldMasterFe.index) + boolean restarted = false + for (int i = 0; i < 30; i++) { + if (cluster.getFeByIndex(oldMasterFe.index).alive) { + restarted = true + break + } + sleep(1000) + } + assertTrue(restarted) + context.reconnectFe() + } + + sql """use @${srcCluster}""" + sql """CREATE DATABASE IF NOT EXISTS ${dbName}""" + sql """use ${dbName}""" + sql """DROP TABLE IF EXISTS ${tableName}""" + sql """CREATE TABLE ${tableName} ( + id INT, + name STRING + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + for (int i = 0; i < 20; i++) { + sql """INSERT INTO ${tableName} VALUES (${i}, 'name_${i}')""" + } + + def periodicJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + PROPERTIES ( + "sync_mode" = "periodic", + "sync_interval_sec" = "1" + ) + """)[0][0] + def onceTableJobId = sql("""WARM UP CLUSTER ${dstCluster} WITH TABLE ${tableName}""")[0][0] + def onceClusterJobId = sql("""WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${auxCluster}""")[0][0] + def eventJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES (INCLUDE '${dbName}.${tableName}') + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + + def queueRows = WarmupMetricsUtils.showWarmupJobsByDst(sqlRunner, dstCluster) + assert queueRows.find { it.jobId == periodicJobId.toString() } != null + assert queueRows.find { it.jobId == onceTableJobId.toString() } != null + assert queueRows.find { it.jobId == onceClusterJobId.toString() } != null + assert queueRows.find { it.jobId == eventJobId.toString() } != null + + def normalJobIds = [periodicJobId, onceTableJobId, onceClusterJobId].collect { it.toString() } + WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner, dstCluster, 30000, normalJobIds) + long runningNormal = WarmupMetricsUtils.countRunningNormalWarmupByDst(sqlRunner, dstCluster) + assertTrue(runningNormal <= 1, + "same dst cluster should have at most one running normal warmup, got ${runningNormal}") + + def eventRow = WarmupMetricsUtils.showWarmupJob(sqlRunner, eventJobId) + assertTrue(WarmupMetricsUtils.isEventDrivenWarmupJob(eventRow)) + assertTrue(eventRow.status in ["RUNNING", "PENDING", "WAITING"]) + + Map<String, Map> beforeRestartSnapshot = WarmupMetricsUtils.snapshotWarmupJobsById(sqlRunner) + restartMasterFe() + sql """use @${srcCluster}""" + sql """use ${dbName}""" + Map<String, Map> afterRestartSnapshot = WarmupMetricsUtils.waitForWarmupJobsRecovered( + sqlRunner, beforeRestartSnapshot, { before, current -> + [periodicJobId, onceTableJobId, onceClusterJobId, eventJobId].every { current.containsKey(it.toString()) } + }, 60000) + assertTrue(afterRestartSnapshot.keySet().containsAll( + [periodicJobId, onceTableJobId, onceClusterJobId, eventJobId].collect { it.toString() })) + + WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner, dstCluster, 30000, normalJobIds) Review Comment: This still does not force the replay path that can lose the per-destination lock. `clusterToRunningJobId` is rebuilt only when a PENDING job calls `tryRegisterRunningJob()`, while replay restores unfinished jobs into the runnable maps and a replayed RUNNING normal job resumes through `runRunningJob()` without registering itself. If FE restarts with one same-destination normal job RUNNING and another PENDING, the empty lock lets the PENDING job register too. Please either restore the lock for replayed RUNNING ONCE/PERIODIC jobs, or make this test deterministic by holding one job RUNNING across restart and proving the pending job cannot run until it releases. ########## regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/on_tables/test_warm_up_event_on_tables_overlap_semantics.groovy: ########## @@ -0,0 +1,125 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.WarmupMetricsUtils + +suite('test_warm_up_event_on_tables_overlap_semantics', 'docker') { + def options = new ClusterOptions() + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'cloud_warm_up_job_scheduler_interval_millisecond=100', + 'cloud_warm_up_table_filter_refresh_interval_ms=1000', + ] + options.beConfigs += [ + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'file_cache_background_monitor_interval_ms=1000', + ] + options.cloudMode = true + options.beNum = 1 + + docker(options) { + Closure sqlRunner = { String q -> sql(q) } + + def srcCluster = "warmup_overlap_src" + def dstCluster = "warmup_overlap_dst" + def dbName = "test_warmup_overlap_semantics_db" + def tableName = "overlap_tbl" + def otherTable = "overlap_other_tbl" + def jobIds = [] + + cluster.addBackend(1, srcCluster) + cluster.addBackend(1, dstCluster) + + sql """use @${srcCluster}""" + sql """CREATE DATABASE IF NOT EXISTS ${dbName}""" + sql """use ${dbName}""" + sql """CREATE TABLE IF NOT EXISTS ${tableName} (id INT, val STRING) + DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + sql """CREATE TABLE IF NOT EXISTS ${otherTable} (id INT, val STRING) + DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("file_cache_ttl_seconds" = "3600")""" + + def preciseJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES (INCLUDE '${dbName}.${tableName}') + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + def overlapJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES ( + INCLUDE '${dbName}.*', + EXCLUDE '${dbName}.${otherTable}' + ) + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + def containerJobId = sql(""" + WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster} + ON TABLES (INCLUDE '${dbName}.*') + PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load") + """)[0][0] + jobIds.addAll([preciseJobId, overlapJobId, containerJobId]) + + assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, preciseJobId, + ["${dbName}.${tableName}".toString()] as Set, + ["${dbName}.${otherTable}".toString()] as Set) == + ["${dbName}.${tableName}".toString()] as Set + assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, overlapJobId, + ["${dbName}.${tableName}".toString()] as Set, + ["${dbName}.${otherTable}".toString()] as Set) + .contains("${dbName}.${tableName}".toString()) + assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, containerJobId, + ["${dbName}.${tableName}".toString(), "${dbName}.${otherTable}".toString()] as Set) + .containsAll(["${dbName}.${tableName}".toString(), "${dbName}.${otherTable}".toString()] as Set) + jobIds.each { jobId -> + WarmupMetricsUtils.waitForJobStatus(sqlRunner, jobId, ["RUNNING"], 60000) + } + Thread.sleep(1000) + + def baseMetrics = WarmupMetricsUtils.getWarmupMetrics(sqlRunner, srcCluster, dstCluster) + for (int i = 0; i < 4; i++) { + sql """INSERT INTO ${tableName} VALUES (${i}, 'target_${i}')""" + sql """INSERT INTO ${otherTable} VALUES (${i}, 'other_${i}')""" + } + def finalMetrics = WarmupMetricsUtils.waitForWarmupFinish(sqlRunner, srcCluster, dstCluster, + baseMetrics.finished + 1, 120000) + assertTrue(finalMetrics.requested > baseMetrics.requested, + "overlap jobs should all receive triggers, before=${baseMetrics}, after=${finalMetrics}") + + def preciseStats = WarmupMetricsUtils.waitForJobSyncStats(sqlRunner, preciseJobId, { Review Comment: These per-job checks only wait for source-side `requested_5m`. The aggregate wait above proves at most one destination-side finish for the whole source/destination pair, but `aggregateStatsForJob()` reports requested stats from the source cluster and finish/gap/fail stats from the destination cluster. If a job is triggered but its selected rowsets never finish warming on the target, these predicates still pass. Please make the per-job waits require destination completion too, for example `finish_5m >= requested_5m`, `gap_5m == 0`, and `fail_5m == 0` for each of the precise/overlap/container jobs. ########## docker/runtime/doris-compose/utils.py: ########## @@ -242,26 +242,44 @@ def _command_available(cmd): except (subprocess.CalledProcessError, FileNotFoundError): return False + def _compose_usable(cmd): + if not _command_available(cmd + ["version"]): + return False + try: + # `ls` talks to the daemon and catches API-version mismatches that + # `version` alone misses on mixed docker/compose installations. + subprocess.run( + cmd + ["ls"], Review Comment: This daemon check is also used for the `docker-compose` fallback below, but `ls` is a Compose v2 subcommand. The function still advertises `docker-compose v1 / wrapper` support and this directory's requirements still install `docker-compose`, so a host with only the legacy v1 client can now fail detection even though `docker-compose up/down` would have worked. Please keep the daemon validation, but use a v1-supported daemon-touching command for the standalone fallback or explicitly remove/narrow v1 support in the contract and requirements. ########## regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_shared_compute_group_conflict.groovy: ########## @@ -0,0 +1,99 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import groovy.json.JsonOutput +import org.apache.doris.regression.suite.ClusterOptions + +suite('test_vcg_warmup_shared_compute_group_conflict', 'docker') { + def options = new ClusterOptions() + options.feNum = 3 + options.feConfigs += [ + 'cloud_cluster_check_interval_second=1', + 'sys_log_verbose_modules=org', + ] + options.cloudMode = true + + docker(options) { + def ms = cluster.getAllMetaservices().get(0) + def msHttpPort = ms.host + ":" + ms.httpPort + def instanceId = "default_instance_id" + def clusterA = "vcg_share_a" + def clusterB = "vcg_share_b" + def clusterC = "vcg_share_c" + + cluster.addBackend(1, clusterA) + cluster.addBackend(1, clusterB) + cluster.addBackend(1, clusterC) + + def addClusterApi = { requestBody, Closure checkFunc -> + httpTest { + endpoint msHttpPort + uri "/MetaService/http/add_cluster?token=$token" + body requestBody + check checkFunc + } + } + + def firstBody = JsonOutput.toJson([ + instance_id: instanceId, + cluster: [ + cluster_name : "vcgShareFirst", + cluster_id : "vcgShareFirstId", + type : "VIRTUAL", + cluster_names : [clusterA, clusterB], + cluster_policy : [type: "ActiveStandby", active_cluster_name: clusterA, + standby_cluster_names: [clusterB]] + ] + ]) + addClusterApi(firstBody) { respCode, body -> + def json = parseJson(body) + assertTrue(json.code.equalsIgnoreCase("OK")) + } + + def secondBody = JsonOutput.toJson([ + instance_id: instanceId, + cluster: [ + cluster_name : "vcgShareSecond", + cluster_id : "vcgShareSecondId", + type : "VIRTUAL", + cluster_names : [clusterA, clusterC], + cluster_policy : [type: "ActiveStandby", active_cluster_name: clusterA, + standby_cluster_names: [clusterC]] + ] + ]) + addClusterApi(secondBody) { respCode, body -> + def json = parseJson(body) + assertTrue(!json.code.equalsIgnoreCase("OK")) Review Comment: This only verifies that the second `add_cluster` call failed somehow; it does not prove the intended shared-subcluster conflict path. `validate_sub_clusters()` returns a specific `INVALID_ARGUMENT` error naming the subcluster and existing VCG, and the C++ HTTP test already asserts that status/code/message. Please assert `respCode == 400`, `json.code == "INVALID_ARGUMENT"`, and a message containing the shared subcluster and existing VCG name for both rejected requests, otherwise unrelated failures such as missing subclusters or malformed policy would still pass this regression. ########## regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy: ########## @@ -265,4 +265,308 @@ class WarmupMetricsUtils { logger.warn("waitForMetricsStable timed out after ${timeoutMs}ms") return getWarmupMetrics(sqlRunner, srcCluster, dstCluster) } + + static List<Map> showWarmupJobs(Closure sqlRunner) { + def rows = sqlRunner("SHOW WARM UP JOB") + return rows.collect { row -> normalizeWarmupJobRow(row) } + } + + static Map showWarmupJob(Closure sqlRunner, Object jobId) { + def rows = sqlRunner("SHOW WARM UP JOB WHERE ID = ${jobId}") + if (rows == null || rows.isEmpty()) { + throw new RuntimeException("warmup job ${jobId} not found") + } + return normalizeWarmupJobRow(rows[0]) + } + + static Map normalizeWarmupJobRow(Object row) { + List values = row as List + return [ + jobId : values[0]?.toString(), + srcCluster : values[1]?.toString(), + dstCluster : values[2]?.toString(), + status : values[3]?.toString(), + type : values[4]?.toString(), + syncMode : values[5]?.toString(), + createTime : values[6]?.toString(), + startTime : values[7]?.toString(), + finishedTime : values[8]?.toString(), + allBatch : values[9]?.toString(), + finishedBatch : values[10]?.toString(), + errMsg : values[11]?.toString(), + tableOrPattern : values.size() > 12 ? values[12]?.toString() : "", + tableFilter : values.size() > 13 ? values[13]?.toString() : "", + matchedTables : values.size() > 14 ? values[14]?.toString() : "", + syncStats : values.size() > 15 ? values[15]?.toString() : "", + raw : values, + ] + } + + static boolean isNormalWarmupJob(Map row) { + String syncMode = row.syncMode ?: "" + return syncMode.startsWith("ONCE") || syncMode.startsWith("PERIODIC") + } + + static boolean isEventDrivenWarmupJob(Map row) { + String syncMode = row.syncMode ?: "" + return syncMode.startsWith("EVENT_DRIVEN") + } + + static boolean isActiveWarmupJob(Map row) { + return row.status in ["PENDING", "RUNNING", "WAITING"] + } + + static boolean isClusterWarmupJob(Map row) { + return row.type == "CLUSTER" + } + + static List<Map> showWarmupJobsByDst(Closure sqlRunner, String dstCluster) { + return showWarmupJobs(sqlRunner).findAll { it.dstCluster == dstCluster } + } + + static long countRunningNormalWarmupByDst(Closure sqlRunner, String dstCluster) { + return showWarmupJobsByDst(sqlRunner, dstCluster).count { + isNormalWarmupJob(it) && it.status == "RUNNING" + } as long + } + + static List<Map> snapshotWarmupJobs(Closure sqlRunner) { + return showWarmupJobs(sqlRunner).collect { new LinkedHashMap<>(it) } + } + + static Map<String, Map> snapshotWarmupJobsById(Closure sqlRunner) { + Map<String, Map> result = [:] + snapshotWarmupJobs(sqlRunner).each { result[it.jobId] = it } + return result + } + + static void waitForOnlyOneRunningNormalWarmup(Closure sqlRunner, String dstCluster, + long timeoutMs = 30000, + Collection<String> expectedJobIds = null, + long observeMs = 5000) { + long deadline = System.currentTimeMillis() + timeoutMs + List<Map> lastRunningRows = [] + while (System.currentTimeMillis() < deadline) { + List<Map> runningRows = showWarmupJobsByDst(sqlRunner, dstCluster).findAll { + isNormalWarmupJob(it) && it.status == "RUNNING" + } + lastRunningRows = runningRows + if (runningRows.size() > 1) { + throw new RuntimeException("expected at most one running normal warmup on ${dstCluster}, " + + "runningRows=${runningRows}") + } + boolean hasExpectedRunning = expectedJobIds == null || runningRows.any { + expectedJobIds.contains(it.jobId) + } + if (runningRows.size() == 1 && hasExpectedRunning) { + assertAtMostOneRunningNormalWarmupDuring(sqlRunner, dstCluster, observeMs) + return + } + Thread.sleep(1000) + } + throw new RuntimeException("expected one normal warmup to reach RUNNING on ${dstCluster}, " + + "expectedJobIds=${expectedJobIds}, lastRunningRows=${lastRunningRows}") + } + + static void assertAtMostOneRunningNormalWarmupDuring(Closure sqlRunner, String dstCluster, long observeMs) { + long deadline = System.currentTimeMillis() + observeMs + while (System.currentTimeMillis() < deadline) { + List<Map> runningRows = showWarmupJobsByDst(sqlRunner, dstCluster).findAll { + isNormalWarmupJob(it) && it.status == "RUNNING" + } + if (runningRows.size() > 1) { + throw new RuntimeException("expected at most one running normal warmup on ${dstCluster}, " + + "runningRows=${runningRows}") + } + Thread.sleep(1000) + } + } + + static Map waitForWarmupJobsRecovered(Closure sqlRunner, Map<String, Map> beforeSnapshot, + Closure<Boolean> predicate, long timeoutMs = 60000) { + long deadline = System.currentTimeMillis() + timeoutMs + Map<String, Map> current = [:] + while (System.currentTimeMillis() < deadline) { + current = snapshotWarmupJobsById(sqlRunner) + if (predicate(beforeSnapshot, current)) { + return current + } + Thread.sleep(1000) + } + return current + } + + static Map waitForPeriodicCycles(Closure sqlRunner, Object jobId, int minTransitions, + long timeoutMs = 120000) { + long deadline = System.currentTimeMillis() + timeoutMs + List<String> states = [] + String lastState = null + while (System.currentTimeMillis() < deadline) { + def row = showWarmupJob(sqlRunner, jobId) + if (row.status != lastState) { + states << row.status + lastState = row.status + } + int transitions = 0 + for (int i = 1; i < states.size(); i++) { + if (states[i - 1] == "RUNNING" && (states[i] == "PENDING" || states[i] == "WAITING")) { + transitions++ + } + } + if (transitions >= minTransitions) { + return [states: states, row: row] + } + Thread.sleep(1000) + } + throw new RuntimeException("expected periodic job ${jobId} to observe ${minTransitions} " + + "RUNNING-to-waiting transitions within ${timeoutMs}ms, states=${states}") + } + + static List<Map> sampleJobTimeline(Closure sqlRunner, Object jobId, long durationMs, + long intervalMs = 1000) { + long deadline = System.currentTimeMillis() + durationMs + List<Map> samples = [] + while (System.currentTimeMillis() < deadline) { + def row = showWarmupJob(sqlRunner, jobId) + samples << [ + ts : System.currentTimeMillis(), + status : row.status, + start : row.startTime, + finished: row.finishedTime, + syncMode: row.syncMode, + ] + Thread.sleep(intervalMs) + } + return samples + } + + static List<Map> collectWarmupJobsByCluster(Closure sqlRunner, String clusterName) { + return showWarmupJobs(sqlRunner).findAll { + it.srcCluster == clusterName || it.dstCluster == clusterName + } + } + + static void assertAffectedJobsCancelled(Closure sqlRunner, Collection<String> jobIds, + long timeoutMs = 60000) { + long deadline = System.currentTimeMillis() + timeoutMs + Map<String, String> lastStates = [:] + while (System.currentTimeMillis() < deadline) { + boolean allCancelled = true + for (String jobId in jobIds) { + def row = showWarmupJob(sqlRunner, jobId) + lastStates[jobId] = row.status + if (row.status != "CANCELLED") { + allCancelled = false + } + } + if (allCancelled) { + return + } + Thread.sleep(1000) + } + throw new RuntimeException("expected jobs cancelled, lastStates=${lastStates}") + } + + static void assertUnrelatedJobsUnaffected(Closure sqlRunner, Collection<String> jobIds, + Collection<String> allowedStatuses = ["RUNNING", "PENDING", "FINISHED"]) { + for (String jobId in jobIds) { + def row = showWarmupJob(sqlRunner, jobId) + if (!allowedStatuses.contains(row.status)) { + throw new RuntimeException("unrelated warmup job ${jobId} entered unexpected status ${row.status}") + } + } + } + + static String expectCreateConflict(Closure sqlRunner, Closure createJob) { + try { + createJob.call() + } catch (Throwable t) { + return t.message ?: t.toString() + } + throw new RuntimeException("expected warmup create conflict, but statement succeeded") + } + + static void assertConflictMessage(String message, Collection<String> expectedKeywords) { + expectedKeywords.each { keyword -> + if (!(message?.toLowerCase()?.contains(keyword.toLowerCase()))) { + throw new RuntimeException("expected conflict message to contain '${keyword}', actual=${message}") + } + } + } + + static List<Map> getVcgWarmupJobs(Closure sqlRunner, String srcCluster, String dstCluster) { + return showWarmupJobs(sqlRunner).findAll { + it.srcCluster == srcCluster && it.dstCluster == dstCluster + && isActiveWarmupJob(it) && isClusterWarmupJob(it) + && (it.syncMode.startsWith("PERIODIC") || it.syncMode.startsWith("EVENT_DRIVEN")) + } + } + + static List<String> getVcgWarmupJobIds(Closure sqlRunner, String srcCluster, String dstCluster) { + return getVcgWarmupJobs(sqlRunner, srcCluster, dstCluster).collect { it.jobId } + } + + static boolean hasPeriodicAndEventDrivenWarmup(Collection<Map> rows) { + return rows.any { it.syncMode.startsWith("PERIODIC") } + && rows.any { it.syncMode.startsWith("EVENT_DRIVEN") } + } + + static List<Map> waitForWarmupJobsByPair(Closure sqlRunner, String srcCluster, String dstCluster, + int minCount = 1, long timeoutMs = 120000) { + long deadline = System.currentTimeMillis() + timeoutMs + List<Map> rows = [] + while (System.currentTimeMillis() < deadline) { + rows = showWarmupJobs(sqlRunner).findAll { Review Comment: This helper still accepts any `SHOW WARM UP JOB` row for the source/destination pair and returns the last rows on timeout. Since SHOW includes historical `cloudWarmUpJobs`, cancelled or finished VCG jobs can satisfy initial checks that only look for PERIODIC/EVENT_DRIVEN rows. Please make this fail on timeout and add an active/type/sync predicate for VCG callers, or have those callers use `getVcgWarmupJobs()` plus `hasPeriodicAndEventDrivenWarmup()` instead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
