Skip to content

Commit

Permalink
add inject points and skip rowsets case
Browse files Browse the repository at this point in the history
  • Loading branch information
bobhan1 committed Jul 30, 2024
1 parent fcc24a4 commit 6e4f5dd
Show file tree
Hide file tree
Showing 8 changed files with 184 additions and 0 deletions.
6 changes: 6 additions & 0 deletions be/src/olap/txn_manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -402,6 +402,12 @@ Status TxnManager::publish_txn(OlapMeta* meta, TPartitionId partition_id,
TTransactionId transaction_id, TTabletId tablet_id,
TabletUid tablet_uid, const Version& version,
TabletPublishStatistics* stats) {
DBUG_EXECUTE_IF("TxnManager::publish_txn.enable_spin_wait", {
auto block_tablet_id = dp->param<int64_t>("tablet_id");
if (tablet_id == block_tablet_id) {
DBUG_EXECUTE_IF("TxnManager::publish_txn.block_publish_txn", DBUG_BLOCK);
}
});
auto tablet = _engine.tablet_manager()->get_tablet(tablet_id);
if (tablet == nullptr) {
return Status::OK();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3

-- !sql --
1 999 999 1 1
2 888 888 2 2
3 777 777 3 3

Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3

-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3

Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
// 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 java.util.Date
import java.text.SimpleDateFormat
import org.apache.http.HttpResponse
import org.apache.http.client.methods.HttpPut
import org.apache.http.impl.client.CloseableHttpClient
import org.apache.http.impl.client.HttpClients
import org.apache.http.entity.ContentType
import org.apache.http.entity.StringEntity
import org.apache.http.client.config.RequestConfig
import org.apache.http.client.RedirectStrategy
import org.apache.http.protocol.HttpContext
import org.apache.http.HttpRequest
import org.apache.http.impl.client.LaxRedirectStrategy
import org.apache.http.client.methods.RequestBuilder
import org.apache.http.entity.StringEntity
import org.apache.http.client.methods.CloseableHttpResponse
import org.apache.http.util.EntityUtils
import org.apache.doris.regression.suite.ClusterOptions

suite("test_partial_update_skip_compaction", "nonConcurrent") {

def table1 = "test_partial_update_skip_compaction"
sql "DROP TABLE IF EXISTS ${table1};"
sql """ CREATE TABLE IF NOT EXISTS ${table1} (
`k1` int NOT NULL,
`c1` int,
`c2` int,
`c3` int,
`c4` int
)UNIQUE KEY(k1)
DISTRIBUTED BY HASH(k1) BUCKETS 1
PROPERTIES (
"disable_auto_compaction" = "true",
"replication_num" = "1"); """

sql "insert into ${table1} values(1,1,1,1,1);"
sql "insert into ${table1} values(2,2,2,2,2);"
sql "insert into ${table1} values(3,3,3,3,3);"
sql "sync;"
order_qt_sql "select * from ${table1};"

def beNodes = sql_return_maparray("show backends;")
def tabletStat = sql_return_maparray("show tablets from ${table1};").get(0)
def tabletBackendId = tabletStat.BackendId
def tabletId = tabletStat.TabletId
def tabletBackend;
for (def be : beNodes) {
if (be.BackendId == tabletBackendId) {
tabletBackend = be
break;
}
}
logger.info("tablet ${tabletId} on backend ${tabletBackend.Host} with backendId=${tabletBackend.BackendId}");

def check_rs_metas = { expected_rs_meta_size, check_func ->
def metaUrl = sql_return_maparray("show tablets from ${table1};").get(0).MetaUrl
def command = "curl ${metaUrl}"
log.info("get_rs_metas execute command: ${command}")
def process = command.execute()
code = process.waitFor()
out = process.text
def jsonMeta = parseJson(out)
assertEquals(code, 0)

assertEquals(jsonMeta.rs_metas.size(), expected_rs_meta_size)
for (def meta : jsonMeta.rs_metas) {
int startVersion = meta.start_version.toInteger()
int endVersion = meta.end_version.toInteger()
int numSegments = meta.num_segments.toInteger()
int numRows = meta.num_rows.toInteger()
String overlapPb = meta.segments_overlap_pb.toString()
logger.info("[${startVersion}-${endVersion}] ${overlapPb} ${meta.num_segments} ${numRows} ${meta.rowset_id_v2}")
check_func(startVersion, endVersion, numSegments, numRows, overlapPb)
}
}

check_rs_metas(4, {int startVersion, int endVersion, int numSegments, int numRows, String overlapPb ->
if (startVersion == 0) {
assertEquals(endVersion, 1)
assertEquals(numSegments, 0)
} else {
assertEquals(startVersion, endVersion)
assertEquals(numSegments, 1)
}
})

try {
GetDebugPoint().clearDebugPointsForAllBEs()

// block the partial update before publish phase
GetDebugPoint().enableDebugPointForAllBEs("TxnManager::publish_txn.enable_spin_wait", [tablet_id: "${tabletId}"])
GetDebugPoint().enableDebugPointForAllBEs("TxnManager::publish_txn.block_publish_txn")

def t1 = Thread.start {
sql "set enable_unique_key_partial_update=true;"
sql "sync;"
sql "insert into ${table1}(k1,c1,c2) values(1,999,999),(2,888,888),(3,777,777);"
}

// trigger full compaction on tablet
logger.info("trigger compaction on another BE ${tabletBackend.Host} with backendId=${tabletBackend.BackendId}")
def (code, out, err) = be_run_full_compaction(tabletBackend.Host, tabletBackend.HttpPort, tabletId)
logger.info("Run compaction: code=" + code + ", out=" + out + ", err=" + err)
assertEquals(code, 0)
def compactJson = parseJson(out.trim())
assertEquals("success", compactJson.status.toLowerCase())

// wait for full compaction to complete
Thread.sleep(1500)

check_rs_metas(1, {int startVersion, int endVersion, int numSegments, int numRows, String overlapPb ->
assertEquals(startVersion, 0)
assertEquals(endVersion, 4)
assertEquals(overlapPb, "NONOVERLAPPING")
})

GetDebugPoint().disableDebugPointForAllBEs("TxnManager::publish_txn.block_publish_txn")
t1.join()

order_qt_sql "select * from ${table1};"

check_rs_metas(2, {int startVersion, int endVersion, int numSegments, int numRows, String overlapPb ->
if (startVersion == 5) {
assertEquals(endVersion, 5)
// checks that partial update skips the alignment process of rowsets produced by compaction and
// doesn't generate new segment in publish phase
assertEquals(numSegments, 1)
}
})

} catch(Exception e) {
logger.info(e.getMessage())
throw e
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
}

sql "DROP TABLE IF EXISTS ${table1};"
}

0 comments on commit 6e4f5dd

Please sign in to comment.