Skip to content

M
MariaDB
Commit cf4cfab2 authored by unknown's avatar unknown
Browse files
Options

WL#1791 Handler: support update of primary key

parent d8a914d4
Show whitespace changes
Inline Side-by-side
Showing with 191 additions and 81 deletions
mysql-test/r/ndb_basic.result
View file @ cf4cfab2
DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7; DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
CREATE TABLE t1 ( CREATE TABLE t1 (
pk1 INT NOT NULL PRIMARY KEY, pk1 INT NOT NULL PRIMARY KEY,
attr1 INT NOT NULL attr1 INT NOT NULL,
attr2 INT,
attr3 VARCHAR(10)
) ENGINE=ndbcluster; ) ENGINE=ndbcluster;
INSERT INTO t1 VALUES (9410,9412); INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9411,9413, 17, '9413');
SELECT pk1 FROM t1; SELECT pk1 FROM t1;
pk1 pk1
9410 9410
9411
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
9410 9412 9410 9412 NULL 9412
9411 9413 17 9413
SELECT t1.* FROM t1; SELECT t1.* FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
9410 9412 9410 9412 NULL 9412
9411 9413 17 9413
UPDATE t1 SET attr1=1 WHERE pk1=9410; UPDATE t1 SET attr1=1 WHERE pk1=9410;
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
9410 1 9410 1 NULL 9412
9411 9413 17 9413
UPDATE t1 SET pk1=2 WHERE attr1=1; UPDATE t1 SET pk1=2 WHERE attr1=1;
ERROR 42000: Table 't1' uses an extension that doesn't exist in this MySQL version
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
9410 1 2 1 NULL 9412
9411 9413 17 9413
UPDATE t1 SET pk1=pk1 + 1;
SELECT * FROM t1;
pk1 attr1 attr2 attr3
9412 9413 17 9413
3 1 NULL 9412
DELETE FROM t1; DELETE FROM t1;
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
INSERT INTO t1 VALUES (9410,9412), (9411, 9413), (9408, 8765), INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9408, 8765, NULL, '8765'),
(7,8), (8,9), (9,10), (10,11), (11,12), (12,13), (13,14); (7,8, NULL, NULL), (8,9, NULL, NULL), (9,10, NULL, NULL), (10,11, NULL, NULL), (11,12, NULL, NULL), (12,13, NULL, NULL), (13,14, NULL, NULL);
UPDATE t1 SET attr1 = 9999; UPDATE t1 SET attr1 = 9999;
SELECT * FROM t1 ORDER BY pk1; SELECT * FROM t1 ORDER BY pk1;
pk1 attr1 pk1 attr1 attr2 attr3
7 9999 7 9999 NULL NULL
8 9999 8 9999 NULL NULL
9 9999 9 9999 NULL NULL
10 9999 10 9999 NULL NULL
11 9999 11 9999 NULL NULL
12 9999 12 9999 NULL NULL
13 9999 13 9999 NULL NULL
9408 9999 9408 9999 NULL 8765
9410 9999 9410 9999 NULL 9412
9411 9999
UPDATE t1 SET attr1 = 9998 WHERE pk1 < 1000; UPDATE t1 SET attr1 = 9998 WHERE pk1 < 1000;
SELECT * FROM t1 ORDER BY pk1; SELECT * FROM t1 ORDER BY pk1;
pk1 attr1 pk1 attr1 attr2 attr3
7 9998 7 9998 NULL NULL
8 9998 8 9998 NULL NULL
9 9998 9 9998 NULL NULL
10 9998 10 9998 NULL NULL
11 9998 11 9998 NULL NULL
12 9998 12 9998 NULL NULL
13 9998 13 9998 NULL NULL
9408 9999 9408 9999 NULL 8765
9410 9999 9410 9999 NULL 9412
9411 9999
UPDATE t1 SET attr1 = 9997 WHERE attr1 = 9999; UPDATE t1 SET attr1 = 9997 WHERE attr1 = 9999;
SELECT * FROM t1 ORDER BY pk1; SELECT * FROM t1 ORDER BY pk1;
pk1 attr1 pk1 attr1 attr2 attr3
7 9998 7 9998 NULL NULL
8 9998 8 9998 NULL NULL
9 9998 9 9998 NULL NULL
10 9998 10 9998 NULL NULL
11 9998 11 9998 NULL NULL
12 9998 12 9998 NULL NULL
13 9998 13 9998 NULL NULL
9408 9997 9408 9997 NULL 8765
9410 9997 9410 9997 NULL 9412
9411 9997
DELETE FROM t1 WHERE pk1 = 9410; DELETE FROM t1 WHERE pk1 = 9410;
SELECT * FROM t1 ORDER BY pk1; SELECT * FROM t1 ORDER BY pk1;
pk1 attr1 pk1 attr1 attr2 attr3
7 9998 7 9998 NULL NULL
8 9998 8 9998 NULL NULL
9 9998 9 9998 NULL NULL
10 9998 10 9998 NULL NULL
11 9998 11 9998 NULL NULL
12 9998 12 9998 NULL NULL
13 9998 13 9998 NULL NULL
9408 9997 9408 9997 NULL 8765
9411 9997
DELETE FROM t1; DELETE FROM t1;
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
INSERT INTO t1 values (1, 4), (2, 4), (3, 5), (4, 4), (5, 5); INSERT INTO t1 values (1, 4, NULL, NULL), (2, 4, NULL, NULL), (3, 5, NULL, NULL), (4, 4, NULL, NULL), (5, 5, NULL, NULL);
DELETE FROM t1 WHERE attr1=4; DELETE FROM t1 WHERE attr1=4;
SELECT * FROM t1 order by pk1; SELECT * FROM t1 order by pk1;
pk1 attr1 pk1 attr1 attr2 attr3
3 5 3 5 NULL NULL
5 5 5 5 NULL NULL
DELETE FROM t1; DELETE FROM t1;
INSERT INTO t1 VALUES (9410,9412), (9411, 9413); INSERT INTO t1 VALUES (9410,9412, NULL, NULL), (9411, 9413, NULL, NULL);
DELETE FROM t1 WHERE pk1 = 9410; DELETE FROM t1 WHERE pk1 = 9410;
SELECT * FROM t1; SELECT * FROM t1;
pk1 attr1 pk1 attr1 attr2 attr3
9411 9413 9411 9413 NULL NULL
DROP TABLE t1; DROP TABLE t1;
CREATE TABLE t1 (id INT, id2 int) engine=ndbcluster; CREATE TABLE t1 (id INT, id2 int) engine=ndbcluster;
INSERT INTO t1 values(3456, 7890); INSERT INTO t1 values(3456, 7890);
......
mysql-test/r/ndb_index_ordered.result
View file @ cf4cfab2
...@@ -91,6 +91,15 @@ a b c ...@@ -91,6 +91,15 @@ a b c
4 6 12 4 6 12
5 7 12 5 7 12
6 7 12 6 7 12
update t1 set a = a + 10 where b > 1 and b < 7;
select * from t1 order by a;
a b c
5 7 12
6 7 12
11 2 13
12 3 13
13 4 12
14 6 12
drop table t1; drop table t1;
CREATE TABLE t1 ( CREATE TABLE t1 (
a int unsigned NOT NULL PRIMARY KEY, a int unsigned NOT NULL PRIMARY KEY,
......
mysql-test/t/ndb_basic.test
View file @ cf4cfab2
...@@ -14,10 +14,12 @@ DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7; ...@@ -14,10 +14,12 @@ DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
# #
CREATE TABLE t1 ( CREATE TABLE t1 (
pk1 INT NOT NULL PRIMARY KEY, pk1 INT NOT NULL PRIMARY KEY,
attr1 INT NOT NULL attr1 INT NOT NULL,
attr2 INT,
attr3 VARCHAR(10)
) ENGINE=ndbcluster; ) ENGINE=ndbcluster;
INSERT INTO t1 VALUES (9410,9412); INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9411,9413, 17, '9413');
SELECT pk1 FROM t1; SELECT pk1 FROM t1;
SELECT * FROM t1; SELECT * FROM t1;
...@@ -27,18 +29,19 @@ SELECT t1.* FROM t1; ...@@ -27,18 +29,19 @@ SELECT t1.* FROM t1;
UPDATE t1 SET attr1=1 WHERE pk1=9410; UPDATE t1 SET attr1=1 WHERE pk1=9410;
SELECT * FROM t1; SELECT * FROM t1;
# Can't UPDATE PK! Test that correct error is returned # Update primary key
-- error 1112
UPDATE t1 SET pk1=2 WHERE attr1=1; UPDATE t1 SET pk1=2 WHERE attr1=1;
SELECT * FROM t1; SELECT * FROM t1;
UPDATE t1 SET pk1=pk1 + 1;
SELECT * FROM t1;
# Delete the record # Delete the record
DELETE FROM t1; DELETE FROM t1;
SELECT * FROM t1; SELECT * FROM t1;
# Insert more records and update them all at once # Insert more records and update them all at once
INSERT INTO t1 VALUES (9410,9412), (9411, 9413), (9408, 8765), INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9408, 8765, NULL, '8765'),
(7,8), (8,9), (9,10), (10,11), (11,12), (12,13), (13,14); (7,8, NULL, NULL), (8,9, NULL, NULL), (9,10, NULL, NULL), (10,11, NULL, NULL), (11,12, NULL, NULL), (12,13, NULL, NULL), (13,14, NULL, NULL);
UPDATE t1 SET attr1 = 9999; UPDATE t1 SET attr1 = 9999;
SELECT * FROM t1 ORDER BY pk1; SELECT * FROM t1 ORDER BY pk1;
...@@ -58,13 +61,13 @@ SELECT * FROM t1; ...@@ -58,13 +61,13 @@ SELECT * FROM t1;
# Insert three records with attr1=4 and two with attr1=5 # Insert three records with attr1=4 and two with attr1=5
# Delete all with attr1=4 # Delete all with attr1=4
INSERT INTO t1 values (1, 4), (2, 4), (3, 5), (4, 4), (5, 5); INSERT INTO t1 values (1, 4, NULL, NULL), (2, 4, NULL, NULL), (3, 5, NULL, NULL), (4, 4, NULL, NULL), (5, 5, NULL, NULL);
DELETE FROM t1 WHERE attr1=4; DELETE FROM t1 WHERE attr1=4;
SELECT * FROM t1 order by pk1; SELECT * FROM t1 order by pk1;
DELETE FROM t1; DELETE FROM t1;
# Insert two records and delete one # Insert two records and delete one
INSERT INTO t1 VALUES (9410,9412), (9411, 9413); INSERT INTO t1 VALUES (9410,9412, NULL, NULL), (9411, 9413, NULL, NULL);
DELETE FROM t1 WHERE pk1 = 9410; DELETE FROM t1 WHERE pk1 = 9410;
SELECT * FROM t1; SELECT * FROM t1;
DROP TABLE t1; DROP TABLE t1;
......
mysql-test/t/ndb_index_ordered.test
View file @ cf4cfab2
...@@ -44,6 +44,9 @@ update t1 set c = 13 where b <= 3; ...@@ -44,6 +44,9 @@ update t1 set c = 13 where b <= 3;
select * from t1 order by a; select * from t1 order by a;
update t1 set b = b + 1 where b > 4 and b < 7; update t1 set b = b + 1 where b > 4 and b < 7;
select * from t1 order by a; select * from t1 order by a;
-- Update primary key
update t1 set a = a + 10 where b > 1 and b < 7;
select * from t1 order by a;
# #
# Delete using ordered index scan # Delete using ordered index scan
......
sql/ha_ndbcluster.cc
View file @ cf4cfab2
...@@ -624,7 +624,7 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf) ...@@ -624,7 +624,7 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
return res; return res;
} }
// Read non-key field(s) unless HA_EXTRA_RETRIEVE_ALL_COLS // Read all wanted non-key field(s) unless HA_EXTRA_RETRIEVE_ALL_COLS
for (i= 0; i < no_fields; i++) for (i= 0; i < no_fields; i++)
{ {
Field *field= table->field[i]; Field *field= table->field[i];
...@@ -657,6 +657,62 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf) ...@@ -657,6 +657,62 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
} }
/*
Read one complementing record from NDB using primary key from old_data
*/
int ha_ndbcluster::complemented_pk_read(const byte *old_data, byte *new_data)
{
uint no_fields= table->fields, i;
NdbConnection *trans= m_active_trans;
NdbOperation *op;
THD *thd= current_thd;
DBUG_ENTER("complemented_pk_read");
if (retrieve_all_fields)
// We have allready retrieved all fields, nothing to complement
DBUG_RETURN(0);
if (!(op= trans->getNdbOperation(m_tabname)) || op->readTuple() != 0)
goto err;
int res;
if (res= set_primary_key_from_old_data(op, old_data))
return res;
// Read all unreferenced non-key field(s)
for (i= 0; i < no_fields; i++)
{
Field *field= table->field[i];
if (!(field->flags & PRI_KEY_FLAG) &&
(thd->query_id != field->query_id))
{
if (get_ndb_value(op, i, field->ptr))
goto err;
}
else
{
// Attribute was not to be read
m_value[i]= NULL;
}
}
if (trans->execute(NoCommit, IgnoreError) != 0)
{
table->status= STATUS_NOT_FOUND;
DBUG_RETURN(ndb_err(trans));
}
// The value have now been fetched from NDB
unpack_record(new_data);
table->status= 0;
DBUG_RETURN(0);
err:
ERR_RETURN(trans->getNdbError());
}
/* /*
Read one record from NDB using unique secondary index Read one record from NDB using unique secondary index
*/ */
...@@ -1173,10 +1229,43 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data) ...@@ -1173,10 +1229,43 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
if (table->timestamp_on_update_now) if (table->timestamp_on_update_now)
update_timestamp(new_data+table->timestamp_on_update_now-1); update_timestamp(new_data+table->timestamp_on_update_now-1);
/* Check for update of primary key and return error */ /* Check for update of primary key for special handling */
if ((table->primary_key != MAX_KEY) && if ((table->primary_key != MAX_KEY) &&
(key_cmp(table->primary_key, old_data, new_data))) (key_cmp(table->primary_key, old_data, new_data)))
DBUG_RETURN(HA_ERR_UNSUPPORTED); {
DBUG_PRINT("info", ("primary key update, doing pk read+insert+delete"));
// Get all old fields, since we optimize away fields not in query
int read_res = complemented_pk_read(old_data, new_data);
if (read_res)
{
DBUG_PRINT("info", ("pk read failed"));
DBUG_RETURN(read_res);
}
// Insert new row
int insert_res = write_row(new_data);
if (!insert_res)
{
// Delete old row
DBUG_PRINT("info", ("insert succeded"));
int delete_res = delete_row(old_data);
if (!delete_res)
{
DBUG_PRINT("info", ("insert+delete succeeded"));
DBUG_RETURN(0);
}
else
{
DBUG_PRINT("info", ("delete failed"));
DBUG_RETURN(delete_row(new_data));
}
}
else
{
DBUG_PRINT("info", ("insert failed"));
DBUG_RETURN(insert_res);
}
}
if (cursor) if (cursor)
{ {
...@@ -1350,7 +1439,6 @@ void ha_ndbcluster::unpack_record(byte* buf) ...@@ -1350,7 +1439,6 @@ void ha_ndbcluster::unpack_record(byte* buf)
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
/* /*
Utility function to print/dump the fetched field Utility function to print/dump the fetched field
*/ */
......
sql/ha_ndbcluster.h
View file @ cf4cfab2
...@@ -156,8 +156,8 @@ class ha_ndbcluster: public handler ...@@ -156,8 +156,8 @@ class ha_ndbcluster: public handler
NDB_INDEX_TYPE get_index_type(uint idx_no) const; NDB_INDEX_TYPE get_index_type(uint idx_no) const;
NDB_INDEX_TYPE get_index_type_from_table(uint index_no) const; NDB_INDEX_TYPE get_index_type_from_table(uint index_no) const;
int pk_read(const byte *key, uint key_len, int pk_read(const byte *key, uint key_len, byte *buf);
byte *buf); int complemented_pk_read(const byte *old_data, byte *new_data);
int unique_index_read(const byte *key, uint key_len, int unique_index_read(const byte *key, uint key_len,
byte *buf); byte *buf);
int ordered_index_scan(const key_range *start_key, int ordered_index_scan(const key_range *start_key,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7