Skip to content

Commit 4c0b2e7

Browse files
mikaelronstrommikaelronstrom
committed
RONDB-768: Various fixes to improved method, removed FROREIGN KEY
1 parent 0759d47 commit 4c0b2e7

8 files changed

+64
-20
lines changed

‎pink/rondis/common.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
#define REDIS_DB_NAME "redis"
1010

11-
#define FOREIGN_KEY_RESTRICT_ERROR 256
11+
#define RESTRICT_VALUE_ROWS_ERROR 6000
1212

1313
#define RONDB_INTERNAL_ERROR 2
1414
#define READ_ERROR 626

‎pink/rondis/sql/STRING_key.sql

+1-1
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ CREATE TABLE string_keys(
1515
num_rows INT UNSIGNED NOT NULL,
1616
value_start VARBINARY(26500) NOT NULL,
1717
-- Redis supports get/set of seconds/milliseconds
18-
expiry_date INT UNSIGNED,
18+
expiry_date INT UNSIGNED NOT NULL,
1919
-- Easier to sort and delete keys this way
2020
KEY expiry_index(expiry_date),
2121
PRIMARY KEY (redis_key_id, redis_key) USING HASH,

‎pink/rondis/sql/STRING_value.sql

+2-3
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ CREATE TABLE string_values(
22
rondb_key BIGINT UNSIGNED NOT NULL,
33
ordinal INT UNSIGNED NOT NULL,
44
value VARBINARY(29500) NOT NULL,
5-
PRIMARY KEY (rondb_key, ordinal),
6-
FOREIGN KEY (rondb_key) REFERENCES string_keys(rondb_key) ON UPDATE RESTRICT ON DELETE CASCADE
5+
PRIMARY KEY (rondb_key, ordinal)
76
) ENGINE NDB,
8-
COMMENT = "NDB_TABLE=PARTITION_BALANCE=RP_BY_LDM_X_8" PARTITION BY KEY (rondb_key);
7+
COMMENT = "NDB_TABLE=PARTITION_BALANCE=RP_BY_LDM_X_8";

‎pink/rondis/string/commands.cc

+8-1
Original file line numberDiff line numberDiff line change
@@ -207,7 +207,7 @@ void rondb_set(
207207
{
208208
// Often unnecessary since it already failed to commit
209209
ndb->closeTransaction(trans);
210-
if (ret_code != FOREIGN_KEY_RESTRICT_ERROR)
210+
if (ret_code != RESTRICT_VALUE_ROWS_ERROR)
211211
{
212212
return;
213213
}
@@ -243,6 +243,10 @@ void rondb_set(
243243
num_value_rows,
244244
prev_num_rows,
245245
Uint32(0));
246+
if (ret_code != 0) {
247+
ndb->closeTransaction(trans);
248+
return;
249+
}
246250
} else if (num_value_rows == 0) {
247251
ndb->closeTransaction(trans);
248252
response->append("+OK\r\n");
@@ -266,6 +270,7 @@ void rondb_set(
266270
&varsize_param[0]);
267271
}
268272
if (ret_code != 0) {
273+
ndb->closeTransaction(trans);
269274
return;
270275
}
271276
ret_code = delete_value_rows(response,
@@ -275,12 +280,14 @@ void rondb_set(
275280
num_value_rows,
276281
prev_num_rows);
277282
if (ret_code != 0) {
283+
ndb->closeTransaction(trans);
278284
return;
279285
}
280286
if (trans->execute(NdbTransaction::Commit,
281287
NdbOperation::AbortOnError) == 0 &&
282288
trans->getNdbError().code != 0)
283289
{
290+
ndb->closeTransaction(trans);
284291
assign_ndb_err_to_response(response,
285292
FAILED_EXEC_TXN,
286293
trans->getNdbError());

‎pink/rondis/string/db_operations.cc

+15-6
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,13 @@ int create_key_row(std::string *response,
4343
value_str,
4444
tot_value_len,
4545
num_value_rows,
46+
prev_num_rows,
4647
row_state,
4748
&recAttr);
4849
if (ret_code != 0) {
4950
return ret_code;
5051
}
51-
if (num_value_rows == 0 && prev_num_rows)
52+
if (num_value_rows == 0 && prev_num_rows == 0)
5253
{
5354
if (trans->execute(NdbTransaction::Commit,
5455
NdbOperation::AbortOnError) == 0 &&
@@ -68,7 +69,7 @@ int create_key_row(std::string *response,
6869
}
6970
}
7071

71-
if (trans->getNdbError().code != FOREIGN_KEY_RESTRICT_ERROR)
72+
if (trans->getNdbError().code != RESTRICT_VALUE_ROWS_ERROR)
7273
{
7374
assign_ndb_err_to_response(response,
7475
FAILED_EXEC_TXN,
@@ -88,23 +89,24 @@ int write_data_to_key_op(std::string *response,
8889
const char *value_str,
8990
Uint32 tot_value_len,
9091
Uint32 num_value_rows,
92+
Uint32 & prev_num_rows,
9193
Uint32 row_state,
9294
NdbRecAttr **recAttr) {
9395
struct key_table key_row;
94-
const Uint32 mask = 0xEF;
95-
const unsigned char *mask_ptr = (const unsigned char *)&mask;
96+
Uint32 mask = 0xFF;
9697
key_row.null_bits = 0;
9798
memcpy(&key_row.redis_key[2], key_str, key_len);
9899
set_length(&key_row.redis_key[0], key_len);
99100
key_row.redis_key_id = redis_key_id;
100101
if (rondb_key == 0)
101102
{
102-
key_row.null_bits = 1;
103+
mask = 0xFB;
103104
}
104105
else
105106
{
106107
key_row.rondb_key = rondb_key;
107108
}
109+
const unsigned char *mask_ptr = (const unsigned char *)&mask;
108110
key_row.tot_value_len = tot_value_len;
109111
key_row.num_rows = num_value_rows;
110112
key_row.value_data_type = row_state;
@@ -120,7 +122,14 @@ int write_data_to_key_op(std::string *response,
120122

121123
Uint32 code_buffer[64];
122124
NdbInterpretedCode code(tab, &code_buffer[0], sizeof(code_buffer));
123-
int ret_code = write_key_row(response, code, tab);
125+
int ret_code = 0;
126+
if (num_value_rows > 0 || prev_num_rows > 0) {
127+
ret_code = write_key_row_no_commit(response, code, tab);
128+
}
129+
else
130+
{
131+
ret_code = write_key_row_commit(response, code, tab);
132+
}
124133
if (ret_code != 0) {
125134
return ret_code;
126135
}

‎pink/rondis/string/db_operations.h

+1
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ int write_data_to_key_op(std::string *response,
3434
const char *value_str,
3535
Uint32 tot_value_len,
3636
Uint32 num_value_rows,
37+
Uint32 &prev_num_rows,
3738
Uint32 row_state,
3839
NdbRecAttr **recAttr);
3940

‎pink/rondis/string/interpreted_code.cc

+30-4
Original file line numberDiff line numberDiff line change
@@ -176,10 +176,9 @@ int write_hset_key_table(Ndb *ndb,
176176
return 0;
177177
}
178178

179-
180-
int write_key_row(std::string *response,
181-
NdbInterpretedCode &code,
182-
const NdbDictionary::Table *tab) {
179+
int write_key_row_no_commit(std::string *response,
180+
NdbInterpretedCode &code,
181+
const NdbDictionary::Table *tab) {
183182
const NdbDictionary::Column *num_rows_col = tab->getColumn(KEY_TABLE_COL_num_rows);
184183
code.load_op_type(REG1); // Read operation type into register 1
185184
code.branch_eq_const(REG1, RONDB_INSERT, LABEL0); // Inserts go to label 0
@@ -205,3 +204,30 @@ int write_key_row(std::string *response,
205204
}
206205
return 0;
207206
}
207+
208+
int write_key_row_commit(std::string *response,
209+
NdbInterpretedCode &code,
210+
const NdbDictionary::Table *tab) {
211+
const NdbDictionary::Column *num_rows_col = tab->getColumn(KEY_TABLE_COL_num_rows);
212+
code.load_op_type(REG1); // Read operation type into register 1
213+
code.branch_eq_const(REG1, RONDB_INSERT, LABEL0); // Inserts go to label 0
214+
/* UPDATE */
215+
code.read_attr(REG7, num_rows_col);
216+
code.branch_eq_const(REG7, 0, LABEL0);
217+
code.interpret_exit_nok(6000);
218+
219+
/* INSERT */
220+
code.def_label(LABEL0);
221+
code.interpret_exit_ok();
222+
223+
// Program end, now compile code
224+
int ret_code = code.finalise();
225+
if (ret_code != 0)
226+
{
227+
assign_ndb_err_to_response(response,
228+
"Failed to create Interpreted code",
229+
code.getNdbError());
230+
return -1;
231+
}
232+
return 0;
233+
}

‎pink/rondis/string/interpreted_code.h

+6-4
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,10 @@ int write_hset_key_table(Ndb *ndb,
3838
std::string std_key_str,
3939
Uint64 & redis_key_id,
4040
std::string *response);
41-
int write_key_row(std::string *response,
42-
NdbInterpretedCode &code,
43-
const NdbDictionary::Table *tab);
44-
41+
int write_key_row_commit(std::string *response,
42+
NdbInterpretedCode &code,
43+
const NdbDictionary::Table *tab);
44+
int write_key_row_no_commit(std::string *response,
45+
NdbInterpretedCode &code,
46+
const NdbDictionary::Table *tab);
4547
#endif

0 commit comments

Comments
 (0)