github-actions[bot] commented on code in PR #65190:
URL: https://github.com/apache/doris/pull/65190#discussion_r3667766360
##########
be/src/common/config.cpp:
##########
@@ -1819,6 +1819,7 @@ DEFINE_Validator(concurrency_stats_dump_interval_ms,
DEFINE_mBool(cloud_mow_sync_rowsets_when_load_txn_begin, "true");
DEFINE_mBool(enable_cloud_make_rs_visible_on_be, "false");
+DEFINE_mBool(enable_cloud_random_segment_id, "true");
Review Comment:
[P1] Gate incompatible rowsets during rolling upgrades
With this default, the first upgraded BE that compacts can publish (for
example) `num_segments=1, segment_ids=[437]` and only `<rowset>_437.dat` to the
shared meta service. A pre-PR BE can sync that visible rowset, but its
CloudPB-to-RowsetMeta conversion cannot copy field 200 and its readers still
enumerate `[0, num_segments)`, so it opens `<rowset>_0.dat` and
queries/compactions fail during a rolling upgrade. Please keep production of
nonzero IDs disabled by default until every reader is capable, or gate it on a
cluster-wide minimum BE version/capability, with a mixed-version test.
##########
be/src/storage/rowset/rowset.cpp:
##########
@@ -108,10 +110,15 @@ bool Rowset::check_rowset_segment() {
std::string Rowset::get_rowset_info_str() {
std::string disk_size = PrettyPrinter::print(
static_cast<uint64_t>(_rowset_meta->total_disk_size()),
TUnit::BYTES);
- return fmt::format("[{}-{}] {} {} {} {} {}", start_version(),
end_version(), num_segments(),
- _rowset_meta->has_delete_predicate() ? "DELETE" :
"DATA",
-
SegmentsOverlapPB_Name(_rowset_meta->segments_overlap()),
- rowset_id().to_string(), disk_size);
+ std::string rowset_info =
+ fmt::format("[{}-{}] {} {} {} {} {}", start_version(),
end_version(), num_segments(),
+ _rowset_meta->has_delete_predicate() ? "DELETE" :
"DATA",
+
SegmentsOverlapPB_Name(_rowset_meta->segments_overlap()),
+ rowset_id().to_string(), disk_size);
+ if (_rowset_meta->has_segment_ids()) {
+ rowset_info += fmt::format(" [{}]",
fmt::join(_rowset_meta->segment_ids(), ","));
Review Comment:
[P1] Keep Cloud show-data size accounting compatible
This turns an explicit-ID rowset status into eight whitespace-separated
tokens, but `regression-test/plugins/cloud_show_data_plugin.groovy:215-222`
counts a rowset only when `fields.length == 7`. The show-data suites trigger
compaction and then compare this API total with MySQL/backend storage, so the
surviving randomized-ID rowset is silently omitted and those checks
under-report or fail. Please preserve the existing representation or update the
parser to handle the optional segment list (and add an explicit-ID parser case).
##########
regression-test/suites/compaction/test_mow_cumulative_compaction_multi_output_segments.groovy:
##########
@@ -0,0 +1,424 @@
+// 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.
+
+suite("test_mow_cumulative_compaction_multi_output_segments", "nonConcurrent")
{
+ if (!isCloudMode()) {
+ logger.info("skip test_mow_cumulative_compaction_multi_output_segments
in non-cloud mode")
+ return
+ }
+
+ def backendIdToHost = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToHost, backendIdToHttpPort)
+
+ def configNames = [
+ "compaction_batch_size",
+ "doris_scanner_row_bytes",
+ "enable_rowid_conversion_correctness_check",
+ "enable_vertical_compaction",
+ "inverted_index_compaction_enable",
+ "vertical_compaction_max_segment_size"
+ ]
+ def originalConfigs = [:]
+
+ def readBeConfig = { String backendId, String configName ->
+ def host = backendIdToHost[backendId]
+ def port = backendIdToHttpPort[backendId]
+ def (code, out, err) = curl(
+ "GET",
"http://${host}:${port}/api/show_config?conf_item=${configName}";)
+ assertEquals(0, code)
+ def configs = parseJson(out)
+ assertEquals(1, configs.size())
+ assertEquals(configName, configs[0][0])
+ return configs[0][2].toString()
+ }
+
+ def updateBeConfig = { String configName, String value ->
+ backendIdToHost.keySet().each { String backendId ->
+ def host = backendIdToHost[backendId]
+ def port = backendIdToHttpPort[backendId]
+ def (code, out, err) = curl(
+ "POST",
"http://${host}:${port}/api/update_config?${configName}=${value}";)
+ assertEquals(0, code)
+ assertTrue(out.contains("OK"), "failed to set
${configName}=${value}: ${out}, ${err}")
+ }
+ }
+
+ backendIdToHost.keySet().each { String backendId ->
+ originalConfigs[backendId] = [:]
+ configNames.each { String configName ->
+ originalConfigs[backendId][configName] = readBeConfig(backendId,
configName)
+ }
+ }
+
+ def resetBeConfigs = {
+ originalConfigs.each { String backendId, Map configs ->
+ def host = backendIdToHost[backendId]
+ def port = backendIdToHttpPort[backendId]
+ configs.each { String configName, String value ->
+ def (code, out, err) = curl(
+ "POST",
"http://${host}:${port}/api/update_config?${configName}=${value}";)
+ assertEquals(0, code)
+ assertTrue(out.contains("OK"),
+ "failed to reset ${configName}=${value}: ${out},
${err}")
+ }
+ }
+ }
+
+ def showTablet = { def backend, String tabletId ->
+ def (code, out, err) =
+ be_show_tablet_status(backend.Host, backend.HttpPort, tabletId)
+ logger.info("Show tablet status: code=${code}, out=${out}, err=${err}")
+ assertEquals(0, code)
+ return parseJson(out.trim())
+ }
+
+ def findRowset = { def tabletStatus, int startVersion, int endVersion ->
+ def rowset =
+ tabletStatus.rowsets.find {
it.startsWith("[${startVersion}-${endVersion}] ") }
+ assertNotNull(rowset,
+ "cannot find rowset [${startVersion}-${endVersion}]:
${tabletStatus.rowsets}")
+ return rowset
+ }
+
+ def parseRowset = { String rowset ->
+ def matcher = rowset =~
+ /\[[0-9]+-[0-9]+\]\s+([0-9]+)\s+DATA\s+([A-Z_]+)\s+([0-9a-f]+)/
+ assertTrue(matcher.find(), "unexpected rowset format: ${rowset}")
+ def segmentIds = []
+ def segmentIdsMatcher = rowset =~ /\s\[([0-9]+(?:,[0-9]+)*)\]$/
+ if (segmentIdsMatcher.find()) {
+ segmentIds = segmentIdsMatcher.group(1).split(",").collect {
it.toInteger() }
+ }
+ return [
+ segmentNum: matcher.group(1).toInteger(),
+ overlap: matcher.group(2),
+ rowsetId: matcher.group(3),
+ segmentIds: segmentIds
+ ]
+ }
+
+ def waitForCompaction = { def backend, String tabletId ->
+ long timeoutMs = 120000
+ long deadline = System.currentTimeMillis() + timeoutMs
+ def lastStatus = null
+ while (System.currentTimeMillis() < deadline) {
+ def (code, out, err) =
+ be_get_compaction_status(backend.Host, backend.HttpPort,
tabletId)
+ logger.info("Get compaction status: code=${code}, out=${out},
err=${err}")
+ assertEquals(0, code)
+ lastStatus = parseJson(out.trim())
+ assertEquals("success", lastStatus.status.toLowerCase())
+ if (!lastStatus.run_status) {
+ return
+ }
+ Thread.sleep(1000)
+ }
+ assertTrue(false,
+ "compaction timeout: tablet=${tabletId},
timeoutMs=${timeoutMs}, last=${lastStatus}")
+ }
+
+ def waitForRowset = { backend, tabletId, startVersion, endVersion ->
+ long timeoutMs = 30000
+ long deadline = System.currentTimeMillis() + timeoutMs
+ def lastStatus = null
+ while (System.currentTimeMillis() < deadline) {
+ lastStatus = showTablet(backend, tabletId)
+ if (lastStatus.rowsets.any {
+ it.startsWith("[${startVersion}-${endVersion}] ")
+ }) {
+ return lastStatus
+ }
+ Thread.sleep(200)
+ }
+ assertTrue(false,
+ "rowset [${startVersion}-${endVersion}] is not visible: " +
+ "tablet=${tabletId}, timeoutMs=${timeoutMs},
last=${lastStatus}")
+ }
+
+ def loadRows = { String tableName, int startKey, int endKey, int valueBase
->
+ def batchTags = [
+ 100000: "batchone",
+ 200000: "batchtwo",
+ 300000: "batchthree",
+ 400000: "batchfour",
+ 500000: "batchfive"
+ ]
+ def batchTag = batchTags[valueBase]
+ assertNotNull(batchTag)
+ StringBuilder content = new StringBuilder()
+ (startKey..endKey).each { int key ->
+ String suffix = key.toString().padLeft(32, "0")
+ String payload =
+ "${batchTag} payload ${suffix} ${(key *
17L).toString().padLeft(32, "0")}"
+ content.append("${key},${valueBase + key},${payload}\n")
+ }
+ streamLoad {
+ table tableName
+ set "column_separator", ","
+ inputStream new ByteArrayInputStream(content.toString().getBytes())
+ time 120000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(endKey - startKey + 1, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ }
+
+ def readRows = { String tableName ->
+ return sql("""
+ SELECT k, v
+ FROM ${tableName}
+ WHERE k IN (1, 16384, 16385, 32768, 32769, 49152, 49153, 65536,
65537, 81920)
+ ORDER BY k
+ """)
+ }
+
+ def readRowsByIndex = { String tableName ->
+ def batchTags = ["batchone", "batchtwo", "batchthree", "batchfour",
"batchfive"]
+ return batchTags.collect { String batchTag ->
+ def result = sql """
+ SELECT /*+ SET_VAR(enable_match_without_inverted_index =
false) */ COUNT(*)
+ FROM ${tableName}
+ WHERE payload MATCH '${batchTag}'
+ """
+ return result[0][0]
+ }
+ }
+
+ def getLocalDeleteBitmap = { def backend, String tabletId ->
+ def (code, out, err) = curl(
+ "GET",
+ "http://${backend.Host}:${backend.HttpPort}"; +
+
"/api/delete_bitmap/count_local?verbose=true&tablet_id=${tabletId}")
+ logger.info("Get local delete bitmap: code=${code}, out=${out},
err=${err}")
+ assertEquals(0, code)
+ return parseJson(out.trim())
+ }
+
+ def getDeleteBitmapAtVersion = { def deleteBitmap, String rowsetId, int
version ->
+ def segmentIds = []
+ long cardinality = 0
+ deleteBitmap.delete_bitmap.each { String key, bitmapVersions ->
+ if (!key.contains("rowset: ${rowsetId},")) {
+ return
+ }
+ def segmentMatcher = key =~ /segment:\s+([0-9]+)/
+ assertTrue(segmentMatcher.find(), "unexpected delete bitmap key:
${key}")
+ bitmapVersions.each { String bitmapVersion ->
+ def versionMatcher = bitmapVersion =~
/v:\s+([0-9]+),\s+c:\s+([0-9]+)/
+ assertTrue(versionMatcher.find(),
+ "unexpected delete bitmap value: ${bitmapVersion}")
+ if (versionMatcher.group(1).toInteger() == version) {
+ segmentIds.add(segmentMatcher.group(1).toInteger())
+ cardinality += versionMatcher.group(2).toLong()
+ }
+ }
+ }
+ return [segmentIds: segmentIds.unique(), cardinality: cardinality]
+ }
+
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ try {
+ updateBeConfig("compaction_batch_size", "512")
+ updateBeConfig("doris_scanner_row_bytes", "1")
+ updateBeConfig("enable_rowid_conversion_correctness_check", "true")
+ updateBeConfig("inverted_index_compaction_enable", "true")
+ updateBeConfig("vertical_compaction_max_segment_size", "8192")
+ updateBeConfig("enable_cloud_random_segment_id", "true")
Review Comment:
[P2] Restore this backend config after the suite
`enable_cloud_random_segment_id` is not included in `configNames`, so its
original value is never captured and `resetBeConfigs()` in the `finally` block
cannot restore it. A run that starts with this disabled is left enabled for
later suites. Please add it to the snapshotted config list before changing it
here.
##########
be/src/storage/compaction/compaction.cpp:
##########
@@ -1962,7 +1976,28 @@ Status
CloudCompactionMixin::construct_output_rowset_writer(RowsetWriterContext&
ctx.tablet = _tablet;
ctx.job_id = _uuid;
+ if (!_is_vertical) {
+ DBUG_EXECUTE_IF(
+
"CloudCompactionMixin.construct_output_rowset_writer.max_rows_per_segment", {
+ ctx.max_rows_per_segment =
+ dp->param<uint32_t>("max_rows_per_segment",
ctx.max_rows_per_segment);
+ DORIS_CHECK_GT(ctx.max_rows_per_segment, 0);
+ });
+ }
_output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(ctx,
_is_vertical));
+ if (config::enable_cloud_random_segment_id) {
Review Comment:
[P1] Update the encryption action before producing nonzero IDs
Both helpers in `be/src/service/http/action/check_encryption_action.cpp`
still call `rs->segment_path(0)` (lines 81 and 132). Once this block assigns a
start above zero, a compacted rowset has no `_0.dat`; the Cloud-enabled
`/api/check_tablet_encryption` action then fails `open_file` and returns HTTP
500 (including `get_footer=true` when that rowset is newest). Please select the
first segment by position, e.g. `rs->segment(0).path()`, in both helpers and
cover a rowset whose only physical ID is nonzero.
--
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]
