Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix 2.0 insert overwrite #114

Merged
merged 4 commits into from
Jun 13, 2024
Merged

fix 2.0 insert overwrite #114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
27f82d2
fix 2.0 insert overwrite
lsy3993 Jun 13, 2024
1c33216
add not null
lsy3993 Jun 13, 2024
aa6e823
Update regression-test/suites/table-sync/test_insert_overwrite.groovy
w41ter Jun 13, 2024
e9c1174
Update regression-test/suites/table-sync/test_add_partition.groovy
w41ter Jun 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 11 additions & 10 deletions regression-test/suites/table-sync/test_add_partition.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,8 @@ suite("test_add_partition") {
sql """
CREATE TABLE if NOT EXISTS ${tableName}
(
`test` INT,
`id` INT
`test` INT NOT NULL,
`id` INT NOT NULL
)
ENGINE=OLAP
UNIQUE KEY(`test`, `id`)
Expand Down Expand Up @@ -261,13 +261,14 @@ suite("test_add_partition") {

assertTrue(checkRestoreFinishTimesOf("${tableName}", 60))

sql """
INSERT OVERWRITE TABLE ${tableName} VALUES (1, 100);
"""
// INSERT OVERWRITE is not supported by branch-2.0 table sync.
// sql """
w41ter marked this conversation as resolved.
Show resolved Hide resolved
// INSERT OVERWRITE TABLE ${tableName} VALUES (1, 100);
// """

assertTrue(checkShowTimesOf("""
SELECT * FROM ${tableName}
WHERE id = 100
""",
exist, 60, "target"))
// assertTrue(checkShowTimesOf("""
// SELECT * FROM ${tableName}
// WHERE id = 100
// """,
// exist, 60, "target"))
}
319 changes: 161 additions & 158 deletions regression-test/suites/table-sync/test_insert_overwrite.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,162 +24,165 @@ suite("test_insert_overwrite") {
// 1. create temp partitions
// 2. insert int temp partitions
// 3. replace overlap partitions
def tableName = "tbl_insert_overwrite_" + UUID.randomUUID().toString().replace("-", "")
def uniqueTable = "${tableName}_unique"
def syncerAddress = "127.0.0.1:9190"
def test_num = 0
def insert_num = 5
def sync_gap_time = 5000
String response

def checkSelectTimesOf = { sqlString, rowSize, times -> Boolean
def tmpRes = target_sql "${sqlString}"
while (tmpRes.size() != rowSize) {
sleep(sync_gap_time)
if (--times > 0) {
tmpRes = target_sql "${sqlString}"
} else {
break
}
}
return tmpRes.size() == rowSize
}

def checkRestoreFinishTimesOf = { checkTable, times -> Boolean
Boolean ret = false
while (times > 0) {
def sqlInfo = target_sql "SHOW RESTORE FROM TEST_${context.dbName}"
for (List<Object> row : sqlInfo) {
if ((row[10] as String).contains(checkTable)) {
ret = row[4] == "FINISHED"
}
}

if (ret) {
break
} else if (--times > 0) {
sleep(sync_gap_time)
}
}

return ret
}

def checkBackupFinishTimesOf = { checkTable, times -> Boolean
Boolean ret = false
while (times > 0) {
def sqlInfo = sql "SHOW BACKUP FROM ${context.dbName}"
for (List<Object> row : sqlInfo) {
if ((row[4] as String).contains(checkTable)) {
ret = row[3] == "FINISHED"
}
}

if (ret) {
break
} else if (--times > 0) {
sleep(sync_gap_time)
}
}

return ret
}

def checkData = { data, beginCol, value -> Boolean
if (data.size() < beginCol + value.size()) {
return false
}

for (int i = 0; i < value.size(); ++i) {
if ((data[beginCol + i] as int) != value[i]) {
return false
}
}

return true
}

sql """
CREATE TABLE if NOT EXISTS ${uniqueTable}
(
`test` INT,
`id` INT
)
ENGINE=OLAP
UNIQUE KEY(`test`, `id`)
PARTITION BY RANGE(id)
(
PARTITION `p1` VALUES LESS THAN ("100"),
PARTITION `p2` VALUES LESS THAN ("200")
)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1",
"binlog.enable" = "true",
"binlog.ttl_seconds" = "180"
)
"""

sql """
INSERT INTO ${uniqueTable} VALUES
(1, 0),
(1, 1),
(1, 2),
(1, 3),
(1, 4)
"""
sql "sync"

// test 1: target cluster follow source cluster
logger.info("=== Test 1: backup/restore case ===")
httpTest {
uri "/create_ccr"
endpoint syncerAddress
def bodyJson = get_ccr_body "${uniqueTable}"
body "${bodyJson}"
op "post"
result response
}
assertTrue(checkRestoreFinishTimesOf("${uniqueTable}", 60))
assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id", 5, 60))
qt_sql "SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id"
qt_target_sql "SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id"

logger.info("=== Test 2: dest cluster follow source cluster case ===")

sql """
INSERT INTO ${uniqueTable} VALUES
(2, 0),
(2, 1),
(2, 2),
(2, 3),
(2, 4)
"""
sql "sync"
assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test=2", 5, 60))
qt_sql "SELECT * FROM ${uniqueTable} WHERE test=2 ORDER BY id"
qt_target_sql "SELECT * FROM ${uniqueTable} WHERE test=2 ORDER BY id"

logger.info("=== Test 3: insert overwrite source table ===")

sql """
INSERT OVERWRITE TABLE ${uniqueTable} VALUES
(3, 0),
(3, 1),
(3, 2),
(3, 3),
(3, 4)
"""
sql "sync"

sleep(10000)
assertTrue(checkBackupFinishTimesOf("${uniqueTable}", 60))
sleep(10000)
assertTrue(checkRestoreFinishTimesOf("${uniqueTable}", 60))

assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test=3", 5, 60))
assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable}", 5, 60))

