feat(storage): add check and truncate full text bf files (openGemini#613)

Signed-off-by: y00699588 <[email protected]>
Co-authored-by: y00699588 <[email protected]>

Author: Chuckie12138

engine/detached_check_metaInfo.go

@@ -49,7 +49,7 @@ func NewDetachedMetaInfo() *DetachedMetaInfo {
 	return &DetachedMetaInfo{}
 }
 
-func (d *DetachedMetaInfo) checkAndTruncateDetachedFiles(dir, msName string, bfCols []string) error {
+func (d *DetachedMetaInfo) checkAndTruncateDetachedFiles(dir, msName string, bfCols []string, isFullTextIdx bool) error {
 	filePath := d.getDetachedPrefixFilePath(dir, msName)
 	//check detached metaIndex file
 	err := d.checkAndTruncateMetaIndexFile(filePath)
@@ -82,7 +82,7 @@ func (d *DetachedMetaInfo) checkAndTruncateDetachedFiles(dir, msName string, bfC
 	}
 
 	//check bloom filter files
-	return d.checkAndTruncateBfFiles(dir, msName, bfCols)
+	return d.checkAndTruncateBfFiles(dir, msName, bfCols, isFullTextIdx)
 }
 
 func (d *DetachedMetaInfo) getDetachedPrefixFilePath(dir, msName string) string {
@@ -343,15 +343,19 @@ func (d *DetachedMetaInfo) checkAndTruncatePkIdxFile(filePath string) error {
 	return nil
 }
 
-func (d *DetachedMetaInfo) checkAndTruncateBfFiles(dir, msName string, bfCols []string) error {
+func (d *DetachedMetaInfo) checkAndTruncateBfFiles(dir, msName string, bfCols []string, isFullTextIdx bool) error {
 	localPath := dir
 	prefixDataPathLength := len(obs.GetPrefixDataPath())
 	remotePath := localPath[prefixDataPathLength:]
-
+	cols := bfCols
+	if isFullTextIdx {
+		cols = append(cols, sparseindex.FullTextIndex)
+	}
 	//get bloom filter cols -> gen bloom filter files
 	constant := logstore.GetConstant(logstore.CurrentLogTokenizerVersion)
-	for i := range bfCols {
-		fdr, fdl, err := d.getBloomFilterFile(remotePath, localPath, msName, bfCols[i])
+
+	for i := range cols {
+		fdr, fdl, err := d.getBloomFilterFile(remotePath, localPath, msName, cols[i])
 		if err != nil {
 			return err
 		}

engine/detached_check_metaInfo_test.go

@@ -91,7 +91,7 @@ func TestCheckDetachedFiles(t *testing.T) {
 
 	require.Equal(t, 4*100, int(sh.rowCount))
 	detachedInfo := NewDetachedMetaInfo()
-	err = detachedInfo.checkAndTruncateDetachedFiles(sh.filesPath, defaultMeasurementName, bfColumn)
+	err = detachedInfo.checkAndTruncateDetachedFiles(sh.filesPath, defaultMeasurementName, bfColumn, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -153,7 +153,7 @@ func TestTestCheckDetachedFilesV2(t *testing.T) {
 		mutable.SetWriteChunk(msInfo, rec)
 		sh.commitSnapshot(sh.activeTbl)
 		detachedInfo := NewDetachedMetaInfo()
-		err = detachedInfo.checkAndTruncateDetachedFiles(sh.filesPath, defaultMeasurementName, bfColumn)
+		err = detachedInfo.checkAndTruncateDetachedFiles(sh.filesPath, defaultMeasurementName, bfColumn, false)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -168,7 +168,7 @@ func TestTestCheckDetachedFilesV2(t *testing.T) {
 func TestCheckMetaIndexFile(t *testing.T) {
 	testDir := t.TempDir()
 	detachedInfo := NewDetachedMetaInfo()
-	err := detachedInfo.checkAndTruncateDetachedFiles("unknownPath", "cpu", nil)
+	err := detachedInfo.checkAndTruncateDetachedFiles("unknownPath", "cpu", nil, false)
 	if err == nil {
 		t.Fatal("should return open metaIndex file path error")
 	}
@@ -349,7 +349,7 @@ func TestCheckBloomFilterFiles(t *testing.T) {
 	testDir := t.TempDir()
 	detachedInfo := NewDetachedMetaInfo()
 	bfCols := []string{"tags"}
-	err := detachedInfo.checkAndTruncateBfFiles("unknownPath", "", bfCols)
+	err := detachedInfo.checkAndTruncateBfFiles("unknownPath", "", bfCols, false)
 	if err == nil {
 		t.Fatal("should return open bloom filter file path error")
 	}
@@ -372,7 +372,7 @@ func TestCheckBloomFilterFiles(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols)
+	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -382,7 +382,51 @@ func TestCheckBloomFilterFiles(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols)
+	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = os.RemoveAll(testDir)
+}
+
+func TestCheckFullTextBfFiles(t *testing.T) {
+	testDir := t.TempDir()
+	detachedInfo := NewDetachedMetaInfo()
+	err := detachedInfo.checkAndTruncateBfFiles("unknownPath", "", nil, true)
+	if err == nil {
+		t.Fatal("should return open bloom filter file path error")
+	}
+	constant := logstore.GetConstant(logstore.CurrentLogTokenizerVersion)
+	buf := make([]byte, constant.FilterDataDiskSize*2)
+	lock := fileops.FileLockOption("")
+	pri := fileops.FilePriorityOption(fileops.IO_PRIORITY_NORMAL)
+	bfCols := []string{sparseindex.FullTextIndex}
+	filePath := sparseindex.GetBloomFilterFilePath(testDir, "", bfCols[0])
+
+	fd, err := fileops.OpenFile(filePath, os.O_CREATE|os.O_RDWR, 0640, lock, pri)
+	if err != nil {
+		log.Error("open detached bloom filter file fail", zap.String("name", testDir), zap.Error(err))
+		t.Fatal(err)
+	}
+	defer func() {
+		_ = fd.Close()
+	}()
+
+	_, err = fd.Write(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	detachedInfo.lastPkMetaEndBlockId = 3
+	_, err = fd.Write(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = detachedInfo.checkAndTruncateBfFiles(testDir, "", bfCols, true)
 	if err != nil {
 		t.Fatal(err)
 	}

engine/immutable/compact_test.go

@@ -3824,3 +3824,30 @@ func TestWriteMemoryBloomFilterData(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, n*2, msBuilder.localBFCount)
 }
+
+func TestGetLocalBloomFilterData(t *testing.T) {
+	testCompDir := t.TempDir()
+	_ = fileops.RemoveAll(testCompDir)
+	sig := interruptsignal.NewInterruptSignal()
+	defer func() {
+		sig.Close()
+		_ = fileops.RemoveAll(testCompDir)
+	}()
+
+	conf := NewColumnStoreConfig()
+	tier := uint64(util.Hot)
+	lockPath := ""
+	store := NewTableStore(testCompDir, &lockPath, &tier, true, conf)
+	defer store.Close()
+
+	fileName := NewTSSPFileName(store.NextSequence(), 0, 0, 0, true, &lockPath)
+	msBuilder := NewMsBuilder(store.path, "mst", &lockPath, conf, 1, fileName, store.Tier(), nil, 2, config.COLUMNSTORE, nil, 0)
+	p := filepath.Join(msBuilder.Path, "full_text_bf.idx")
+	skipIndexFilePaths := make([]string, 0, 1)
+	skipIndexFilePaths = append(skipIndexFilePaths, p)
+	msBuilder.localBFCount = 1
+	_, err := msBuilder.getLocalBloomFilterData(skipIndexFilePaths)
+	if err == nil {
+		t.Fatal("should return no such file or dir")
+	}
+}

engine/partition.go

@@ -37,6 +37,7 @@ import (
 	"github.com/openGemini/openGemini/lib/index"
 	"github.com/openGemini/openGemini/lib/interruptsignal"
 	"github.com/openGemini/openGemini/lib/logger"
+	"github.com/openGemini/openGemini/lib/logstore"
 	"github.com/openGemini/openGemini/lib/metaclient"
 	"github.com/openGemini/openGemini/lib/netstorage"
 	"github.com/openGemini/openGemini/lib/statisticsPusher/statistics"
@@ -664,7 +665,7 @@ func (dbPT *DBPTInfo) loadProcess(opId uint64, thermalShards map[uint64]struct{}
 
 func checkAndTruncateDetachedFiles(d *DetachedMetaInfo, mstInfo *meta.MeasurementInfo, sh *shard) error {
 	d.obsOpt = mstInfo.ObsOptions
-	err := d.checkAndTruncateDetachedFiles(sh.filesPath, mstInfo.Name, mstInfo.IndexRelation.GetBloomFilterColumns())
+	err := d.checkAndTruncateDetachedFiles(sh.filesPath, mstInfo.Name, mstInfo.IndexRelation.GetBloomFilterColumns(), logstore.IsFullTextIdx(&mstInfo.IndexRelation))
 	if err != nil {
 		return err
 	}