Skip to content
Permalink
Browse files
fixed #888 query uid at percolate index to be BIGINT attribute type
  • Loading branch information
tomatolog committed Jun 11, 2019
1 parent bb9f560 commit b08147ee24712b5eaa51f261626b5d16c6242fdd
@@ -1366,7 +1366,8 @@ server_id
~~~~~~~~~

Integer number that serves as server identificator used as seed to generate an unique short UUID for nodes that are part of a replication cluster.
The server_id must be unique across the nodes of a cluster. If server_id is not set, MAC address or a random number will be used as seed for the short UUID.
The server_id must be unique across the nodes of a cluster and in range from 0 to 127.
If server_id is not set, MAC address or a random number will be used as seed for the short UUID.

Example:

@@ -25,7 +25,7 @@ struct StoredQueryDesc_t

CSphString m_sQuery;
CSphString m_sTags;
uint64_t m_uQUID = 0;
int64_t m_iQUID = 0;
bool m_bQL = true;
};

@@ -59,7 +59,7 @@ struct ReplicationCommand_t
CSphScopedPtr<StoredQuery_i> m_pStored { nullptr };

// delete
CSphVector<uint64_t> m_dDeleteQueries;
CSphVector<int64_t> m_dDeleteQueries;
CSphString m_sDeleteTags;

// truncate
@@ -98,8 +98,8 @@ TEST_F ( PQ_merge, JustOneResult )

// ensure we take correct matches
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength(), dResult.m_iQueriesMatched );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_uQID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_uQID, 101 );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_iQUID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_iQUID, 101 );
}

// one result, but with list of docs
@@ -129,8 +129,8 @@ TEST_F ( PQ_merge, JustOneResultWithDocs )

// ensure we take correct matches
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_uQID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_uQID, 101 );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_iQUID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_iQUID, 101 );
}

// one result, but with times
@@ -161,8 +161,8 @@ TEST_F ( PQ_merge, JustOneResultVerbose )

// ensure we take correct matches
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_uQID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_uQID, 101 );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_iQUID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_iQUID, 101 );
}

// one result, with everything
@@ -197,8 +197,8 @@ TEST_F ( PQ_merge, JustOneResultWithEverything )

// ensure we take correct matches
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_uQID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_uQID, 101 );
ASSERT_EQ ( dResult.m_dQueryDesc[0].m_iQUID, 100 );
ASSERT_EQ ( dResult.m_dQueryDesc[1].m_iQUID, 101 );
}

// full result, with everything
@@ -230,7 +230,7 @@ TEST_F ( PQ_merge, FullResult )
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
int j = 0;
for ( auto qid : { 100, 101, 102, 103, 180, 190 } )
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_uQID, qid );
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_iQUID, qid );
}

// full result, with everything
@@ -266,7 +266,7 @@ TEST_F ( PQ_merge, FullResultWithDocs )
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
j = 0;
for ( auto qid : { 100, 101, 102, 103, 180, 190 } )
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_uQID, qid );
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_iQUID, qid );
}

// full result, with everything
@@ -302,7 +302,7 @@ TEST_F ( PQ_merge, FullResultVerbose )
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
j = 0;
for ( auto qid : { 100, 101, 102, 103, 180, 190 } )
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_uQID, qid );
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_iQUID, qid );
}