qt_sql "SELECT * FROM ${uniqueTable} ORDER BY test, id"
qt_target_sql "SELECT * FROM ${uniqueTable} ORDER BY test, id"


// INSERT OVERWRITE is not supported by branch-2.0 table sync.
// def tableName = "tbl_insert_overwrite_" + UUID.randomUUID().toString().replace("-", "")
w41ter marked this conversation as resolved.
Show resolved Hide resolved
// def uniqueTable = "${tableName}_unique"
// def syncerAddress = "127.0.0.1:9190"
// def test_num = 0
// def insert_num = 5
// def sync_gap_time = 5000
// String response

// def checkSelectTimesOf = { sqlString, rowSize, times -> Boolean
// def tmpRes = target_sql "${sqlString}"
// while (tmpRes.size() != rowSize) {
// sleep(sync_gap_time)
// if (--times > 0) {
// tmpRes = target_sql "${sqlString}"
// } else {
// break
// }
// }
// return tmpRes.size() == rowSize
// }

// def checkRestoreFinishTimesOf = { checkTable, times -> Boolean
// Boolean ret = false
// while (times > 0) {
// def sqlInfo = target_sql "SHOW RESTORE FROM TEST_${context.dbName}"
// for (List<Object> row : sqlInfo) {
// if ((row[10] as String).contains(checkTable)) {
// ret = row[4] == "FINISHED"
// }
// }

// if (ret) {
// break
// } else if (--times > 0) {
// sleep(sync_gap_time)
// }
// }

// return ret
// }

// def checkBackupFinishTimesOf = { checkTable, times -> Boolean
// Boolean ret = false
// while (times > 0) {
// def sqlInfo = sql "SHOW BACKUP FROM ${context.dbName}"
// for (List<Object> row : sqlInfo) {
// if ((row[4] as String).contains(checkTable)) {
// ret = row[3] == "FINISHED"
// }
// }

// if (ret) {
// break
// } else if (--times > 0) {
// sleep(sync_gap_time)
// }
// }

// return ret
// }

// def checkData = { data, beginCol, value -> Boolean
// if (data.size() < beginCol + value.size()) {
// return false
// }

// for (int i = 0; i < value.size(); ++i) {
// if ((data[beginCol + i] as int) != value[i]) {
// return false
// }
// }

// return true
// }

// sql """
// CREATE TABLE if NOT EXISTS ${uniqueTable}
// (
// `test` INT,
// `id` INT
// )
// ENGINE=OLAP
// UNIQUE KEY(`test`, `id`)
// PARTITION BY RANGE(id)
// (
// PARTITION `p1` VALUES LESS THAN ("100"),
// PARTITION `p2` VALUES LESS THAN ("200")
// )
// DISTRIBUTED BY HASH(id) BUCKETS 1
// PROPERTIES (
// "replication_allocation" = "tag.location.default: 1",
// "binlog.enable" = "true",
// "binlog.ttl_seconds" = "180"
// )
// """

// sql """
// INSERT INTO ${uniqueTable} VALUES
// (1, 0),
// (1, 1),
// (1, 2),
// (1, 3),
// (1, 4)
// """
// sql "sync"

// // test 1: target cluster follow source cluster
// logger.info("=== Test 1: backup/restore case ===")
// httpTest {
// uri "/create_ccr"
// endpoint syncerAddress
// def bodyJson = get_ccr_body "${uniqueTable}"
// body "${bodyJson}"
// op "post"
// result response
// }
// assertTrue(checkRestoreFinishTimesOf("${uniqueTable}", 60))
// assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id", 5, 60))
// qt_sql "SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id"
// qt_target_sql "SELECT * FROM ${uniqueTable} WHERE test = 1 ORDER BY id"

// logger.info("=== Test 2: dest cluster follow source cluster case ===")

// sql """
// INSERT INTO ${uniqueTable} VALUES
// (2, 0),
// (2, 1),
// (2, 2),
// (2, 3),
// (2, 4)
// """
// sql "sync"
// assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test=2", 5, 60))
// qt_sql "SELECT * FROM ${uniqueTable} WHERE test=2 ORDER BY id"
// qt_target_sql "SELECT * FROM ${uniqueTable} WHERE test=2 ORDER BY id"

// logger.info("=== Test 3: insert overwrite source table ===")

// sql """
// INSERT OVERWRITE TABLE ${uniqueTable} VALUES
// (3, 0),
// (3, 1),
// (3, 2),
// (3, 3),
// (3, 4)
// """
// sql "sync"

// sleep(10000)
// assertTrue(checkBackupFinishTimesOf("${uniqueTable}", 60))
// sleep(10000)
// assertTrue(checkRestoreFinishTimesOf("${uniqueTable}", 60))

// assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable} WHERE test=3", 5, 60))
// assertTrue(checkSelectTimesOf("SELECT * FROM ${uniqueTable}", 5, 60))

// qt_sql "SELECT * FROM ${uniqueTable} ORDER BY test, id"
// qt_target_sql "SELECT * FROM ${uniqueTable} ORDER BY test, id"
}
Loading