数据库内核月报－ 2017/10 - MySQL · 引擎介绍 · Sphinx源码剖析(三) - 《数据库内核月报》

在本节中我会介绍Sphinx在构建索引之前做的一些事情，主要是从mysql拉取数据保存，然后分词排序保存到内存等等一系列的操作。下面是几个相关指令


1. sql_query = \
2. SELECT id, group_id, UNIX_TIMESTAMP(date_added) AS date_added, \
3. title, content \
4. FROM documents
5. sql_query_range = SELECT MIN(id),MAX(id) FROM documents
6. sql_range_step = 1000

其中sql_query是sphinx每次从mysql拉取数据的sql，而sql_query_range则是取得需要从mysql拉取的数据条目，而sql_rang_step则是表示每次从mysql拉取多少数据。sql_rang_range执行分两种情况，第一种是第一次拉取数据的时候，第二种是当当前的range数据读取完毕之后。

首先来看CSphSource_SQL::NextDocument函数，这个函数的主要作用是从mysql读取数据然后切分保存，首先我们来看读取数据这一部分，这里步骤很简单，就是执行对应的sql，然后判断当前range的数据是否读取完毕，如果读取完毕则继续执行sql_query_rang(RunQueryStep)。这里要注意的是，sphinx读取数据是一条一条的读取然后执行的.


1. do
2. {
3. // try to get next row
4. bool bGotRow = SqlFetchRow ();

6. // when the party's over...
7. while ( !bGotRow )
8. {
9. // is that an error?
10. if ( SqlIsError() )
11. {
12. sError.SetSprintf ( "sql_fetch_row: %s", SqlError() );
13. m_tDocInfo.m_uDocID = 1; // 0 means legal eof
14. return NULL;
15. }

17. // maybe we can do next step yet?
18. if ( !RunQueryStep ( m_tParams.m_sQuery.cstr(), sError ) )
19. {
20. // if there's a message, there's an error
21. // otherwise, we're just over
22. if ( !sError.IsEmpty() )
23. {
24. m_tDocInfo.m_uDocID = 1; // 0 means legal eof
25. return NULL;
26. }

28. } else
29. {
30. // step went fine; try to fetch
31. bGotRow = SqlFetchRow ();
32. continue;
33. }

35. SqlDismissResult ();

37. // ok, we're over
38. ARRAY_FOREACH ( i, m_tParams.m_dQueryPost )
39. {
40. if ( !SqlQuery ( m_tParams.m_dQueryPost[i].cstr() ) )
41. {
42. sphWarn ( "sql_query_post[%d]: error=%s, query=%s",
43. i, SqlError(), m_tParams.m_dQueryPost[i].cstr() );
44. break;
45. }
46. SqlDismissResult ();
47. }

49. m_tDocInfo.m_uDocID = 0; // 0 means legal eof
50. return NULL;
51. }

53. // get him!
54. m_tDocInfo.m_uDocID = VerifyID ( sphToDocid ( SqlColumn(0) ) );
55. m_uMaxFetchedID = Max ( m_uMaxFetchedID, m_tDocInfo.m_uDocID );
56. } while ( !m_tDocInfo.m_uDocID );

上面的代码我们可以看到一个很关键的字段m_uDocID,这个字段表示当前doc的id(因此数据库的表设计必须有这个id字段).

读取完毕数据之后，开始处理读取的数据，这里会按照字段来切分，主要是将对应的数据库字段保存到索引fielld


1. // split columns into fields and attrs
2. for ( int i=0; i<m_iPlainFieldsLength; i++ )
3. {
4. // get that field
5. #if USE_ZLIB
6. if ( m_dUnpack[i]!=SPH_UNPACK_NONE )
7. {
8. DWORD uUnpackedLen = 0;
9. m_dFields[i] = (BYTE*) SqlUnpackColumn ( i, uUnpackedLen, m_dUnpack[i] );
10. m_dFieldLengths[i] = (int)uUnpackedLen;
11. continue;
12. }
13. #endif
14. m_dFields[i] = (BYTE*) SqlColumn ( m_tSchema.m_dFields[i].m_iIndex );
15. m_dFieldLengths[i] = SqlColumnLength ( m_tSchema.m_dFields[i].m_iIndex );
16. }