// full result, with everything
@@ -341,5 +341,5 @@ TEST_F ( PQ_merge, FullResultWithEverything )
ASSERT_EQ ( dResult.m_dQueryDesc.GetLength (), dResult.m_iQueriesMatched );
j=0;
for ( auto qid : { 100, 101, 102, 103, 180, 190 } )
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_uQID, qid );
ASSERT_EQ ( dResult.m_dQueryDesc[j++].m_iQUID, qid );
}
@@ -12769,7 +12769,7 @@ struct PqReplyParser_t : public IReplyParser_t
dResult.m_dQueryDesc.Reset ( iRows );
for ( auto &tDesc : dResult.m_dQueryDesc )
{
tDesc.m_uQID = tReq.GetUint64 ();
tDesc.m_iQUID = tReq.GetUint64 ();
if ( bDumpDocs )
{
int iCount = tReq.GetInt ();
@@ -12857,7 +12857,7 @@ static void SendAPIPercolateReply ( CachedOutputBuffer_c &tOut, const CPqResult
tOut.SendInt ( tRes.m_dQueryDesc.GetLength () );
for ( const auto &tDesc : tRes.m_dQueryDesc )
{
tOut.SendUint64 ( tDesc.m_uQID );
tOut.SendUint64 ( tDesc.m_iQUID );
if ( bDumpDocs )
{
// document count + document id(s)
@@ -12935,7 +12935,7 @@ static void SendMysqlPercolateReply ( SqlRowBuffer_c &tOut
iColumns += 3;
tOut.HeadBegin ( iColumns );

tOut.HeadColumn ( "id", MYSQL_COL_LONGLONG, MYSQL_COL_UNSIGNED_FLAG );
tOut.HeadColumn ( "id", MYSQL_COL_LONGLONG, 0 );
if ( bDumpDocs )
tOut.HeadColumn ( "documents", MYSQL_COL_STRING );
if ( bQuery )
@@ -12954,7 +12954,7 @@ static void SendMysqlPercolateReply ( SqlRowBuffer_c &tOut
StringBuilder_c sDocs;
for ( const auto &tDesc : tRes.m_dQueryDesc )
{
tOut.PutNumAsString ( tDesc.m_uQID );
tOut.PutNumAsString ( tDesc.m_iQUID );
if ( bDumpDocs )
{
sDocs.StartBlock ( "," );
@@ -13925,7 +13925,7 @@ void sphHandleMysqlInsert ( StmtErrorReporter_i & tOut, SqlStmt_t & tStmt, bool
PercolateQueryArgs_t tArgs ( dFilters, dFilterTree );
tArgs.m_sQuery = dStrings[0];
tArgs.m_sTags = dStrings[1];
tArgs.m_uQUID = tDoc.GetAttr(tIdLoc);
tArgs.m_iQUID = tDoc.GetAttr(tIdLoc);
tArgs.m_bReplace = bReplace;
tArgs.m_bQL = true;

@@ -13940,7 +13940,7 @@ void sphHandleMysqlInsert ( StmtErrorReporter_i & tOut, SqlStmt_t & tStmt, bool
pCmd->m_sCluster = tStmt.m_sCluster;
pCmd->m_pStored = pStored;

dIds.Add ( pStored->m_uQUID );
dIds.Add ( pStored->m_iQUID );
}
} else
{
@@ -24121,6 +24121,12 @@ void ConfigureSearchd ( const CSphConfig & hConf, bool bOptPIDFile )
{
g_iServerID = hSearchd.GetInt ( "server_id", g_iServerID );
g_bServerID = true;
const int iServerMask = 0x7f;
if ( g_iServerID>iServerMask )
{
g_iServerID &= iServerMask;
sphWarning ( "server_id out of range 0 - 127, clamped to %d", g_iServerID );
}
}

//////////////////////////////////////////////////
@@ -24322,8 +24328,9 @@ void OpenDaemonLog ( const CSphConfigSection & hSearchd, bool bCloseIfOpened=fal
g_bLogTty = isatty ( g_iLogFile )!=0;
}

static void SetUuidShort ( bool bTestMode )
static void SetUidShort ( bool bTestMode )
{
const int iServerMask = 0x7f;
int iServerId = g_iServerID;
uint64_t uStartedSec = 0;

@@ -24345,6 +24352,7 @@ static void SetUuidShort ( bool bTestMode )
}
// fold MAC into 1 byte
iServerId = Pearson8 ( (const BYTE *)sMAC.cstr(), sMAC.Length() );
iServerId &= iServerMask;
}

// start time Unix timestamp as middle part of counter
@@ -24359,7 +24367,7 @@ static void SetUuidShort ( bool bTestMode )
uStartedSec = 100000;
iServerId = g_iServerID;
}
UuidShortSetup ( iServerId, (int)uStartedSec );
UidShortSetup ( iServerId, (int)uStartedSec );
}

int WINAPI ServiceMain ( int argc, char **argv ) REQUIRES (!MainThread)
@@ -24991,7 +24999,7 @@ int WINAPI ServiceMain ( int argc, char **argv ) REQUIRES (!MainThread)
sphRTConfigure ( hSearchd, bTestMode );
SetPercolateQueryParserFactory ( PercolateQueryParserFactory );
SetPercolateThreads ( g_iDistThreads );
SetUuidShort ( bTestMode );
SetUidShort ( bTestMode );

if ( bOptPIDFile )
{
@@ -1103,7 +1103,7 @@ static void EncodePercolateMatchResult ( const PercolateMatchResult_t & tRes, co
for ( const auto& tDesc : tRes.m_dQueryDesc )
{
ScopedComma_c sQueryComma ( tOut, ",","{"," }");
tOut.Sprintf ( R"("_index":"%s","_type":"doc","_id":"%U","_score":"1")", sIndex.cstr(), tDesc.m_uQID );
tOut.Sprintf ( R"("_index":"%s","_type":"doc","_id":"%U","_score":"1")", sIndex.cstr(), tDesc.m_iQUID );
if ( !tDesc.m_bQL )
tOut.Sprintf ( R"("_source":{"query":%s})", tDesc.m_sQuery.cstr () );
else
@@ -1193,12 +1193,12 @@ bool HttpHandlerPQ_c::DoCallPQ ( const CSphString & sIndex, const JsonObj_c & tP
}


static void EncodePercolateQueryResult ( bool bReplace, const CSphString & sIndex, uint64_t uID, StringBuilder_c & tOut )
static void EncodePercolateQueryResult ( bool bReplace, const CSphString & sIndex, int64_t iID, StringBuilder_c & tOut )
{
if ( bReplace )
tOut.Sprintf (R"({"index":"%s","type":"doc","_id":"%U","result":"updated","forced_refresh":true})", sIndex.cstr(), uID);
tOut.Sprintf (R"({"index":"%s","type":"doc","_id":"%U","result":"updated","forced_refresh":true})", sIndex.cstr(), iID);
else
tOut.Sprintf ( R"({"index":"%s","type":"doc","_id":"%U","result":"created"})", sIndex.cstr (), uID );
tOut.Sprintf ( R"({"index":"%s","type":"doc","_id":"%U","result":"created"})", sIndex.cstr (), iID );
}


@@ -1232,9 +1232,9 @@ bool HttpHandlerPQ_c::InsertOrReplaceQuery ( const CSphString& sIndex, const Jso
return false;
}

uint64_t uID = 0;
int64_t iID = 0;
if ( pUID && !pUID->IsEmpty() )
uID = strtoull ( pUID->cstr(), nullptr, 10 );
iID = strtoll ( pUID->cstr(), nullptr, 10 );

JsonObj_c tTagsArray = tRoot.GetArrayItem ( "tags", sError, true );
if ( !sError.IsEmpty() )
@@ -1293,7 +1293,7 @@ bool HttpHandlerPQ_c::InsertOrReplaceQuery ( const CSphString& sIndex, const Jso
PercolateQueryArgs_t tArgs ( dFilters, dFilterTree );
tArgs.m_sQuery = sQuery;
tArgs.m_sTags = sTags.cstr();
tArgs.m_uQUID = uID;
tArgs.m_iQUID = iID;
tArgs.m_bReplace = bReplace;
tArgs.m_bQL = bQueryQL;

@@ -1308,7 +1308,7 @@ bool HttpHandlerPQ_c::InsertOrReplaceQuery ( const CSphString& sIndex, const Jso
pCmd->m_sIndex = sIndex;
pCmd->m_pStored = pStored;
// refresh query's UID for reply as it might be auto-generated
uID = pStored->m_uQUID;
iID = pStored->m_iQUID;

bOk = HandleCmdReplicate ( tAcc, sError, nullptr );
}
@@ -1319,7 +1319,7 @@ bool HttpHandlerPQ_c::InsertOrReplaceQuery ( const CSphString& sIndex, const Jso
else
{
StringBuilder_c sRes;
EncodePercolateQueryResult ( bReplace, sIndex, uID, sRes );
EncodePercolateQueryResult ( bReplace, sIndex, iID, sRes );
BuildReply ( sRes, SPH_HTTP_STATUS_200 );
}

@@ -1500,7 +1500,7 @@ bool ParseCmdReplicated ( const BYTE * pData, int iLen, bool bIsolated, const CS

StoredQueryDesc_t tPQ;
LoadStoredQuery ( pRequest, iRequestLen, tPQ );
sphLogDebugRpl ( "pq-add, index '%s', uid " UINT64_FMT " query %s", pCmd->m_sIndex.cstr(), tPQ.m_uQUID, tPQ.m_sQuery.cstr() );
sphLogDebugRpl ( "pq-add, index '%s', uid " INT64_FMT " query %s", pCmd->m_sIndex.cstr(), tPQ.m_iQUID, tPQ.m_sQuery.cstr() );

CSphString sError;
PercolateQueryArgs_t tArgs ( tPQ );
@@ -1614,7 +1614,7 @@ bool HandleCmdReplicated ( RtAccum_t & tAcc )
auto * pIndex = (RtIndex_i * ) pRDesc->m_pIndex;
assert ( tCmd.m_eCommand!=ReplicationCommand_e::TRUNCATE );
sphLogDebugRpl ( "pq-commit, index '%s', uid " INT64_FMT ", queries %d, tags %s",
tCmd.m_sIndex.cstr(), ( tCmd.m_pStored ? tCmd.m_pStored->m_uQUID : int64_t(0) ),
tCmd.m_sIndex.cstr(), ( tCmd.m_pStored ? tCmd.m_pStored->m_iQUID : int64_t(0) ),
tCmd.m_dDeleteQueries.GetLength(), tCmd.m_sDeleteTags.scstr() );
pIndex->Commit ( nullptr, &tAcc );

@@ -1703,7 +1703,7 @@ bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, int * pDeletedC
assert ( tCmd.m_pStored );
SaveStoredQuery ( *tCmd.m_pStored.Ptr(), dBufQueries );

uQueryHash = sphFNV64 ( &tCmd.m_pStored->m_uQUID, sizeof(tCmd.m_pStored->m_uQUID), uQueryHash );
uQueryHash = sphFNV64 ( &tCmd.m_pStored->m_iQUID, sizeof(tCmd.m_pStored->m_iQUID), uQueryHash );
break;

case ReplicationCommand_e::PQUERY_DELETE:
@@ -1583,7 +1583,7 @@ class ISphBinlog : ISphNoncopyable
virtual bool IsActive () = 0;

virtual void BinlogPqAdd ( int64_t * pTID, const char * sIndexName, const StoredQueryDesc_t & tStored ) = 0;
virtual void BinlogPqDelete ( int64_t * pTID, const char * sIndexName, const uint64_t * pQueries, int iCount, const char * sTags ) = 0;
virtual void BinlogPqDelete ( int64_t * pTID, const char * sIndexName, const int64_t * pQueries, int iCount, const char * sTags ) = 0;
};

//////////////////////////////////////////////////////////////////////////

0 comments on commit b08147e

Please sign in to comment.