fix: sync analyzer runtime options (#50997)

issue: #50931

## Summary
Initialize analyzer runtime options with Lindera download URLs and the
default dictionary path. Register runtime config callbacks so analyzer
yaml updates are propagated to the Rust analyzer layer.

Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

internal/datanode/data_node.go

@@ -221,9 +221,14 @@ func (node *DataNode) Init() error {
 		err := index.InitSegcore(serverID)
 		if err != nil {
 			initError = err
+			return
 		}
 
-		analyzer.InitOptions()
+		if err := analyzer.InitOptions(); err != nil {
+			log.Error(node.ctx, "DataNode init analyzer options failed", mlog.Err(err))
+			initError = err
+			return
+		}
 		log.Info(node.ctx, "init datanode done", mlog.String("Address", node.address))
 	})
 	return initError

internal/querynodev2/server.go

@@ -350,7 +350,11 @@ func (node *QueryNode) Init() error {
 
 		node.factory.Init(paramtable.Get())
 		// init analyzer options
-		analyzer.InitOptions()
+		if err := analyzer.InitOptions(); err != nil {
+			mlog.Error(node.ctx, "QueryNode init analyzer options failed", mlog.Err(err))
+			initError = err
+			return
+		}
 
 		localRootPath := paramtable.Get().LocalStorageCfg.Path.GetValue()
 		localUsedSize, err := segcore.GetLocalUsedSize(localRootPath)

internal/streamingnode/server/server.go

@@ -10,6 +10,7 @@ import (
 	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/service"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/walmanager"
+	"github.com/milvus-io/milvus/internal/util/analyzer"
 	"github.com/milvus-io/milvus/internal/util/fileresource"
 	"github.com/milvus-io/milvus/internal/util/initcore"
 	"github.com/milvus-io/milvus/internal/util/sessionutil"
@@ -47,6 +48,10 @@ func (s *Server) init() {
 	// init file resource manager
 	fileresource.InitManager(resource.Resource().ChunkManager(), fileresource.ParseMode(paramtable.Get().CommonCfg.QNFileResourceMode.GetValue()))
 
+	if err := analyzer.InitOptions(); err != nil {
+		panic(fmt.Sprintf("init analyzer options failed, %+v", err))
+	}
+
 	mlog.Info(context.TODO(), "init query segcore...")
 	if err := initcore.InitQueryNode(context.TODO()); err != nil {
 		panic(fmt.Sprintf("init query node segcore failed, %+v", err))

internal/util/analyzer/analyzer.go

@@ -27,6 +27,6 @@ func BuildExtraResourceInfo(storage string, resources []*internalpb.FileResource
 	return canalyzer.BuildExtraResourceInfo(storage, resources)
 }
 
-func InitOptions() {
-	canalyzer.InitOptions()
+func InitOptions() error {
+	return canalyzer.InitOptions()
 }

internal/util/analyzer/canalyzer/c_analyzer_factory.go

@@ -11,6 +11,7 @@ import "C"
 import (
 	"context"
 	"encoding/json"
+	"strings"
 	"sync"
 	"unsafe"
 
@@ -18,35 +19,72 @@ import (
 
 	"github.com/milvus-io/milvus/internal/util/analyzer/interfaces"
 	"github.com/milvus-io/milvus/internal/util/pathutil"
+	"github.com/milvus-io/milvus/pkg/v3/config"
 	"github.com/milvus-io/milvus/pkg/v3/mlog"
 	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 const (
-	LinderaDictURLKey = "lindera_download_urls"
-	ResourceMapKey    = "resource_map"
-	ResourcePathKey   = "resource_path"
-	StorageNameKey    = "storage_name"
+	AnalyzerConfigPrefix = "function.analyzer."
+	LinderaDictURLKey    = "lindera_download_urls"
+	DefaultDictPathKey   = "default_dict_path"
+	ResourceMapKey       = "resource_map"
+	ResourcePathKey      = "resource_path"
+	StorageNameKey       = "storage_name"
 )
 
-var initOnce sync.Once
+var (
+	initMu             sync.Mutex
+	optionsInitialized bool
+)
 
-func InitOptions() {
-	initOnce.Do(func() {
-		UpdateParams()
-	})
+func InitOptions() error {
+	initMu.Lock()
+	defer initMu.Unlock()
+
+	if optionsInitialized {
+		return nil
+	}
+
+	if err := UpdateParams(); err != nil {
+		return err
+	}
+
+	paramtable.Get().WatchKeyPrefix(AnalyzerConfigPrefix, config.NewHandler("analyzer.runtime_options", func(*config.Event) {
+		if err := updateParams(); err != nil {
+			mlog.Error(context.TODO(), "update analyzer option failed", mlog.Err(err))
+		}
+	}))
+	optionsInitialized = true
+	return nil
 }
 
-func UpdateParams() {
+func BuildRuntimeOptions() map[string]any {
 	cfg := paramtable.Get()
-	params := map[string]any{}
-	params[LinderaDictURLKey] = cfg.FunctionCfg.LinderaDownloadUrls.GetValue()
-	params[ResourcePathKey] = pathutil.GetPath(pathutil.FileResourcePath, paramtable.GetNodeID())
+	return map[string]any{
+		LinderaDictURLKey:  buildLinderaDownloadURLs(cfg.FunctionCfg.LinderaDownloadUrls.GetValue()),
+		DefaultDictPathKey: cfg.FunctionCfg.LocalResourcePath.GetValue(),
+		ResourcePathKey:    pathutil.GetPath(pathutil.FileResourcePath, paramtable.GetNodeID()),
+	}
+}
 
-	bytes, err := json.Marshal(params)
+func buildLinderaDownloadURLs(values map[string]string) map[string][]string {
+	urls := make(map[string][]string, len(values))
+	for kind, value := range values {
+		urls[strings.TrimPrefix(kind, ".")] = paramtable.ParseAsStings(value)
+	}
+	return urls
+}
+
+func UpdateParams() error {
+	return updateParams()
+}
+
+func updateParams() error {
+	bytes, err := json.Marshal(BuildRuntimeOptions())
 	if err != nil {
-		mlog.Panic(context.TODO(), "init analyzer option failed", mlog.Err(err))
+		return err
 	}
 
 	paramPtr := C.CString(string(bytes))
@@ -54,8 +92,9 @@ func UpdateParams() {
 
 	status := C.set_tokenizer_option(paramPtr)
 	if err := HandleCStatus(&status, "failed to init segcore analyzer option"); err != nil {
-		mlog.Panic(context.TODO(), "init analyzer option failed", mlog.Err(err))
+		return err
 	}
+	return nil
 }
 
 func BuildExtraResourceInfo(storage string, resources []*internalpb.FileResourceInfo) (string, error) {

internal/util/analyzer/canalyzer/c_analyzer_test.go

@@ -155,7 +155,7 @@ func TestAnalyzer(t *testing.T) {
 }
 
 func TestValidateAnalyzer(t *testing.T) {
-	InitOptions()
+	require.NoError(t, InitOptions())
 
 	// valid analyzer
 	{
@@ -182,7 +182,7 @@ func TestValidateAnalyzer(t *testing.T) {
 	{
 		resourcePath := pathutil.GetPath(pathutil.FileResourcePath, paramtable.GetNodeID())
 		defer os.RemoveAll(resourcePath)
-		UpdateParams()
+		require.NoError(t, UpdateParams())
 		resourceID := int64(100)
 
 		// mock remote resource file
@@ -208,7 +208,7 @@ func TestValidateAnalyzer(t *testing.T) {
 	{
 		resourcePath := pathutil.GetPath(pathutil.FileResourcePath, paramtable.GetNodeID())
 		defer os.RemoveAll(resourcePath)
-		UpdateParams()
+		require.NoError(t, UpdateParams())
 		resourceID := int64(100)
 
 		// mock remote resource file
@@ -235,3 +235,25 @@ func TestValidateAnalyzer(t *testing.T) {
 		assert.Equal(t, ids[0], resourceID)
 	}
 }
+
+func TestBuildRuntimeOptions(t *testing.T) {
+	params := paramtable.Get()
+	localResourcePath := "/tmp/milvus-analyzer-test"
+	urlKey := "function.analyzer.lindera.download_urls.ipadic"
+	require.NoError(t, params.Save(params.FunctionCfg.LocalResourcePath.Key, localResourcePath))
+	require.NoError(t, params.SaveGroup(map[string]string{urlKey: "http://a.test/ipadic.tar.gz, http://b.test/ipadic.tar.gz"}))
+	defer params.Reset(params.FunctionCfg.LocalResourcePath.Key)
+	defer params.Reset(urlKey)
+
+	options := BuildRuntimeOptions()
+	assert.Equal(t, localResourcePath, options[DefaultDictPathKey])
+	assert.Equal(t, pathutil.GetPath(pathutil.FileResourcePath, paramtable.GetNodeID()), options[ResourcePathKey])
+	assert.Equal(t,
+		map[string][]string{"ipadic": {"http://a.test/ipadic.tar.gz", "http://b.test/ipadic.tar.gz"}},
+		options[LinderaDictURLKey],
+	)
+	assert.Equal(t,
+		map[string][]string{"ipadic": {"http://a.test/ipadic.tar.gz"}},
+		buildLinderaDownloadURLs(map[string]string{".ipadic": "http://a.test/ipadic.tar.gz"}),
+	)
+}