Merge branch '1570_1572_1577_nested_queries_core' into 1574_subQuery_…

…go_interface

.clang-format

@@ -6,7 +6,7 @@ AlignAfterOpenBracket: Align
 AlignConsecutiveAssignments: false
 AlignConsecutiveDeclarations: false
 AlignEscapedNewlinesLeft: true
-AlignOperands:   true
+AlignOperands: true
 AlignTrailingComments: true
 AllowAllParametersOfDeclarationOnNextLine: true
 AllowShortBlocksOnASingleLine: false
@@ -20,7 +20,7 @@ AlwaysBreakBeforeMultilineStrings: true
 AlwaysBreakTemplateDeclarations: true
 BinPackArguments: true
 BinPackParameters: true
-BraceWrapping:   
+BraceWrapping:
   AfterClass:      false
   AfterControlStatement: false
   AfterEnum:       false
@@ -45,7 +45,6 @@ ConstructorInitializerIndentWidth: 4
 ContinuationIndentWidth: 4
 Cpp11BracedListStyle: true
 DerivePointerAlignment: true
-PointerAlignment: Left
 DisableFormat:   false
 ExperimentalAutoDetectBinPacking: false
 ForEachMacros:   [ foreach, Q_FOREACH, BOOST_FOREACH ]

bindings/consts.go

