Skip to content

Commit 72ceac7

Browse files
GroundWuGroundWu
authored
Merge pull request #155 from GroundWu/test_for_group_commit_view
add test case for group commit sys view
2 parents 6863cc7 + 51021ea commit 72ceac7

File tree

1 file changed

+203
-0
lines changed

1 file changed

+203
-0
lines changed

‎src/test/java/com/alipay/oceanbase/rpc/ObTableGroupCommitTest.java‎

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
/*-
2+
* #%L
3+
* OBKV Table Client Framework
4+
* %%
5+
* Copyright (C) 2023 OceanBase
6+
* %%
7+
* OBKV Table Client Framework is licensed under Mulan PSL v2.
8+
* You can use this software according to the terms and conditions of the Mulan PSL v2.
9+
* You may obtain a copy of Mulan PSL v2 at:
10+
* http://license.coscl.org.cn/MulanPSL2
11+
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
12+
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
13+
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
14+
* See the Mulan PSL v2 for more details.
15+
* #L%
16+
*/
17+
18+
package com.alipay.oceanbase.rpc;
19+
20+
import com.alipay.oceanbase.rpc.property.Property;
21+
import com.alipay.oceanbase.rpc.util.ObTableClientTestUtil;
22+
import org.junit.Assert;
23+
import org.junit.Assume;
24+
import org.junit.Before;
25+
import org.junit.Test;
26+
27+
import java.sql.Connection;
28+
import java.sql.ResultSet;
29+
import java.sql.SQLException;
30+
import java.sql.Statement;
31+
import java.util.ArrayList;
32+
33+
import static com.alipay.oceanbase.rpc.mutation.MutationFactory.colVal;
34+
import static com.alipay.oceanbase.rpc.mutation.MutationFactory.row;
35+
import static org.junit.Assert.assertEquals;
36+
37+
public class ObTableGroupCommitTest {
38+
public ObTableClient client;
39+
40+
@Before
41+
public void setup() throws Exception {
42+
System.setProperty("ob_table_min_rslist_refresh_interval_millis", "1");
43+
44+
final ObTableClient obTableClient = ObTableClientTestUtil.newTestClient();
45+
obTableClient.init();
46+
47+
this.client = obTableClient;
48+
}
49+
50+
/**
51+
CREATE TABLE IF NOT EXISTS `test_varchar_table` (
52+
`c1` varchar(20) NOT NULL,
53+
`c2` varchar(20) DEFAULT NULL,
54+
PRIMARY KEY (`c1`)
55+
);
56+
**/
57+
@Test
58+
public void test_group_commit() throws Exception {
59+
Assume.assumeTrue("support after ob version 4.2.5", ObGlobal.OB_VERSION >= ObGlobal.calcVersion(4,(short)2,(byte)5,(byte)0));
60+
String tableName = "test_varchar_table";
61+
long executeTime = 10000; // 10s
62+
int threadCnt = 100;
63+
ArrayList<Thread> workers = new ArrayList<>();
64+
deleteTable(tableName);
65+
switchGroupCommit(true);
66+
for (int i = 0; i < threadCnt; i++) {
67+
InsertWorker worker = new InsertWorker(i, tableName, this.client, executeTime);
68+
workers.add(worker);
69+
}
70+
71+
for (int i = 0; i < threadCnt; i++) {
72+
GetWorker worker = new GetWorker(i, tableName, this.client, executeTime);
73+
workers.add(worker);
74+
}
75+
76+
for (int i = 0; i < 2*threadCnt; i++) {
77+
workers.get(i).start();
78+
}
79+
80+
for (int i = 0; i < 2*threadCnt; i++) {
81+
workers.get(i).join();
82+
}
83+
checkSystemView();
84+
deleteTable(tableName);
85+
switchGroupCommit(false);
86+
this.client.close();
87+
}
88+
89+
private void checkSystemView() throws SQLException {
90+
// mysql tenant
91+
Connection mysql_conn = ObTableClientTestUtil.getConnection();
92+
Statement statement = mysql_conn.createStatement();
93+
statement.execute("select b.tenant_id as tenant_id, b.tenant_name as tenant_name, a.group_type as group_type, a.batch_size as batch_size " +
94+
" from oceanbase.GV$OB_KV_GROUP_COMMIT_STATUS a inner join " +
95+
"oceanbase.DBA_OB_TENANTS b on a.tenant_id = b.tenant_id group by a.group_type");
96+
ResultSet resultSet = statement.getResultSet();
97+
int resCount = 0;
98+
System.out.println("visit by mysql tenant:");
99+
while (resultSet.next()) {
100+
long tenant_id = resultSet.getLong("tenant_id");
101+
String tenant_name = resultSet.getString("tenant_name");
102+
String group_type = resultSet.getString("group_type");
103+
long batch_size = resultSet.getLong("batch_size");
104+
System.out.println("tenant_id:" + tenant_id+", tenant_name: "+ tenant_name +", group_type: "+group_type+", batch_size: "+batch_size);
105+
resCount++;
106+
}
107+
Assert.assertTrue(resCount >= 3);
108+
mysql_conn.close();
109+
110+
// sys tenant
111+
Connection sys_conn = ObTableClientTestUtil.getSysConnection();
112+
Statement statement2 = sys_conn.createStatement();
113+
statement2.execute("select b.tenant_id as tenant_id, b.tenant_name as tenant_name, a.group_type as group_type, a.batch_size as batch_size " +
114+
" from oceanbase.GV$OB_KV_GROUP_COMMIT_STATUS a inner join " +
115+
"oceanbase.__all_tenant b on a.tenant_id = b.tenant_id where b.tenant_name in ('sys', '"+ObTableClientTestUtil.getTenantName()+"') group by b.tenant_name, a.group_type;");
116+
resultSet = statement2.getResultSet();
117+
resCount = 0;
118+
System.out.println("visit by sys tenant:");
119+
while (resultSet.next()) {
120+
long tenant_id = resultSet.getLong("tenant_id");
121+
String tenant_name = resultSet.getString("tenant_name");
122+
String group_type = resultSet.getString("group_type");
123+
long batch_size = resultSet.getLong("batch_size");
124+
System.out.println("tenant_id:" + tenant_id+", tenant_name: "+ tenant_name +", group_type: "+group_type+", batch_size: "+batch_size);
125+
resCount++;
126+
}
127+
Assert.assertTrue(resCount >= 4);
128+
sys_conn.close();
129+
}
130+
131+
class InsertWorker extends Thread {
132+
private int id;
133+
private String tableName;
134+
private ObTableClient obTableClient;
135+
private long executeTime; // in millisecond
136+
137+
public InsertWorker(int id, String tableName, ObTableClient obTableClient, long executeTime) {
138+
this.id = id;
139+
this.tableName = tableName;
140+
this.obTableClient = obTableClient;
141+
this.executeTime = executeTime;
142+
}
143+
144+
public void run() {
145+
long start = System.currentTimeMillis();
146+
int counter = 0;
147+
while ((System.currentTimeMillis() - start) < executeTime) {
148+
try {
149+
String c1 = String.format("rk_%d_%d", id, counter);
150+
String c2 = String.format("col_%d_%d", id, counter);
151+
obTableClient.insert(tableName).setRowKey(row(colVal("c1", c1)))
152+
.addMutateRow(row(colVal("c2",c2))).execute();
153+
counter++;
154+
} catch (Exception e) {
155+
e.printStackTrace();
156+
System.out.println("thread " + id + " get occurs exception !");
157+
}
158+
}
159+
}
160+
}
161+
162+
class GetWorker extends Thread {
163+
private int id;
164+
private String tableName;
165+
private ObTableClient obTableClient;
166+
private long executeTime; // in millisecond
167+
168+
public GetWorker(int id, String tableName, ObTableClient obTableClient, long executeTime) {
169+
this.id = id;
170+
this.tableName = tableName;
171+
this.obTableClient = obTableClient;
172+
this.executeTime = executeTime;
173+
}
174+
175+
public void run() {
176+
long start = System.currentTimeMillis();
177+
int counter = 0;
178+
while ((System.currentTimeMillis() - start) < executeTime) {
179+
try {
180+
String c1 = String.format("rk_%d_%d", id, counter);
181+
obTableClient.get(tableName, new String[] { "c1" }, new String[] { c1 });
182+
counter++;
183+
} catch (Exception e) {
184+
e.printStackTrace();
185+
System.out.println("thread " + id + " get occurs exception !");
186+
}
187+
}
188+
}
189+
}
190+
191+
private void switchGroupCommit(boolean is_enable) throws SQLException {
192+
int batch_size = is_enable ? 10 : 1;
193+
Connection mysql_conn = ObTableClientTestUtil.getConnection();
194+
Statement statement = mysql_conn.createStatement();
195+
statement.execute("alter system set kv_group_commit_batch_size = "+ batch_size);
196+
}
197+
198+
private void deleteTable(String tableName) throws SQLException {
199+
Connection mysql_conn = ObTableClientTestUtil.getConnection();
200+
Statement statement = mysql_conn.createStatement();
201+
statement.execute("delete from "+ tableName);
202+
}
203+
}

0 commit comments

Comments
 (0)