run splitVector failed[cannot Decode to nil value], give up parallel fetching

[pengzhenyi2015]

关联 issue#832
下面的 splitVector 代码写的有问题： res 只声明，没有初始化，导致 decode to nil value

[pengzhenyi2015]

再补充一下完整信息。

操作系统： CentOS7.6
mongo-shake 版本: v2.8.6 (测试了 github 上的 release 二进制，以及自己用 go1.22.10 编译的版本)
MongoDB 内核版本：r4.2.25 和 r4.0.3

错误信息：

1. 报错 cannot Decode to nil value， 然后 give parallel fetching
2. 报错 run splitVector return empty result， 然后 give parallel fetching

根据代码分析，错误应该集中在 DocumentSplitter::Run 方法。
先处理了第1个错误，推测是 var res bson.M 这个语句有问题，没有分配内存，于是改成 res := bson.M{} 解决。
但是随后就推进到了第 2 个错误，推测是 []interface{} 的解析有问题。因此，将 splitVector 结果处理的逻辑进行了一次大的修改。
然后才能正常的执行 parallel fetching...

下面是我修改的代码，供参考：

diff --git a/collector/docsyncer/doc_reader.go b/collector/docsyncer/doc_reader.go
index cebaebe..4e3d117 100644
--- a/collector/docsyncer/doc_reader.go
+++ b/collector/docsyncer/doc_reader.go
@@ -91,6 +91,10 @@ func (ds *DocumentSplitter) String() string {
                utils.BlockMongoUrlPassword(ds.src, "***"), ds.ns, ds.count, ds.pieceByteSize/utils.MB, ds.pieceNumber)
 }
 
+type SplitVectorResult struct {
+       SplitKeys []bson.Raw `bson:"splitKeys"` // 分裂点数组
+}
+
 // TODO, need add retry
 func (ds *DocumentSplitter) Run() error {
        // close channel
@@ -106,13 +110,13 @@ func (ds *DocumentSplitter) Run() error {
 
        LOG.Info("splitter[%s] enable split, waiting splitVector return...", ds)
 
-       var res bson.M
+       res := SplitVectorResult{}
        err := ds.client.Client.Database(ds.ns.Database).RunCommand(nil, bson.D{
                {"splitVector", ds.ns.Str()},
                {"keyPattern", bson.M{conf.Options.FullSyncReaderParallelIndex: 1}},
                // {"maxSplitPoints", ds.pieceNumber - 1},
                {"maxChunkSize", ds.pieceByteSize / utils.MB},
-       }).Decode(res)
+       }).Decode(&res)
        // if failed, do not panic, run single thread fetching
        if err != nil {
                LOG.Warn("splitter[%s] run splitVector failed[%v], give up parallel fetching", ds, err)
@@ -123,43 +127,44 @@ func (ds *DocumentSplitter) Run() error {
 
        LOG.Info("splitter[%s] run splitVector result: %v", ds, res)
 
-       if splitKeys, ok := res["splitKeys"]; ok {
-               if splitKeysList, ok := splitKeys.([]interface{}); ok && len(splitKeysList) > 0 {
-                       // return list is sorted
-                       ds.pieceNumber = len(splitKeysList) + 1
-
-                       var start interface{}
-                       cnt := 0
-                       for i, keyDoc := range splitKeysList {
-                               // check key == conf.Options.FullSyncReaderParallelIndex
-                               key, val, err := parseDocKeyValue(keyDoc)
-                               if err != nil {
-                                       LOG.Crash("splitter[%s] parse doc key failed: %v", ds, err)
-                               }
-                               if key != conf.Options.FullSyncReaderParallelIndex {
-                                       LOG.Crash("splitter[%s] parse doc invalid key: %v", ds, key)
-                               }
-
-                               LOG.Info("splitter[%s] piece[%d] create reader with boundary(%v, %v]", ds, cnt, start, val)
-                               // inject new DocumentReader into channel
-                               ds.readerChan <- NewDocumentReader(cnt, ds.src, ds.ns, key, start, val, ds.sslRootCaFile)
-
-                               // new start
-                               start = val
-                               cnt++
-
-                               // last one
-                               if i == len(splitKeysList)-1 {
-                                       LOG.Info("splitter[%s] piece[%d] create reader with boundary(%v, INF)", ds, cnt, start)
-                                       // inject new DocumentReader into channel
-                                       ds.readerChan <- NewDocumentReader(cnt, ds.src, ds.ns, key, start, nil, ds.sslRootCaFile)
-                               }
+       if len(res.SplitKeys) > 0 {
+               // return list is sorted
+               ds.pieceNumber = len(res.SplitKeys) + 1
+
+               var start interface{}
+               cnt := 0
+               for i, keyDoc := range res.SplitKeys {
+                       // check key == conf.Options.FullSyncReaderParallelIndex
+                       doc := bson.M{}
+                       err := bson.Unmarshal(keyDoc, &doc)
+                       if err != nil {
+                               LOG.Crash("splitter[%s] unmarshal doc [%v] failed: %v", ds, keyDoc, err)
+                       }
+                       key, val, err := parseDocKeyValue(doc)
+                       if err != nil {
+                               LOG.Crash("splitter[%s] parse doc key failed: %v", ds, err)
+                       }
+                       if key != conf.Options.FullSyncReaderParallelIndex {
+                               LOG.Crash("splitter[%s] parse doc invalid key: %v", ds, key)
                        }
 
-                       return nil
-               } else {
-                       LOG.Warn("splitter[%s] run splitVector return empty result[%v]", ds, res)
+                       LOG.Info("splitter[%s] piece[%d] create reader with boundary(%v, %v]", ds, cnt, start, val)
+                       // inject new DocumentReader into channel
+                       ds.readerChan <- NewDocumentReader(cnt, ds.src, ds.ns, key, start, val, ds.sslRootCaFile)
+
+                       // new start
+                       start = val
+                       cnt++
+
+                       // last one
+                       if i == len(res.SplitKeys)-1 {
+                               LOG.Info("splitter[%s] piece[%d] create reader with boundary(%v, INF)", ds, cnt, start)
+                               // inject new DocumentReader into channel
+                               ds.readerChan <- NewDocumentReader(cnt, ds.src, ds.ns, key, start, nil, ds.sslRootCaFile)
+                       }
                }
+
+               return nil
        } else {
                LOG.Warn("splitter[%s] run splitVector return null result[%v]", ds, res)
        }

[zhongli-james]

感谢，正好还没来得及测试。我会结合你提供的能成功运行的代码来修改。

[zhongli-james]

跟随v2.8.7发布