database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-30 16:24:05 +03:00
Code Activity

Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.0

Browse Source 
into  mysql.com:/windows/Linux_space/MySQL/mysql-5.0


sql/ha_ndbcluster.cc:
  Auto merged
This commit is contained in:
unknown
2006-08-15 15:57:36 +02:00
parent 0cd51b1afd 6464d62c7b
commit ed481c39e7
5 changed files with 107 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
mysql-test/r/ndb_lock.result
View File

@ -64,17 +64,26 @@ pk u o
insert into t1 values (1,1,1);
drop table t1;
create table t1 (x integer not null primary key, y varchar(32), z integer, key(z)) engine = ndb;
insert into t1 values (1,'one',1), (2,'two',2),(3,"three",3);
insert into t1 values (1,'one',1);
begin;
select * from t1 where x = 1 for update;
x y z
1 one 1
begin;
select * from t1 where x = 1 for update;
ERROR HY000: Lock wait timeout exceeded; try restarting transaction
rollback;
rollback;
insert into t1 values (2,'two',2),(3,"three",3);
begin;
select * from t1 where x = 1 for update;
x y z
1 one 1
select * from t1 where x = 1 for update;
ERROR HY000: Lock wait timeout exceeded; try restarting transaction
select * from t1 where x = 2 for update;
x y z
2 two 2
select * from t1 where x = 1 for update;
ERROR HY000: Lock wait timeout exceeded; try restarting transaction
rollback;
commit;
begin;

15
mysql-test/t/ndb_lock.test
View File

@ -73,7 +73,7 @@ drop table t1;
create table t1 (x integer not null primary key, y varchar(32), z integer, key(z)) engine = ndb;
insert into t1 values (1,'one',1), (2,'two',2),(3,"three",3);
insert into t1 values (1,'one',1);
# PK access
connection con1;
@ -82,11 +82,22 @@ select * from t1 where x = 1 for update;
connection con2;
begin;
select * from t1 where x = 2 for update;
--error 1205
select * from t1 where x = 1 for update;
rollback;
connection con1;
rollback;
insert into t1 values (2,'two',2),(3,"three",3);
begin;
select * from t1 where x = 1 for update;
connection con2;
--error 1205
select * from t1 where x = 1 for update;
select * from t1 where x = 2 for update;
rollback;
connection con1;
commit;

2
ndb/include/ndbapi/NdbTransaction.hpp
View File

