Merge remote-tracking branch 'origin/dev' into dev

kernel/model/index.go

@@ -105,7 +105,7 @@ func index(boxID string) {
 
 		cache.PutDocIAL(file.path, docIAL)
 		treenode.ReindexBlockTree(tree)
-		sql.UpsertTreeQueue(tree)
+		sql.IndexTreeQueue(box.ID, file.path)
 
 		util.IncBootProgress(bootProgressPart, fmt.Sprintf(Conf.Language(92), util.ShortPathForBootingDisplay(tree.Path)))
 		treeSize += file.size

kernel/model/transaction.go

@@ -1447,7 +1447,7 @@ func reindexTree0(tree *parse.Tree, i, size int) {
 		indexWriteJSONQueue(tree)
 	} else {
 		treenode.ReindexBlockTree(tree)
-		sql.UpsertTreeQueue(tree)
+		sql.IndexTreeQueue(tree.Box, tree.Path)
 	}
 	util.PushStatusBar(fmt.Sprintf(Conf.Language(183), i, size, html.EscapeHTMLStr(path.Base(tree.HPath))))
 }

kernel/sql/queue.go

@@ -17,6 +17,9 @@
 package sql
 
 import (
+	"database/sql"
+	"errors"
+	"fmt"
 	"path"
 	"sync"
 	"time"
@@ -37,12 +40,13 @@ var (
 
 type dbQueueOperation struct {
 	inQueueTime time.Time
-	action      string // upsert/delete/delete_id/rename/delete_box/delete_box_refs/insert_refs
+	action      string // upsert/delete/delete_id/rename/delete_box/delete_box_refs/insert_refs/index
 
+	indexPath                     string      // index
 	upsertTree                    *parse.Tree // upsert/insert_refs
 	removeTreeBox, removeTreePath string      // delete
 	removeTreeIDBox, removeTreeID string      // delete_id
-	box                           string      // delete_box/delete_box_refs
+	box                           string      // delete_box/delete_box_refs/index
 	renameTree                    *parse.Tree // rename
 	renameTreeOldHPath            string      // rename
 }
@@ -102,55 +106,31 @@ func FlushQueue() {
 		return
 	}
 
+	var execOps int
 	context := map[string]interface{}{eventbus.CtxPushMsg: eventbus.CtxPushMsgToStatusBar}
-	execOps := 0
 	for i, op := range ops {
 		if util.IsExiting {
-			break
+			return
-		}
-
-		switch op.action {
-		case "upsert":
-			err = upsertTree(tx, op.upsertTree, context)
-		case "delete":
-			err = batchDeleteByPathPrefix(tx, op.removeTreeBox, op.removeTreePath)
-		case "delete_id":
-			err = deleteByRootID(tx, op.removeTreeID)
-		case "rename":
-			err = batchUpdateHPath(tx, op.renameTree.Box, op.renameTree.ID, op.renameTreeOldHPath, op.renameTree.HPath)
-			if nil != err {
-				break
-			}
-			err = updateRootContent(tx, path.Base(op.renameTree.HPath), op.renameTree.Root.IALAttr("updated"), op.renameTree.ID)
-		case "delete_box":
-			err = deleteByBoxTx(tx, op.box)
-		case "delete_box_refs":
-			err = deleteRefsByBoxTx(tx, op.box)
-		case "insert_refs":
-			err = insertRefs(tx, op.upsertTree)
-		case "update_refs":
-			err = upsertRefs(tx, op.upsertTree)
-		default:
-			logging.LogErrorf("unknown operation [%s]", op.action)
-			break
 		}
 
+		err = execOp(op, tx, context)
 		execOps++
+
 		if nil != err {
 			logging.LogErrorf("queue operation failed: %s", err)
-			break
+			return
 		}
 
-		if 0 < i && 0 == i%64 {
+		if 0 < i && 0 == execOps%64 {
 			if err = commitTx(tx); nil != err {
 				logging.LogErrorf("commit tx failed: %s", err)
-				break
+				return
 			}
 
 			execOps = 0
 			tx, err = beginTx()
 			if nil != err {
-				break
+				return
 			}
 		}
 	}
@@ -166,6 +146,38 @@ func FlushQueue() {
 	}
 }
 
+func execOp(op *dbQueueOperation, tx *sql.Tx, context map[string]interface{}) (err error) {
+	switch op.action {
+	case "index":
+		err = indexTree(tx, op.box, op.indexPath, context)
+	case "upsert":
+		err = upsertTree(tx, op.upsertTree, context)
+	case "delete":
+		err = batchDeleteByPathPrefix(tx, op.removeTreeBox, op.removeTreePath)
+	case "delete_id":
+		err = deleteByRootID(tx, op.removeTreeID)
+	case "rename":
+		err = batchUpdateHPath(tx, op.renameTree.Box, op.renameTree.ID, op.renameTreeOldHPath, op.renameTree.HPath)
+		if nil != err {
+			break
+		}
+		err = updateRootContent(tx, path.Base(op.renameTree.HPath), op.renameTree.Root.IALAttr("updated"), op.renameTree.ID)
+	case "delete_box":
+		err = deleteByBoxTx(tx, op.box)
+	case "delete_box_refs":
+		err = deleteRefsByBoxTx(tx, op.box)
+	case "insert_refs":
+		err = insertRefs(tx, op.upsertTree)
+	case "update_refs":
+		err = upsertRefs(tx, op.upsertTree)
+	default:
+		msg := fmt.Sprint("unknown operation [%s]", op.action)
+		logging.LogErrorf(msg)
+		err = errors.New(msg)
+	}
+	return
+}
+
 func mergeUpsertTrees() (ops []*dbQueueOperation) {
 	dbQueueLock.Lock()
 	defer dbQueueLock.Unlock()
@@ -231,6 +243,20 @@ func DeleteBoxQueue(boxID string) {
 	operationQueue = append(operationQueue, newOp)
 }
 
+func IndexTreeQueue(box, p string) {
+	dbQueueLock.Lock()
+	defer dbQueueLock.Unlock()
+
+	newOp := &dbQueueOperation{indexPath: p, box: box, inQueueTime: time.Now(), action: "index"}
+	for i, op := range operationQueue {
+		if "index" == op.action && op.indexPath == p && op.box == box { // 相同树则覆盖
+			operationQueue[i] = newOp
+			return
+		}
+	}
+	operationQueue = append(operationQueue, newOp)
+}
+
 func UpsertTreeQueue(tree *parse.Tree) {
 	dbQueueLock.Lock()
 	defer dbQueueLock.Unlock()

kernel/sql/upsert.go

@@ -21,6 +21,7 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"fmt"
+	"github.com/siyuan-note/siyuan/kernel/filesys"
 	"strings"
 
 	"github.com/88250/lute/parse"
@@ -392,6 +393,23 @@ func insertRefs(tx *sql.Tx, tree *parse.Tree) (err error) {
 	return err
 }
 
+func indexTree(tx *sql.Tx, box, p string, context map[string]interface{}) (err error) {
+	tree, err := filesys.LoadTree(box, p, luteEngine)
+	if nil != err {
+		return
+	}
+
+	err = insertTree(tx, tree, context)
+	return
+}
+
+func insertTree(tx *sql.Tx, tree *parse.Tree, context map[string]interface{}) (err error) {
+	blocks, spans, assets, attributes := fromTree(tree.Root, tree)
+	refs, fileAnnotationRefs := refsFromTree(tree)
+	err = insertTree0(tx, tree, context, blocks, spans, assets, attributes, refs, fileAnnotationRefs)
+	return
+}
+
 func upsertTree(tx *sql.Tx, tree *parse.Tree, context map[string]interface{}) (err error) {
 	oldBlockHashes := queryBlockHashes(tree.ID)
 	blocks, spans, assets, attributes := fromTree(tree.Root, tree)
@@ -420,7 +438,10 @@ func upsertTree(tx *sql.Tx, tree *parse.Tree, context map[string]interface{}) (e
 	for _, b := range blocks {
 		toRemoves = append(toRemoves, b.ID)
 	}
-	deleteBlocksByIDs(tx, toRemoves)
+
+	if err = deleteBlocksByIDs(tx, toRemoves); nil != err {
+		return
+	}
 
 	if err = deleteSpansByPathTx(tx, tree.Box, tree.Path); nil != err {
 		return
@@ -438,11 +459,20 @@ func upsertTree(tx *sql.Tx, tree *parse.Tree, context map[string]interface{}) (e
 		return
 	}
 
+	refs, fileAnnotationRefs := refsFromTree(tree)
+	if err = insertTree0(tx, tree, context, blocks, spans, assets, attributes, refs, fileAnnotationRefs); nil != err {
+		return
+	}
+	return err
+}
+
+func insertTree0(tx *sql.Tx, tree *parse.Tree, context map[string]interface{},
+	blocks []*Block, spans []*Span, assets []*Asset, attributes []*Attribute,
+	refs []*Ref, fileAnnotationRefs []*FileAnnotationRef) (err error) {
 	if err = insertBlocks(tx, blocks, context); nil != err {
 		return
 	}
 
-	refs, fileAnnotationRefs := refsFromTree(tree)
 	if err = insertBlockRefs(tx, refs); nil != err {
 		return
 	}
@@ -465,5 +495,5 @@ func upsertTree(tx *sql.Tx, tree *parse.Tree, context map[string]interface{}) (e
 	if err = insertAttributes(tx, attributes); nil != err {
 		return
 	}
-	return err
+	return
 }