Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-5.1-new (d7901a2a) · Commits · Software / OSDI20 Artifacts / mariadb

storage/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp

+1 −1

Original line number	Diff line number	Diff line
		@@ -448,7 +448,7 @@ void Dbtup::execTUP_COMMITREQ(Signal* signal)
		if(!regOperPtr.p->is_first_operation())
		{
		/**
		* Out of order commit
		* Out of order commit XXX check effect on triggers
		*/
		fix_commit_order(regOperPtr);
		}

storage/ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp

+10 −1

Original line number	Diff line number	Diff line
		@@ -559,8 +559,9 @@ Dbtup::fireDetachedTriggers(KeyReqStruct *req_struct,

		/**
		* Set correct operation type and fix change mask
		* Note ALLOC is set in "orig" tuple
		*/
		if(req_struct->m_tuple_ptr->m_header_bits & Tuple_header::ALLOC)
		if(save_ptr->m_header_bits & Tuple_header::ALLOC)
		{
		if(save == ZDELETE)
		{
		@@ -571,6 +572,14 @@ Dbtup::fireDetachedTriggers(KeyReqStruct *req_struct,
		}
		regOperPtr->op_struct.op_type = ZINSERT;
		}
		else if (save == ZINSERT)
		/**
		* Tuple was not created but last op is INSERT.
		* This is possible only on DELETE + INSERT
		*/
		{
		regOperPtr->op_struct.op_type = ZUPDATE;
		}

		ndbrequire(regOperPtr->is_first_operation());
		triggerList.first(trigPtr);

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

+3 −3

Original line number	Diff line number	Diff line
		@@ -376,10 +376,10 @@ NdbEventOperationImpl::receive_event()
		AttributeHeader(*aAttrPtr).getAttributeId());
		receive_data(tAttr, aDataPtr, tDataSz);
		if (is_update)
		{
		receive_data(tAttr1, aDataPtr, tDataSz);
		else
		tAttr1->setUNDEFINED(); // do not leave unspecified
		tAttr1= tAttr1->next();
		}
		// next
		aAttrPtr++;
		aDataPtr+= (tDataSz + 3) >> 2;

storage/ndb/test/ndbapi/test_event_merge.cpp

+100 −70

Original line number	Diff line number	Diff line
		@@ -47,8 +47,8 @@
		* There are 5 ways (ignoring NUL operand) to compose 2 ops:
		* 5.0 bugs 5.1 bugs
		* INS o DEL = NUL
		* INS o UPD = INS 5.1
		* DEL o INS = UPD type=INS 5.1
		* INS o UPD = INS type=INS
		* DEL o INS = UPD type=INS type=INS
		* UPD o DEL = DEL no event
		* UPD o UPD = UPD
		*/
		@@ -78,6 +78,7 @@ static NdbOperation* g_op = 0;
		static const char* g_tabname = "tem1";
		static const char* g_evtname = "tem1ev1";
		static const uint g_charlen = 5;
		static const char* g_charval = "abcde";
		static const char* g_csname = "latin1_swedish_ci";

		static const NdbDictionary::Table* g_tab = 0;
		@@ -229,9 +230,9 @@ createtable()
		chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
		chkdb(g_op->insertTuple() == 0);
		Uint32 pk1;
		char pk2[g_charlen];
		char pk2[g_charlen + 1];
		pk1 = g_maxpk;
		memset(pk2, 0x20, g_charlen);
		sprintf(pk2, "%-*u", g_charlen, pk1);
		chkdb(g_op->equal("pk1", (char*)&pk1) == 0);
		chkdb(g_op->equal("pk2", (char*)&pk2[0]) == 0);
		chkdb(g_con->execute(Commit) == 0);
		@@ -256,9 +257,9 @@ droptable()

		struct Data {
		Uint32 pk1;
		char pk2[g_charlen];
		char pk2[g_charlen + 1];
		Uint32 seq;
		char cc1[g_charlen];
		char cc1[g_charlen + 1];
		void* ptr[g_ncol];
		int ind[g_ncol]; // -1 = no data, 1 = NULL, 0 = not NULL
		void init() {
		@@ -276,6 +277,30 @@ struct Data {
		}
		};

		static int
		cmpdata(const Data& d1, const Data& d2)
		{
		uint i;
		for (i = 0; i < g_ncol; i++) {
		const Col& c = getcol(i);
		if (d1.ind[i] != d2.ind[i])
		return 1;
		if (d1.ind[i] == 0 && memcmp(d1.ptr[i], d2.ptr[i], c.size) != 0)
		return 1;
		}
		return 0;
		}

		static int
		cmpdata(const Data (&d1)[2], const Data (&d2)[2])
		{
		if (cmpdata(d1[0], d2[0]) != 0)
		return 1;
		if (cmpdata(d1[1], d2[1]) != 0)
		return 1;
		return 0;
		}

		static NdbOut&
		operator<<(NdbOut& out, const Data& d)
		{
		@@ -304,7 +329,7 @@ operator<<(NdbOut& out, const Data& d)
		break;
		n--;
		}
		out << buf;
		out << "'" << buf << "'";
		}
		break;
		default:
		@@ -370,7 +395,7 @@ operator<<(NdbOut& out, Op::Type t)
		static NdbOut&
		operator<<(NdbOut& out, const Op& op)
		{
		out << "t=" << op.type;
		out << op.type;
		out << " " << op.data[0];
		out << " [" << op.data[1] << "]";
		return out;
		@@ -504,13 +529,9 @@ checkop(const Op* op, Uint32& pk1)
		if (t == Op::INS) {
		chkrc(d[1].ind[i] == -1);
		} else if (t == Op::DEL) {
		#ifdef ndb_event_cares_about_pk_pre_data
		chkrc(d[1].ind[i] == -1);
		#endif
		} else {
		#ifdef ndb_event_cares_about_pk_pre_data
		chkrc(d[1].ind[i] == 0);
		#endif
		}
		} else {
		if (t == Op::INS) {
		@@ -547,43 +568,24 @@ copycol(const Col& c, const Data& d1, Data& d3)
		}

		static void
		copykeys(const Data& d1, Data& d3)
		{
		uint i;
		for (i = 0; i < g_ncol; i++) {
		const Col& c = g_col[i];
		if (c.pk)
		copycol(c, d1, d3);
		}
		}

		static void
		copydata(const Data& d1, Data& d3)
		copydata(const Data& d1, Data& d3, bool pk, bool nonpk)
		{
		uint i;
		for (i = 0; i < g_ncol; i++) {
		const Col& c = g_col[i];
		if (c.pk && pk \|\| ! c.pk && nonpk)
		copycol(c, d1, d3);
		}
		}

		static void
		copyop(const Op* op1, Op* op3)
		{
		op3->type = op1->type;
		copydata(op1->data[0], op3->data[0]);
		copydata(op1->data[1], op3->data[1]);
		Uint32 pk1_tmp;
		reqrc(checkop(op3, pk1_tmp) == 0);
		}

		// not needed for ops
		static void
		compdata(const Data& d1, const Data& d2, Data& d3) // d2 overrides d1
		compdata(const Data& d1, const Data& d2, Data& d3, bool pk, bool nonpk)
		{
		uint i;
		for (i = 0; i < g_ncol; i++) {
		const Col& c = g_col[i];
		if (c.pk && pk \|\| ! c.pk && nonpk) {
		const Data* d = 0;
		if (d1.ind[i] == -1 && d2.ind[i] == -1)
		d3.ind[i] = -1;
		@@ -597,6 +599,17 @@ compdata(const Data& d1, const Data& d2, Data& d3) // d2 overrides d1
		copycol(c, *d, d3);
		}
		}
		}

		static void
		copyop(const Op* op1, Op* op3)
		{
		op3->type = op1->type;
		copydata(op1->data[0], op3->data[0], true, true);
		copydata(op1->data[1], op3->data[1], true, true);
		Uint32 pk1_tmp;
		reqrc(checkop(op3, pk1_tmp) == 0);
		}

		static int
		compop(const Op* op1, const Op* op2, Op* op3) // op1 o op2 = op3
		@@ -612,11 +625,14 @@ compop(const Op* op1, const Op* op2, Op* op3) // op1 o op2 = op3
		}
		chkrc((comp = comptype(op1->type, op2->type)) != 0);
		op3->type = comp->t3;
		copykeys(op2->data[0], op3->data[0]);
		// sucks
		copydata(op2->data[0], op3->data[0], true, false);
		if (op3->type != Op::DEL)
		copydata(op2->data[0], op3->data[0]);
		copydata(op2->data[0], op3->data[0], false, true);
		if (op3->type != Op::INS)
		copydata(op1->data[1], op3->data[1]);
		copydata(op1->data[1], op3->data[1], false, true);
		if (op3->type == Op::UPD)
		copydata(op2->data[0], op3->data[1], true, false);
		Uint32 pk1_tmp;
		reqrc(checkop(op3, pk1_tmp) == 0);
		// not eliminating identical post-pre fields
		@@ -705,9 +721,9 @@ waitgci() // wait for event to be installed and for at least 1 GCI to pass
		chkdb((g_con = g_ndb->startTransaction()) != 0);
		{ // forced to exec a dummy op
		Uint32 pk1;
		char pk2[g_charlen];
		char pk2[g_charlen + 1];
		pk1 = g_maxpk;
		memset(pk2, 0x20, g_charlen);
		sprintf(pk2, "%-*u", g_charlen, pk1);
		chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
		chkdb(g_op->readTuple() == 0);
		chkdb(g_op->equal("pk1", (char*)&pk1) == 0);
		@@ -723,6 +739,7 @@ waitgci() // wait for event to be installed and for at least 1 GCI to pass
		break;
		}
		i = 1;
		sleep(1);
		}
		return 0;
		}
		@@ -732,11 +749,14 @@ makeop(Op* op, Uint32 pk1, Op::Type t, const Op* prev_op)
		{
		op->type = t;
		if (t != Op::INS)
		copydata(prev_op->data[0], op->data[1]);
		copydata(prev_op->data[0], op->data[1], true, true);
		uint i;
		for (i = 0; i < g_ncol; i++) {
		const Col& c = getcol(i);
		Data (&d)[2] = op->data;
		if (c.pk && t == Op::DEL) {
		d[1].ind[i] = -1;
		}
		if (i == getcol("pk1").no) {
		d[0].pk1 = pk1;
		d[0].ind[i] = 0;
		@@ -758,7 +778,7 @@ makeop(Op* op, Uint32 pk1, Op::Type t, const Op* prev_op)
		}
		uint u;
		u = urandom(100);
		if (c.nullable && u < 20) {
		if (c.nullable && u < 10) {
		d[0].ind[i] = 1;
		continue;
		}
		@@ -776,8 +796,8 @@ makeop(Op* op, Uint32 pk1, Op::Type t, const Op* prev_op)
		char* p = (char*)d[0].ptr[i];
		uint j;
		for (j = 0; j < g_charlen; j++) {
		uint v = urandom(3);
		p[j] = j < u ? "abcde"[v] : 0x20;
		uint v = urandom(strlen(g_charval));
		p[j] = j < u ? g_charval[v] : 0x20;
		}
		}
		break;
		@@ -993,9 +1013,9 @@ static int
		matchevent(Op* ev)
		{
		Op::Type t = ev->type;
		Data (&d)[2] = ev->data;
		Data (&d2)[2] = ev->data;
		// get PK
		Uint32 pk1 = d[0].pk1;
		Uint32 pk1 = d2[0].pk1;
		chkrc(pk1 < g_opts.maxpk);
		// on error repeat and print details
		uint loop = 0;
		@@ -1016,30 +1036,35 @@ matchevent(Op* ev)
		op = op->next_op;
		}
		if (com_op->type != Op::NUL) {
		if (com_op->type == t) {
		const Data (&d2)[2] = com_op->data;
		if (t == Op::INS && d2[0].seq == d[0].seq \|\|
		t == Op::DEL && d2[1].seq == d[1].seq \|\|
		t == Op::UPD && d2[0].seq == d[0].seq) {
		if (cnt == g_ev_cnt[pk1]) {
		if (! com_op->match) {
		ll2("match pos " << cnt);
		const Data (&d1)[2] = com_op->data;
		if (cmpdata(d1, d2) == 0) {
		bool tmpok = true;
		if (com_op->type != t) {
		ll2("***: wrong type " << com_op->type << " != " << t);
		tmpok = false;
		}
		if (com_op->match) {
		ll2("***: duplicate match");
		tmpok = false;
		}
		if (cnt != g_ev_cnt[pk1]) {
		ll2("***: wrong pos " << cnt << " != " << g_ev_cnt[pk1]);
		tmpok = false;
		}
		if (tmpok) {
		ok = com_op->match = true;
		} else {
		ll2("duplicate match");
		}
		} else {
		ll2("match bad pos event=" << g_ev_cnt[pk1] << " op=" << cnt);
		}
		ll2("===: match");
		}
		}
		cnt++;
		}
		com_op = com_op->next_com;
		}
		if (ok)
		if (ok) {
		ll1("matchevent: match");
		return 0;
		ll2("no match");
		}
		ll1("matchevent: ERROR: no match");
		if (g_loglevel >= 2)
		return -1;
		loop++;
		@@ -1100,12 +1125,13 @@ runevents()
		{
		ll1("runevents");
		NdbEventOperation* evt_op;
		uint npoll = 3;
		uint mspoll = 1000;
		uint npoll = 7; // strangely long delay
		while (npoll != 0) {
		npoll--;
		int ret;
		ll1("poll");
		ret = g_ndb->pollEvents(1000);
		ret = g_ndb->pollEvents(mspoll);
		if (ret <= 0)
		continue;
		while (1) {
		@@ -1180,7 +1206,7 @@ runtest()
		chkrc(createtable() == 0);
		chkrc(createevent() == 0);
		uint n;
		for (n = 0; n < g_opts.loop; n++) {
		for (n = 0; g_opts.loop == 0 \|\| n < g_opts.loop; n++) {
		ll0("loop " << n);
		setseed(n);
		resetmem();
		@@ -1280,6 +1306,10 @@ main(int argc, char** argv)
		return NDBT_ProgramExit(NDBT_OK);
		}
		}
		if (g_evt_op != 0) {
		(void)dropeventop();
		g_evt_op = 0;
		}
		delete g_ndb;
		delete g_ncc;
		return NDBT_ProgramExit(NDBT_FAILED);