@@ -83,6 +83,9 @@ const (
 	QueryUpdateFieldV2          = 25
 	QueryBetweenFieldsCondition = 26
 	QueryAlwaysFalseCondition   = 27
+	QueryAlwaysTrueCondition    = 28
+	QuerySubQueryCondition      = 29
+	QueryFieldSubQueryCondition = 30
 
 	LeftJoin    = 0
 	InnerJoin   = 1

cpp_src/CMakeLists.txt

@@ -53,25 +53,28 @@ include (TargetArch)
 target_architecture(COMPILER_TARGET_ARCH)
 
 # Configure compile options
-string(REPLACE "-DNDEBUG" "" CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO}")
-string(REPLACE "-O2" "-O3" CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}")
-string(REPLACE "-O2" "-O3" CMAKE_C_FLAGS "${CMAKE_C_FLAGS}")
-if (NOT ${COMPILER_TARGET_ARCH} STREQUAL "e2k")
-    string(REPLACE "-g" "-g1" CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO}")
-else()
-    string(REPLACE "-g" "-g0" CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO}")
-endif()
-
+if (MSVC)
+  set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O2 -g1")
+  set(CMAKE_C_FLAGS_RELWITHDEBINFO "-O2 -g1")
+  set(CMAKE_CXX_FLAGS_RELEASE "-O2 -DNDEBUG")
+  set(CMAKE_C_FLAGS_RELEASE "-O2 -DNDEBUG")
+else ()
+  set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O3 -g1")
+  set(CMAKE_C_FLAGS_RELWITHDEBINFO "-O3 -g1")
+  set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG")
+  set(CMAKE_C_FLAGS_RELEASE "-O3 -DNDEBUG")
+endif ()
 if (${COMPILER_TARGET_ARCH} STREQUAL "e2k")
+  set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O3 -g0")
   add_definitions(-D__E2K__)
   add_definitions(-D__LCC__)
-endif()
+endif ()
 
 if (NOT MSVC AND NOT APPLE)
   check_linker_flag (-gz cxx_linker_supports_gz)
   if (cxx_linker_supports_gz) 
-     set (CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO "${CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO} -gz")
-   endif ()
+    set (CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO "${CMAKE_EXE_LINKER_FLAGS_RELWITHDEBINFO} -gz")
+  endif ()
 endif ()
 
 if (MSVC)

cpp_src/client/coroqueryresults.cc

@@ -212,9 +212,10 @@ Item CoroQueryResults::Iterator::GetItem() {
 		if (err.ok()) {
 			return item;
 		}
+		return Item();
 	} catch (const Error &) {
+		return Item();
 	}
-	return Item();
 }
 
 int64_t CoroQueryResults::Iterator::GetLSN() {

cpp_src/client/cororpcclient.cc

@@ -158,7 +158,7 @@ Error CoroRPCClient::modifyItem(std::string_view nsName, Item& item, int mode, s
 			}
 			CoroQueryResults qr;
 			InternalRdxContext ctxCompl = ctx.WithCompletion(nullptr);
-			auto ret = selectImpl(Query(std::string(nsName)).Limit(0), qr, netTimeout, ctxCompl);
+			auto ret = selectImpl(Query(nsName).Limit(0), qr, netTimeout, ctxCompl);
 			if (ret.code() == errTimeout) {
 				return Error(errTimeout, "Request timeout");
 			}
@@ -242,7 +242,7 @@ Error CoroRPCClient::Delete(const Query& query, CoroQueryResults& result, const
 	query.Serialize(ser);
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = CoroQueryResults(&conn_, std::move(nsArray), 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -263,7 +263,7 @@ Error CoroRPCClient::Update(const Query& query, CoroQueryResults& result, const
 	query.Serialize(ser);
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = CoroQueryResults(&conn_, std::move(nsArray), 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -313,10 +313,10 @@ Error CoroRPCClient::selectImpl(const Query& query, CoroQueryResults& result, se
 	WrSerializer qser, pser;
 	int flags = result.fetchFlags_ ? result.fetchFlags_ : (kResultsWithPayloadTypes | kResultsCJson);
 	flags |= kResultsSupportIdleTimeout;
-	bool hasJoins = !query.joinQueries_.empty();
+	bool hasJoins = !query.GetJoinQueries().empty();
 	if (!hasJoins) {
-		for (auto& mq : query.mergeQueries_) {
-			if (!mq.joinQueries_.empty()) {
+		for (auto& mq : query.GetMergeQueries()) {
+			if (!mq.GetJoinQueries().empty()) {
 				hasJoins = true;
 				break;
 			}
@@ -328,7 +328,7 @@ Error CoroRPCClient::selectImpl(const Query& query, CoroQueryResults& result, se
 	}
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		vers.push_back(ns->tagsMatcher_.version() ^ ns->tagsMatcher_.stateToken());

cpp_src/client/queryresults.cc

@@ -250,9 +250,10 @@ Item QueryResults::Iterator::GetItem() {
 		if (err.ok()) {
 			return item;
 		}
+		return Item();
 	} catch (const Error &) {
+		return Item();
 	}
-	return Item();
 }
 
 int64_t QueryResults::Iterator::GetLSN() {

cpp_src/client/resultserializer.cc

@@ -50,6 +50,8 @@ void ResultSerializer::GetRawQueryParams(ResultSerializer::QueryParams& ret, con
 			case QueryResultExplain:
 				ret.explainResults = std::string(data);
 				break;
+			default:
+				throw Error(errLogic, "Unexpected Query tag: %d", tag);
 		}
 	}
 }

cpp_src/client/rpcclient.cc

@@ -3,6 +3,7 @@
 #include <functional>
 #include "client/itemimpl.h"
 #include "core/namespacedef.h"
+#include "core/schema.h"
 #include "gason/gason.h"
 #include "tools/cpucheck.h"
 #include "tools/errors.h"
@@ -98,6 +99,7 @@ void RPCClient::run(size_t thIdx) {
 
 	workers_[thIdx].stop_.start();
 	delayedUpdates_.clear();
+	serialDelays_ = 0;
 
 	for (size_t i = thIdx; int(i) < config_.ConnPoolSize; i += config_.WorkerThreads) {
 		connections_[i].reset(new cproto::ClientConnection(workers_[thIdx].loop_, &connectData_,
@@ -381,7 +383,7 @@ Error RPCClient::Delete(const Query& query, QueryResults& result, const Internal
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -408,7 +410,7 @@ Error RPCClient::Update(const Query& query, QueryResults& result, const Internal
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -469,10 +471,10 @@ Error RPCClient::selectImpl(const Query& query, QueryResults& result, cproto::Cl
 							const InternalRdxContext& ctx) {
 	WrSerializer qser, pser;
 	int flags = result.fetchFlags_ ? result.fetchFlags_ : (kResultsWithPayloadTypes | kResultsCJson);
-	bool hasJoins = !query.joinQueries_.empty();
+	bool hasJoins = !query.GetJoinQueries().empty();
 	if (!hasJoins) {
-		for (auto& mq : query.mergeQueries_) {
-			if (!mq.joinQueries_.empty()) {
+		for (auto& mq : query.GetMergeQueries()) {
+			if (!mq.GetJoinQueries().empty()) {
 				hasJoins = true;
 				break;
 			}
@@ -484,7 +486,7 @@ Error RPCClient::selectImpl(const Query& query, QueryResults& result, cproto::Cl
 	}
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		shared_lock<shared_timed_mutex> lck(ns->lck_);
@@ -717,21 +719,39 @@ void RPCClient::onUpdates(net::cproto::RPCAnswer& ans, cproto::ClientConnection*
 			// If tagsMatcher has been updated but there is no bundled tagsMatcher in cjson
 			// then we need to ask server to send tagsMatcher.
 
+			++serialDelays_;
 			// Delay this update and all the further updates until we get responce from server.
 			ans.EnsureHold();
 			delayedUpdates_.emplace_back(std::move(ans));
 
 			QueryResults* qr = new QueryResults;
 			Select(Query(std::string(nsName)).Limit(0), *qr,
-				   InternalRdxContext(nullptr,
-									  [=](const Error& err) {
-										  delete qr;
-										  // If there are delayed updates then send them to client
-										  auto uq = std::move(delayedUpdates_);
-										  delayedUpdates_.clear();
-										  if (err.ok())
-											  for (auto& a1 : uq) onUpdates(a1, conn);
-									  }),
+				   InternalRdxContext(
+					   nullptr,
+					   [this, qr, conn](const Error& err) {
+						   delete qr;
+						   // If there are delayed updates then send them to client
+						   auto uq = std::move(delayedUpdates_);
+						   delayedUpdates_.clear();
+
+						   if (!err.ok() || serialDelays_ > 1) {
+							   // This update was already dealyed, but was not able to synchronize tagsmatcher.
+							   // Such situation usually means, that master's namespace was recreated and must be synchronized via force
+							   // sync.
+							   // Current fix is suboptimal and in some cases even incorrect (but still better, than previous
+							   // implementation) - proper fix requires some versioning info about namespaces, which exists
+							   // in v4 only
+							   std::string_view nsName(std::string_view(uq.front().GetArgs(1)[1]));
+							   logPrintf(
+								   LogWarning,
+								   "[repl:%s] Unable to sync tags matcher via online-replication (err: '%s'). Calling UpdatesLost fallback",
+								   nsName, err.what());
+							   serialDelays_ = 0;
+							   observers_.OnUpdatesLost(nsName);
+						   } else {
+							   for (auto& a1 : uq) onUpdates(a1, conn);
+						   }
+					   }),
 				   conn);
 			return;
 		} else {
@@ -748,7 +768,17 @@ void RPCClient::onUpdates(net::cproto::RPCAnswer& ans, cproto::ClientConnection*
 				ns->tagsMatcher_.deserialize(rdser, wrec.itemModify.tmVersion, ns->tagsMatcher_.stateToken());
 			}
 		}
+	} else if (wrec.type == WalTagsMatcher) {
+		TagsMatcher tm;
+		Serializer ser(wrec.data.data(), wrec.data.size());
+		const auto version = ser.GetVarint();
+		const auto stateToken = ser.GetVarint();
+		tm.deserialize(ser, version, stateToken);
+		auto ns = getNamespace(nsName);
+		std::lock_guard lck(ns->lck_);
+		ns->tagsMatcher_ = std::move(tm);
 	}
+	serialDelays_ = 0;
 	observers_.OnWALUpdate(LSNPair(lsn, originLSN), nsName, wrec);
 }
 

cpp_src/client/rpcclient.h

@@ -90,6 +90,7 @@ class RPCClient {
 		ev::async stop_;
 		std::atomic_bool running;
 	};
+
 	Error selectImpl(std::string_view query, QueryResults &result, cproto::ClientConnection *, seconds netTimeout,
 					 const InternalRdxContext &ctx);
 	Error selectImpl(const Query &query, QueryResults &result, cproto::ClientConnection *, seconds netTimeout,
@@ -122,6 +123,7 @@ class RPCClient {
 	UpdatesObservers observers_;
 	std::atomic<net::cproto::ClientConnection *> updatesConn_;
 	std::vector<net::cproto::RPCAnswer> delayedUpdates_;
+	uint64_t serialDelays_ = 0;
 	cproto::ClientConnection::ConnectData connectData_;
 };
 

cpp_src/client/rpcclientmock.cc

@@ -37,7 +37,7 @@ Error RPCClientMock::Delete(const Query& query, QueryResults& result, const Inte
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -69,7 +69,7 @@ Error RPCClientMock::Update(const Query& query, QueryResults& result, const Inte
 	auto conn = getConn();
 
 	NsArray nsArray;
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 
 	result = QueryResults(conn, std::move(nsArray), nullptr, 0, config_.FetchAmount, config_.RequestTimeout);
 
@@ -289,10 +289,10 @@ Error RPCClientMock::selectImpl(const Query& query, QueryResults& result, cproto
 		flags = result.fetchFlags_ ? result.fetchFlags_ : (kResultsWithPayloadTypes | kResultsCJson);
 	}
 
-	bool hasJoins = !query.joinQueries_.empty();
+	bool hasJoins = !query.GetJoinQueries().empty();
 	if (!hasJoins) {
-		for (auto& mq : query.mergeQueries_) {
-			if (!mq.joinQueries_.empty()) {
+		for (auto& mq : query.GetMergeQueries()) {
+			if (!mq.GetJoinQueries().empty()) {
 				hasJoins = true;
 				break;
 			}
@@ -308,7 +308,7 @@ Error RPCClientMock::selectImpl(const Query& query, QueryResults& result, cproto
 
 	NsArray nsArray;
 	query.Serialize(qser);
-	query.WalkNested(true, true, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
+	query.WalkNested(true, true, false, [this, &nsArray](const Query& q) { nsArray.push_back(getNamespace(q.NsName())); });
 	h_vector<int32_t, 4> vers;
 	for (auto& ns : nsArray) {
 		shared_lock<shared_timed_mutex> lck(ns->lck_);