This is an automated email from the ASF dual-hosted git repository.
dataroaring 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 ca579c10a5e [Fix](full compaction) Full compaction should not do
ordered data compaction (#44359)
ca579c10a5e is described below
commit ca579c10a5ed8da4cedf2064a3c2e10056c24316
Author: abmdocrt <lianyuk...@selectdb.com>
AuthorDate: Thu Nov 21 23:23:42 2024 +0800
[Fix](full compaction) Full compaction should not do ordered data
compaction (#44359)
Problem: For a duplicate table with the following distribution, if it
has already completed cumulative compaction and then undergoes full
compaction, it will cause a BE core issue. Check failed: new_point ==
Tablet::K_INVALID_CUMULATIVE_POINT || new_point >= _cumulative_point
Unexpected cumulative point: 1087, origin: 2801.
"rowsets": [
"[0-386] 0 DATA NONOVERLAPPING
02000000000198aabe4290f2b0f5f35610c08a233a061892 0",
"[387-387] 0 DELETE OVERLAP_UNKNOWN
0200000000541310ac4d76e7580a708a2823a4d7a4f06090 0",
"[388-388] 0 DELETE OVERLAP_UNKNOWN
0200000000541d76ac4d76e7580a708a2823a4d7a4f06090 0",
"[389-389] 0 DELETE OVERLAP_UNKNOWN
0200000000543b4dac4d76e7580a708a2823a4d7a4f06090 0",
"[390-390] 0 DELETE OVERLAP_UNKNOWN
02000000005453aeac4d76e7580a708a2823a4d7a4f06090 0",
"[391-391] 0 DELETE OVERLAP_UNKNOWN
0200000000546a44ac4d76e7580a708a2823a4d7a4f06090 0",
"[392-392] 0 DELETE OVERLAP_UNKNOWN
02000000005480dbac4d76e7580a708a2823a4d7a4f06090 0",
"[393-393] 0 DELETE OVERLAP_UNKNOWN
0200000000548cb3ac4d76e7580a708a2823a4d7a4f06090 0",
"[394-394] 0 DELETE OVERLAP_UNKNOWN
0200000000549a25ac4d76e7580a708a2823a4d7a4f06090 0",
"[395-395] 0 DELETE OVERLAP_UNKNOWN
020000000054b359ac4d76e7580a708a2823a4d7a4f06090 0",
"[396-396] 0 DELETE OVERLAP_UNKNOWN
020000000054c19dac4d76e7580a708a2823a4d7a4f06090 0",
"[397-397] 0 DELETE OVERLAP_UNKNOWN
020000000054d757ac4d76e7580a708a2823a4d7a4f06090 0",
...
"[1085-1085] 0 DELETE OVERLAP_UNKNOWN
02000000002a0b20bd4798638f237008ff42fbca276b52a2 0",
"[1087-1506] 1 DATA NONOVERLAPPING
020000000000047e3b452de14ceaad2e78a87526026d2290 326.10 KB",
"[1087-1506] 1 DATA NONOVERLAPPING
020000000000047e3b452de14ceaad2e78a87526026d2290 326.10 KB",
...
"[2800-2800] 0 DELETE OVERLAP_UNKNOWN
02000000002f12d6bd4798638f237008ff42fbca276b52a2 0"
Reason: The duplicate table will go through ordered data compaction. Due
to the special distribution of the table, the input rowset will be cut
by the ordered data compaction, resulting in the full compaction only
being performed on a part of the rowsets.
Solution: For full compaction, prohibit ordered data compaction.
---
be/src/olap/compaction.cpp | 6 +-
be/src/olap/full_compaction.cpp | 3 +
.../test_full_compaction_with_ordered_data.groovy | 208 +++++++++++++++++++++
3 files changed, 215 insertions(+), 2 deletions(-)
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index 738087a702f..d7073491320 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -44,6 +44,7 @@
#include "io/fs/file_system.h"
#include "io/fs/file_writer.h"
#include "io/fs/remote_file_system.h"
+#include "io/io_common.h"
#include "olap/cumulative_compaction_policy.h"
#include "olap/cumulative_compaction_time_series_policy.h"
#include "olap/data_dir.h"
@@ -345,8 +346,9 @@ bool CompactionMixin::handle_ordered_data_compaction() {
if (!config::enable_ordered_data_compaction) {
return false;
}
- if (compaction_type() == ReaderType::READER_COLD_DATA_COMPACTION) {
- // The remote file system does not support to link files.
+ if (compaction_type() == ReaderType::READER_COLD_DATA_COMPACTION ||
+ compaction_type() == ReaderType::READER_FULL_COMPACTION) {
+ // The remote file system and full compaction does not support to link
files.
return false;
}
if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
diff --git a/be/src/olap/full_compaction.cpp b/be/src/olap/full_compaction.cpp
index 9d675f73192..529efa2e069 100644
--- a/be/src/olap/full_compaction.cpp
+++ b/be/src/olap/full_compaction.cpp
@@ -59,6 +59,9 @@ Status FullCompaction::prepare_compact() {
std::unique_lock cumu_lock(tablet()->get_cumulative_compaction_lock());
tablet()->set_is_full_compaction_running(true);
+ DBUG_EXECUTE_IF("FullCompaction.prepare_compact.set_cumu_point",
+ {
tablet()->set_cumulative_layer_point(tablet()->max_version_unlocked() + 1); })
+
// 1. pick rowsets to compact
RETURN_IF_ERROR(pick_rowsets_to_compact());
diff --git
a/regression-test/suites/fault_injection_p0/test_full_compaction_with_ordered_data.groovy
b/regression-test/suites/fault_injection_p0/test_full_compaction_with_ordered_data.groovy
new file mode 100644
index 00000000000..c6dfa6b885c
--- /dev/null
+++
b/regression-test/suites/fault_injection_p0/test_full_compaction_with_ordered_data.groovy
@@ -0,0 +1,208 @@
+// 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.codehaus.groovy.runtime.IOGroovyMethods
+
+suite("test_full_compaction_with_ordered_data","nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+ def tableName = "test_full_compaction_with_ordered_data"
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+
+ String backend_id;
+
+ def backendId_to_backendIP = [:]
+ def backendId_to_backendHttpPort = [:]
+ getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+ backend_id = backendId_to_backendIP.keySet()[0]
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `k` int ,
+ `v` int ,
+ ) engine=olap
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k)
+ BUCKETS 3
+ properties(
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true")
+ """
+ sql """ INSERT INTO ${tableName} VALUES (0,0),(1,1),(2,2)"""
+ sql """ delete from ${tableName} where k=0"""
+ sql """ delete from ${tableName} where k=1"""
+ sql """ delete from ${tableName} where k=2"""
+
+ def exception = false;
+ try {
+ def tablets = sql_return_maparray """ show tablets from ${tableName};
"""
+
+ def replicaNum = get_table_replica_num(tableName)
+ logger.info("get table replica num: " + replicaNum)
+ // before full compaction, there are 12 rowsets.
+ int rowsetCount = 0
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ (code, out, err) = curl("GET", tablet.CompactionStatus)
+ logger.info("Show tablets status: code=" + code + ", out=" + out +
", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 5 * replicaNum * 3)
+
+ // trigger full compactions for all tablets in ${tableName}
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ backend_id = tablet.BackendId
+ times = 1
+
+ do{
+ (code, out, err) =
be_run_full_compaction(backendId_to_backendIP.get(backend_id),
backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Run compaction: code=" + code + ", out=" + out +
", err=" + err)
+ ++times
+ sleep(2000)
+ } while (parseJson(out.trim()).status.toLowerCase()!="success" &&
times<=10)
+
+ }
+
+ // wait for full compaction done
+ for (def tablet in tablets) {
+ boolean running = true
+ do {
+ Thread.sleep(1000)
+ String tablet_id = tablet.TabletId
+ backend_id = tablet.BackendId
+ (code, out, err) =
be_get_compaction_status(backendId_to_backendIP.get(backend_id),
backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
+ assertEquals(code, 0)
+ def compactionStatus = parseJson(out.trim())
+ assertEquals("success", compactionStatus.status.toLowerCase())
+ running = compactionStatus.run_status
+ } while (running)
+ }
+
+ // after full compaction, there is only 1 rowset.
+
+ rowsetCount = 0
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ (code, out, err) = curl("GET", tablet.CompactionStatus)
+ logger.info("Show tablets status: code=" + code + ", out=" + out +
", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 1 * replicaNum * 3)
+ } catch (Exception e) {
+ logger.info(e.getMessage())
+ exception = true;
+ } finally {
+ assertFalse(exception)
+ }
+
+ sql """ delete from ${tableName} where k=0"""
+ sql """ delete from ${tableName} where k=1"""
+ sql """ delete from ${tableName} where k=2"""
+ sql """ delete from ${tableName} where k=3"""
+ sql """ delete from ${tableName} where k=4"""
+ sql """ delete from ${tableName} where k=5"""
+ sql """ delete from ${tableName} where k=6"""
+ sql """ delete from ${tableName} where k=7"""
+ sql """ delete from ${tableName} where k=8"""
+ sql """ delete from ${tableName} where k=9"""
+ sql """ INSERT INTO ${tableName} VALUES (10,10)"""
+
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ exception = false;
+ try {
+
GetDebugPoint().enableDebugPointForAllBEs("FullCompaction.prepare_compact.set_cumu_point")
+ def tablets = sql_return_maparray """ show tablets from ${tableName};
"""
+
+ def replicaNum = get_table_replica_num(tableName)
+ logger.info("get table replica num: " + replicaNum)
+ // before full compaction, there are 12 rowsets.
+ int rowsetCount = 0
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ (code, out, err) = curl("GET", tablet.CompactionStatus)
+ logger.info("Show tablets status: code=" + code + ", out=" + out +
", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 12 * replicaNum * 3)
+
+ // trigger full compactions for all tablets in ${tableName}
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ backend_id = tablet.BackendId
+ times = 1
+
+ do{
+ (code, out, err) =
be_run_full_compaction(backendId_to_backendIP.get(backend_id),
backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Run compaction: code=" + code + ", out=" + out +
", err=" + err)
+ ++times
+ sleep(2000)
+ } while (parseJson(out.trim()).status.toLowerCase()!="success" &&
times<=10)
+
+ }
+
+ // wait for full compaction done
+ for (def tablet in tablets) {
+ boolean running = true
+ do {
+ Thread.sleep(1000)
+ String tablet_id = tablet.TabletId
+ backend_id = tablet.BackendId
+ (code, out, err) =
be_get_compaction_status(backendId_to_backendIP.get(backend_id),
backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Get compaction status: code=" + code + ", out=" +
out + ", err=" + err)
+ assertEquals(code, 0)
+ def compactionStatus = parseJson(out.trim())
+ assertEquals("success", compactionStatus.status.toLowerCase())
+ running = compactionStatus.run_status
+ } while (running)
+ }
+
+ // after full compaction, there is only 1 rowset.
+
+ rowsetCount = 0
+ for (def tablet in tablets) {
+ String tablet_id = tablet.TabletId
+ (code, out, err) = curl("GET", tablet.CompactionStatus)
+ logger.info("Show tablets status: code=" + code + ", out=" + out +
", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 1 * replicaNum * 3)
+ } catch (Exception e) {
+ logger.info(e.getMessage())
+ exception = true;
+ } finally {
+
GetDebugPoint().disableDebugPointForAllBEs("FullCompaction.prepare_compact.set_cumu_point")
+ assertFalse(exception)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
