fixed crash on replication of update with JSON attribute; fixed #3365…

…; added test 456

src/searchdreplication.cpp

@@ -613,7 +613,8 @@ std::pair<int, CSphString> WaitClusterCommit ( const CSphString& sCluster, int i
 // ver 0x104 added docstore from RT index
 // ver 0x105 fixed CSphWordHit serialization - instead of direct raw blob copy only fields sent (16 bytes vs 24)
 // ver 0x106 add total indexed bytes to accum
-static const WORD g_iReplicateCommandVer = 0x106;
+// ver 0x107 add blobs vector to replicate update statement
+static const WORD g_iReplicateCommandVer = 0x107;
 
 // log debug info about cluster nodes as current nodes views that
 static void LogGroupView ( const wsrep_view_info_t * pView )
@@ -5901,6 +5902,7 @@ void SaveUpdate ( const CSphAttrUpdate & tUpd, CSphVector<BYTE> & dOut )
 	SaveArray ( tUpd.m_dDocids, tWriter );
 	SaveArray ( tUpd.m_dRowOffset, tWriter );
 	SaveArray ( tUpd.m_dPool, tWriter );
+	SaveArray ( tUpd.m_dBlobs, tWriter );
 }
 
 int LoadUpdate ( const BYTE * pBuf, int iLen, CSphAttrUpdate & tUpd, bool & bBlob )
@@ -5922,6 +5924,7 @@ int LoadUpdate ( const BYTE * pBuf, int iLen, CSphAttrUpdate & tUpd, bool & bBlo
 	GetArray ( tUpd.m_dDocids, tIn );
 	GetArray ( tUpd.m_dRowOffset, tIn );
 	GetArray ( tUpd.m_dPool, tIn );
+	GetArray ( tUpd.m_dBlobs, tIn );
 
 	return tIn.GetPos();
 }

test/test_456/model.bin

@@ -0,0 +1 @@
+a:1:{i:0;a:13:{i:0;a:2:{s:8:"sphinxql";s:104:"create table rt1 (id bigint, attrs json, idd integer, content text indexed, s string) rt_mem_limit='12M'";s:14:"total_affected";i:0;}i:1;a:2:{s:8:"sphinxql";s:19:"create cluster test";s:14:"total_affected";i:0;}i:2;a:2:{s:8:"sphinxql";s:26:"alter cluster test add rt1";s:14:"total_affected";i:0;}i:3;a:2:{s:8:"sphinxql";s:169:"insert into test:rt1 (id, content, attrs, idd, s) values (1, 'text content first object', '{"phones":{"n":[971526025257],"t":[1]},"mails":[],"version":1}', 11, 'some 1')";s:14:"total_affected";i:1;}i:4;a:2:{s:8:"sphinxql";s:170:"insert into test:rt1 (id, content, attrs, idd, s) values (2, 'text content second document', '{"phones":{"n":[971526025257],"t":[1]},"mails":[],"version":1}', 2, 'other')";s:14:"total_affected";i:1;}i:5;a:3:{s:5:"agent";s:1:"1";s:8:"sphinxql";s:37:"join cluster test at '%addr_connect%'";s:14:"total_affected";i:0;}i:6;a:3:{s:5:"agent";s:1:"1";s:8:"sphinxql";s:101:"UPDATE test:rt1 set attrs='{"phones":{"n":[966549463121],"t":[1]},"mails":[],"version":6}' where id=2";s:14:"total_affected";i:1;}i:7;a:3:{s:5:"agent";s:1:"1";s:8:"sphinxql";s:44:"UPDATE test:rt1 set s='another 1' where id=1";s:14:"total_affected";i:1;}i:8;a:4:{s:5:"agent";s:1:"1";s:8:"sphinxql";s:46:"show status like 'cluster_test_last_committed'";s:10:"total_rows";i:1;s:4:"rows";a:1:{i:0;a:2:{s:7:"Counter";s:27:"cluster_test_last_committed";s:5:"Value";s:1:"5";}}}i:9;a:3:{s:8:"sphinxql";s:57:"debug wait test status 5 like 'state' option 'timeout'=30";s:10:"total_rows";i:1;s:4:"rows";a:1:{i:0;a:2:{s:4:"name";s:5:"state";s:5:"Value";s:1:"5";}}}i:10;a:3:{s:8:"sphinxql";s:38:"SELECT * FROM test:rt1 order by id asc";s:10:"total_rows";i:2;s:4:"rows";a:2:{i:0;a:4:{s:2:"id";s:1:"1";s:3:"idd";s:2:"11";s:1:"s";s:9:"another 1";s:5:"attrs";s:62:"{"phones":{"n":[971526025257],"t":[1]},"mails":[],"version":1}";}i:1;a:4:{s:2:"id";s:1:"2";s:3:"idd";s:1:"2";s:1:"s";s:5:"other";s:5:"attrs";s:62:"{"phones":{"n":[966549463121],"t":[1]},"mails":[],"version":6}";}}}i:11;a:2:{s:8:"sphinxql";s:27:"ALTER CLUSTER test DROP rt1";s:14:"total_affected";i:0;}i:12;a:2:{s:8:"sphinxql";s:14:"DROP TABLE rt1";s:14:"total_affected";i:0;}}}

test/test_456/test.xml

@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="utf-8"?>
+<test>
+<name>replication of update with blobs</name>
+
+<requires>
+	<replication/>
+	<non-windows/>
+	<force-rt/>
+	<heavy/>
+</requires>
+<skip_indexer/>
+<num_agents>2</num_agents>
+
+<config>
+
+<agent0>
+searchd
+{
+	<searchd_settings/>
+	data_dir = <data_path path="data0"/>
+	server_id = <agent_id/>
+}
+</agent0>
+<agent1>
+searchd
+{
+	<searchd_settings/>
+	data_dir = <data_path path="data1"/>
+	server_id = <agent_id/>
+}
+</agent1>
+
+</config>
+
+
+<queries>
+
+<!-- request to node 0 -->
+<sphinxql d="0">
+create table rt1 (id bigint, attrs json, idd integer, content text indexed, s string) rt_mem_limit='12M';
+create cluster test;
+alter cluster test add rt1;
+insert into test:rt1 (id, content, attrs, idd, s) values (1, 'text content first object', '{"phones":{"n":[971526025257],"t":[1]},"mails":[],"version":1}', 11, 'some 1');
+insert into test:rt1 (id, content, attrs, idd, s) values (2, 'text content second document', '{"phones":{"n":[971526025257],"t":[1]},"mails":[],"version":1}', 2, 'other');
+</sphinxql>
+
+<!-- request to node 1 -->
+<sphinxql d="1" cluster_connect="0">
+	join cluster test at '%addr_connect%';
+</sphinxql>
+
+<!-- stream to node 1 -->
+<!-- crashed node 0 due to missed blob replication -->
+<sphinxql d="1">
+UPDATE test:rt1 set attrs='{"phones":{"n":[966549463121],"t":[1]},"mails":[],"version":6}' where id=2;
+UPDATE test:rt1 set s='another 1' where id=1;
+show status like 'cluster_test_last_committed';
+</sphinxql>
+
+<!-- request to node 0 -->
+<sphinxql d="0">
+debug wait test status 5 like 'state' option 'timeout'=30;
+SELECT * FROM test:rt1 order by id asc;
+</sphinxql>
+
+<!-- cleanup -->
+<sphinxql d="0">
+	ALTER CLUSTER test DROP rt1;
+	DROP TABLE rt1;
+</sphinxql>
+
+</queries>
+
+</test>