@ -140,6 +140,7 @@ class NdbTransaction
friend class NdbIndexOperation;
friend class NdbIndexScanOperation;
friend class NdbBlob;
friend class ha_ndbcluster;
#endif
public:
@ -791,6 +792,7 @@ private:
// optim: any blobs
bool theBlobFlag;
Uint8 thePendingBlobOps;
inline bool hasBlobOperation() { return theBlobFlag; }
static void sendTC_COMMIT_ACK(NdbApiSignal *,
Uint32 transId1, Uint32 transId2,

87
sql/ha_ndbcluster.cc
View File

@ -228,13 +228,15 @@ static int ndb_to_mysql_error(const NdbError *err)
inline
int execute_no_commit(ha_ndbcluster *h, NdbTransaction *trans)
int execute_no_commit(ha_ndbcluster *h, NdbTransaction *trans,
bool force_release)
{
#ifdef NOT_USED
int m_batch_execute= 0;
if (m_batch_execute)
return 0;
#endif
h->release_completed_operations(trans, force_release);
return trans->execute(NdbTransaction::NoCommit,
NdbTransaction::AbortOnError,
h->m_force_send);
@ -267,13 +269,15 @@ int execute_commit(THD *thd, NdbTransaction *trans)
}
inline
int execute_no_commit_ie(ha_ndbcluster *h, NdbTransaction *trans)
int execute_no_commit_ie(ha_ndbcluster *h, NdbTransaction *trans,
bool force_release)
{
#ifdef NOT_USED
int m_batch_execute= 0;
if (m_batch_execute)
return 0;
#endif
h->release_completed_operations(trans, force_release);
return trans->execute(NdbTransaction::NoCommit,
NdbTransaction::AO_IgnoreError,
h->m_force_send);
@ -290,6 +294,7 @@ Thd_ndb::Thd_ndb()
all= NULL;
stmt= NULL;
error= 0;
query_state&= NDB_QUERY_NORMAL;
}
Thd_ndb::~Thd_ndb()
@ -1443,7 +1448,7 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
if ((res= define_read_attrs(buf, op)))
DBUG_RETURN(res);
if (execute_no_commit_ie(this,trans) != 0)
if (execute_no_commit_ie(this,trans,false) != 0)
{
table->status= STATUS_NOT_FOUND;
DBUG_RETURN(ndb_err(trans));
@ -1490,7 +1495,7 @@ int ha_ndbcluster::complemented_pk_read(const byte *old_data, byte *new_data)
ERR_RETURN(trans->getNdbError());
}
}
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
{
table->status= STATUS_NOT_FOUND;
DBUG_RETURN(ndb_err(trans));
@ -1630,7 +1635,7 @@ int ha_ndbcluster::peek_indexed_rows(const byte *record)
}
last= trans->getLastDefinedOperation();
if (first)
res= execute_no_commit_ie(this,trans);
res= execute_no_commit_ie(this,trans,false);
else
{
// Table has no keys
@ -1679,7 +1684,7 @@ int ha_ndbcluster::unique_index_read(const byte *key,
if ((res= define_read_attrs(buf, op)))
DBUG_RETURN(res);
if (execute_no_commit_ie(this,trans) != 0)
if (execute_no_commit_ie(this,trans,false) != 0)
{
table->status= STATUS_NOT_FOUND;
DBUG_RETURN(ndb_err(trans));
@ -1727,7 +1732,7 @@ inline int ha_ndbcluster::fetch_next(NdbScanOperation* cursor)
*/
if (m_ops_pending && m_blobs_pending)
{
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
DBUG_RETURN(ndb_err(trans));
m_ops_pending= 0;
m_blobs_pending= FALSE;
@ -1759,7 +1764,7 @@ inline int ha_ndbcluster::fetch_next(NdbScanOperation* cursor)
{
if (m_transaction_on)
{
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
DBUG_RETURN(-1);
}
else
@ -2063,7 +2068,7 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
DBUG_RETURN(res);
}
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
DBUG_RETURN(ndb_err(trans));
DBUG_RETURN(next_result(buf));
@ -2096,7 +2101,7 @@ int ha_ndbcluster::full_table_scan(byte *buf)
if ((res= define_read_attrs(buf, op)))
DBUG_RETURN(res);
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
DBUG_RETURN(ndb_err(trans));
DBUG_PRINT("exit", ("Scan started successfully"));
DBUG_RETURN(next_result(buf));
@ -2228,7 +2233,7 @@ int ha_ndbcluster::write_row(byte *record)
m_bulk_insert_not_flushed= FALSE;
if (m_transaction_on)
{
if (execute_no_commit(this,trans) != 0)
if (execute_no_commit(this,trans,false) != 0)
{
m_skip_auto_increment= TRUE;
no_uncommitted_rows_execute_failure();
@ -2428,7 +2433,7 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
}
// Execute update operation
if (!cursor && execute_no_commit(this,trans) != 0) {
if (!cursor && execute_no_commit(this,trans,false) != 0) {
no_uncommitted_rows_execute_failure();
DBUG_RETURN(ndb_err(trans));
}
@ -2499,7 +2504,7 @@ int ha_ndbcluster::delete_row(const byte *record)
}
// Execute delete operation
if (execute_no_commit(this,trans) != 0) {
if (execute_no_commit(this,trans,false) != 0) {
no_uncommitted_rows_execute_failure();
DBUG_RETURN(ndb_err(trans));
}
@ -2928,6 +2933,26 @@ int ha_ndbcluster::close_scan()
NdbScanOperation *cursor= m_active_cursor ? m_active_cursor : m_multi_cursor;
if (m_lock_tuple)
{
/*
Lock level m_lock.type either TL_WRITE_ALLOW_WRITE
(SELECT FOR UPDATE) or TL_READ_WITH_SHARED_LOCKS (SELECT
LOCK WITH SHARE MODE) and row was not explictly unlocked
with unlock_row() call
*/
NdbOperation *op;
// Lock row
DBUG_PRINT("info", ("Keeping lock on scanned row"));
if (!(op= cursor->lockCurrentTuple()))
{
m_lock_tuple= false;
ERR_RETURN(trans->getNdbError());
}
m_ops_pending++;
}
m_lock_tuple= false;
if (m_ops_pending)
{
/*
@ -2935,7 +2960,7 @@ int ha_ndbcluster::close_scan()
deleteing/updating transaction before closing the scan
*/
DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));
if (execute_no_commit(this,trans) != 0) {
if (execute_no_commit(this,trans,false) != 0) {
no_uncommitted_rows_execute_failure();
DBUG_RETURN(ndb_err(trans));
}
@ -3345,7 +3370,7 @@ int ha_ndbcluster::end_bulk_insert()
m_bulk_insert_not_flushed= FALSE;
if (m_transaction_on)
{
if (execute_no_commit(this, trans) != 0)
if (execute_no_commit(this, trans,false) != 0)
{
no_uncommitted_rows_execute_failure();
my_errno= error= ndb_err(trans);
@ -3518,6 +3543,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
ERR_RETURN(ndb->getNdbError());
no_uncommitted_rows_reset(thd);
thd_ndb->stmt= trans;
thd_ndb->query_state&= NDB_QUERY_NORMAL;
trans_register_ha(thd, FALSE, &ndbcluster_hton);
}
else
@ -3533,6 +3559,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
ERR_RETURN(ndb->getNdbError());
no_uncommitted_rows_reset(thd);
thd_ndb->all= trans;
thd_ndb->query_state&= NDB_QUERY_NORMAL;
trans_register_ha(thd, TRUE, &ndbcluster_hton);
/*
@ -3739,6 +3766,7 @@ int ha_ndbcluster::start_stmt(THD *thd, thr_lock_type lock_type)
thd_ndb->stmt= trans;
trans_register_ha(thd, FALSE, &ndbcluster_hton);
}
thd_ndb->query_state&= NDB_QUERY_NORMAL;
m_active_trans= trans;
// Start of statement
@ -5986,6 +6014,30 @@ int ha_ndbcluster::write_ndb_file()
DBUG_RETURN(error);
}
void
ha_ndbcluster::release_completed_operations(NdbTransaction *trans,
bool force_release)
{
if (trans->hasBlobOperation())
{
/* We are reading/writing BLOB fields,
releasing operation records is unsafe
*/
return;
}
if (!force_release)
{
if (get_thd_ndb(current_thd)->query_state & NDB_QUERY_MULTI_READ_RANGE)
{
/* We are batching reads and have not consumed all fetched
rows yet, releasing operation records is unsafe
*/
return;
}
}
trans->releaseCompletedOperations();
}
int
ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
KEY_MULTI_RANGE *ranges,
@ -6000,6 +6052,7 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
NDB_INDEX_TYPE index_type= get_index_type(active_index);
ulong reclength= table->s->reclength;
NdbOperation* op;
Thd_ndb *thd_ndb= get_thd_ndb(current_thd);
if (uses_blob_value(m_retrieve_all_fields))
{
@ -6013,7 +6066,7 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
sorted,
buffer));
}
thd_ndb->query_state|= NDB_QUERY_MULTI_READ_RANGE;
m_disable_multi_read= FALSE;
/**
@ -6160,7 +6213,7 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
*/
m_current_multi_operation=
lastOp ? lastOp->next() : m_active_trans->getFirstDefinedOperation();
if (!(res= execute_no_commit_ie(this, m_active_trans)))
if (!(res= execute_no_commit_ie(this, m_active_trans, true)))
{
m_multi_range_defined= multi_range_curr;
multi_range_curr= ranges;

12
sql/ha_ndbcluster.h
View File

@ -435,6 +435,11 @@ class Ndb_cond_traverse_context
Ndb_rewrite_context *rewrite_stack;
};
typedef enum ndb_query_state_bits {
NDB_QUERY_NORMAL = 0,
NDB_QUERY_MULTI_READ_RANGE = 1
} NDB_QUERY_STATE_BITS;
/*
Place holder for ha_ndbcluster thread specific data
*/
@ -451,6 +456,7 @@ class Thd_ndb
NdbTransaction *stmt;
int error;
List<NDB_SHARE> changed_tables;
uint query_state;
};
class ha_ndbcluster: public handler
@ -672,8 +678,8 @@ private:
NdbScanOperation* op);
friend int execute_commit(ha_ndbcluster*, NdbTransaction*);
friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*);
friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*);
friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*, bool);
friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*, bool);
NdbTransaction *m_active_trans;
NdbScanOperation *m_active_cursor;
@ -716,6 +722,8 @@ private:
bool m_force_send;
ha_rows m_autoincrement_prefetch;
bool m_transaction_on;
void release_completed_operations(NdbTransaction*, bool);
Ndb_cond_stack *m_cond_stack;
bool m_disable_multi_read;
byte *m_multi_range_result_ptr;