紧接着就是处理attribute，后续我们会详细介绍attribute，现在我们只需要知道它是一个类似二级索引的东西(不进入全文索引).


1. switch ( tAttr.m_eAttrType )
2. {
3. case SPH_ATTR_STRING:
4. case SPH_ATTR_JSON:
5. // memorize string, fixup NULLs
6. m_dStrAttrs[i] = SqlColumn ( tAttr.m_iIndex );
7. if ( !m_dStrAttrs[i].cstr() )
8. m_dStrAttrs[i] = "";

10. m_tDocInfo.SetAttr ( tAttr.m_tLocator, 0 );
11. break;
12. ..................................
13. default:
14. // just store as uint by default
15. m_tDocInfo.SetAttr ( tAttr.m_tLocator, sphToDword ( SqlColumn ( tAttr.m_iIndex ) ) ); // FIXME? report conversion errors maybe?
16. break;
17. }

然后我们来看Sphinx如何处理得到的数据,核心代码在 RtIndex_t::AddDocument中，这个函数主要是用来分词(IterateHits中)然后保存数据到对应的数据结构,而核心的数据结构是RtAccum_t，也就是最终sphinx在写索引到文件之前，会将数据保存到这个数据结构，这里要注意一般来说sphinx会保存很多数据，然后最后一次性提交给索引引擎来处理.而索引引擎中处理的就是这个数据结构.因此最终会调用RtAccum_t::AddDocument.

这里需要注意两个地方，第一个是m_dAccum这个域，这个域是一个vector，而这个vector里面保存了CSphWordHit这个结构，我们来看这个结构的定义


1. struct CSphWordHit
2. {
3. SphDocID_t        m_uDocID;        ///< document ID
4. SphWordID_t        m_uWordID;        ///< word ID in current dictionary
5. Hitpos_t        m_uWordPos;        ///< word position in current document
6. };

可以看到其实这个结构也就是保存了对应分词的信息.

然后我们来看核心代码，这里主要是便利刚才从mysql得到的数据，去重然后保存数据.


1. int iHits = 0;
2. if ( pHits && pHits->Length() )
3. {
4. CSphWordHit tLastHit;
5. tLastHit.m_uDocID = 0;
6. tLastHit.m_uWordID = 0;
7. tLastHit.m_uWordPos = 0;

9. iHits = pHits->Length();
10. m_dAccum.Reserve ( m_dAccum.GetLength()+iHits );
11. for ( const CSphWordHit * pHit = pHits->First(); pHit<=pHits->Last(); pHit++ )
12. {
13. // ignore duplicate hits
14. if ( pHit->m_uDocID==tLastHit.m_uDocID && pHit->m_uWordID==tLastHit.m_uWordID && pHit->m_uWordPos==tLastHit.m_uWordPos )
15. continue;

17. // update field lengths
18. if ( pFieldLens && HITMAN::GetField ( pHit->m_uWordPos )!=HITMAN::GetField ( tLastHit.m_uWordPos ) )
19. pFieldLens [ HITMAN::GetField ( tLastHit.m_uWordPos ) ] = HITMAN::GetPos ( tLastHit.m_uWordPos );

21. // accumulate
22. m_dAccum.Add ( *pHit );
23. tLastHit = *pHit;
24. }
25. if ( pFieldLens )
26. pFieldLens [ HITMAN::GetField ( tLastHit.m_uWordPos ) ] = HITMAN::GetPos ( tLastHit.m_uWordPos );
27. }

做完上面这些事情之后，就需要提交数据给索引处理引擎了，这里核心的代码都是在RtIndex_t::Commit中.

这个函数主要做两个事情，第一个提取出前面我们构造好的RtAccum_t,然后对于所有的doc进行排序，创建segment，也就是对应的索引块(ram chunk)，最后调用CommitReplayable来提交ram chunk到磁盘.

其实可以这么理解，保存在内存中的索引也就是segment,然后当内存的大小到达限制后就会刷新内存中的索引到磁盘.


1. void RtIndex_t::Commit ( int * pDeleted, ISphRtAccum * pAccExt )
2. {
3. assert ( g_bRTChangesAllowed );
4. MEMORY ( MEM_INDEX_RT );

6. RtAccum_t * pAcc = AcquireAccum ( NULL, pAccExt, true );
7. if ( !pAcc )
8. return;

10. ...................................
11. pAcc->Sort();

13. RtSegment_t * pNewSeg = pAcc->CreateSegment ( m_tSchema.GetRowSize(), m_iWordsCheckpoint );
14. .............................................

16. // now on to the stuff that needs locking and recovery
17. CommitReplayable ( pNewSeg, pAcc->m_dAccumKlist, pDeleted );
18. ......................................
19. }

然后我们来看RtAccum_t::CreateSegment函数，这个函数用来将分词好的数据保存到ram chunk，这里需要注意两个数据结构分别是RtDoc_t和RtWord_t,这两个数据结构分别表示doc信息和分词信息.

结构很简单，后面的注释都很详细


1. template < typename DOCID = SphDocID_t >
2. struct RtDoc_T
3. {
4. DOCID                        m_uDocID;    ///< my document id
5. DWORD                        m_uDocFields;    ///< fields mask
6. DWORD                        m_uHits;    ///< hit count
7. DWORD                        m_uHit;        ///< either index into segment hits, or the only hit itself (if hit count is 1)
8. };

10. template < typename WORDID=SphWordID_t >
11. struct RtWord_T
12. {
13. union
14. {
15. WORDID                    m_uWordID;    ///< my keyword id
16. const BYTE *            m_sWord;
17. };
18. DWORD                        m_uDocs;    ///< document count (for stats and/or BM25)
19. DWORD                        m_uHits;    ///< hit count (for stats and/or BM25)
20. DWORD                        m_uDoc;        ///< index into segment docs
21. };

然后来看代码，首先是初始化对应的写结构,可以看到都是会写到我们创建好的segment中.


1. RtDocWriter_t tOutDoc ( pSeg );
2. RtWordWriter_t tOutWord ( pSeg, m_bKeywordDict, iWordsCheckpoint );
3. RtHitWriter_t tOutHit ( pSeg );

然后就是写数据了，这里主要是做一个聚合，也就是将相同的keyword对应的属性聚合起来.


1. ARRAY_FOREACH ( i, m_dAccum )
2. {
3. .......................................
4. // new keyword; flush current keyword
5. if ( tHit.m_uWordID!=tWord.m_uWordID )
6. {
7. tOutDoc.ZipRestart ();
8. if ( tWord.m_uWordID )
9. {
10. if ( m_bKeywordDict )
11. {
12. const BYTE * pPackedWord = pPacketBase + tWord.m_uWordID;
13. assert ( pPackedWord[0] && pPackedWord[0]+1<m_pDictRt->GetPackedLen() );
14. tWord.m_sWord = pPackedWord;
15. }
16. tOutWord.ZipWord ( tWord );
17. }

19. tWord.m_uWordID = tHit.m_uWordID;
20. tWord.m_uDocs = 0;
21. tWord.m_uHits = 0;
22. tWord.m_uDoc = tOutDoc.ZipDocPtr();
23. uPrevHit = EMPTY_HIT;
24. }
25. ..................
26. }

这次就分析到这里，下次我们将会分析最核心的部分就是Sphinx如何刷新数据到磁盘.

原文：http://mysql.taobao.org/monthly/2017/10/06/