package coordinator

import (
	"context"
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"time"

	"github.com/tidwall/gjson"
	"github.com/tikv/client-go/v2/txnkv"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/atomic"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
	"github.com/milvus-io/milvus/internal/datacoord"
	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
	"github.com/milvus-io/milvus/internal/kv/tikv"
	"github.com/milvus-io/milvus/internal/querycoordv2"
	"github.com/milvus-io/milvus/internal/rootcoord"
	streamingcoord "github.com/milvus-io/milvus/internal/streamingcoord/server"
	"github.com/milvus-io/milvus/internal/types"
	"github.com/milvus-io/milvus/internal/util/dependency"
	"github.com/milvus-io/milvus/internal/util/pathutil"
	"github.com/milvus-io/milvus/internal/util/proxyutil"
	"github.com/milvus-io/milvus/internal/util/sessionutil"
	"github.com/milvus-io/milvus/pkg/v3/common"
	"github.com/milvus-io/milvus/pkg/v3/kv"
	"github.com/milvus-io/milvus/pkg/v3/metrics"
	"github.com/milvus-io/milvus/pkg/v3/mlog"
	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
	"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
	"github.com/milvus-io/milvus/pkg/v3/proto/proxypb"
	"github.com/milvus-io/milvus/pkg/v3/proto/querypb"
	"github.com/milvus-io/milvus/pkg/v3/proto/rootcoordpb"
	"github.com/milvus-io/milvus/pkg/v3/util"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/metricsinfo"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/syncutil"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

var Params *paramtable.ComponentParam = paramtable.Get()

type mixCoordImpl struct {
	rootcoordServer  *rootcoord.Core
	queryCoordServer *querycoordv2.Server
	datacoordServer  *datacoord.Server
	streamingCoord   *streamingcoord.Server

	ctx     context.Context
	cancel  context.CancelFunc
	wg      sync.WaitGroup
	etcdCli *clientv3.Client
	tikvCli *txnkv.Client
	address string

	proxyCreator       proxyutil.ProxyCreator
	proxyWatcher       *proxyutil.ProxyWatcher
	proxyClientManager proxyutil.ProxyClientManagerInterface

	metricsCacheManager *metricsinfo.MetricsCacheManager
	stateCode           atomic.Int32
	initOnce            sync.Once
	startOnce           sync.Once
	session             *sessionutil.Session

	factory dependency.Factory

	metricsRequest *metricsinfo.MetricsRequest

	metaKVCreator  func() kv.MetaKv
	mixCoordClient types.MixCoordClient

	// POSIX directory cleanup task
	posixCleanupCancel    context.CancelFunc
	posixCleanupWg        sync.WaitGroup
	posixCleanupStartOnce sync.Once
	posixCleanupStopOnce  sync.Once

	// file resource observer
	fileResourceObserver *FileResourceObserver
}

func NewMixCoordServer(c context.Context, factory dependency.Factory) (*mixCoordImpl, error) {
	ctx, cancel := context.WithCancel(c)
	rootCoordServer, _ := rootcoord.NewCore(ctx, factory)
	queryCoordServer, _ := querycoordv2.NewQueryCoord(c)
	dataCoordServer := datacoord.CreateServer(c, factory)

	return &mixCoordImpl{
		ctx:              ctx,
		cancel:           cancel,
		rootcoordServer:  rootCoordServer,
		queryCoordServer: queryCoordServer,
		datacoordServer:  dataCoordServer,
		factory:          factory,
	}, nil
}

// Register register mixcoord at etcd
func (s *mixCoordImpl) Register() error {
	s.session.Register()
	afterRegister := func() {
		metrics.NumNodes.WithLabelValues(paramtable.GetStringNodeID(), typeutil.MixCoordRole).Inc()
		mlog.Info(s.ctx, "MixCoord Register Finished")
	}
	go func() {
		if err := s.session.ProcessActiveStandBy(s.activateFunc); err != nil {
			if s.ctx.Err() == context.Canceled {
				mlog.Info(s.ctx, "standby process canceled due to server shutdown")
				return
			}
			mlog.Error(s.ctx, "failed to activate standby server", mlog.Err(err))
			panic(err)
		}
		afterRegister()
	}()
	return nil
}

func (s *mixCoordImpl) Init() error {
	var initErr error
	if initErr = s.initSession(); initErr != nil {
		return initErr
	}
	s.factory.Init(Params)
	s.initKVCreator()
	s.initStreamingCoord()
	s.UpdateStateCode(commonpb.StateCode_StandBy)
	mlog.Info(s.ctx, "MixCoord enter standby mode successfully")
	return nil
}

func (s *mixCoordImpl) activateFunc() error {
	mlog.Info(s.ctx, "mixCoord switch from standby to active, activating")
	var err error
	s.initOnce.Do(func() {
		if err = s.initInternal(); err != nil {
			mlog.Error(s.ctx, "mixCoord init failed", mlog.Err(err))
		}
	})
	if err != nil {
		return err
	}
	mlog.Info(s.ctx, "mixCoord startup success", mlog.String("address", s.session.GetAddress()))
	s.startAndUpdateHealthy()
	return err
}

func (s *mixCoordImpl) initInternal() error {
	s.rootcoordServer.SetMixCoord(s)
	s.datacoordServer.SetMixCoord(s)
	s.queryCoordServer.SetMixCoord(s)
	s.fileResourceObserver = NewFileResourceObserver(s.ctx)

	// Register WAL callbacks
	RegisterWALCallbacks(s)

	if err := s.streamingCoord.Start(s.ctx, s.fileResourceObserver); err != nil {
		mlog.Error(s.ctx, "streamCoord start failed", mlog.Err(err))
		return err
	}

	s.rootcoordServer.SetFileResourceObserver(s.fileResourceObserver)
	if err := s.rootcoordServer.Init(); err != nil {
		mlog.Error(s.ctx, "rootCoord init failed", mlog.Err(err))
		return err
	}

	if err := s.rootcoordServer.Start(); err != nil {
		mlog.Error(s.ctx, "rootCoord start failed", mlog.Err(err))
		return err
	}

	// DataCoord and QueryCoord are independent of each other;
	// both only depend on RootCoord being ready. Initialize and start them in parallel.
	g, _ := errgroup.WithContext(s.ctx)
	g.Go(func() error {
		s.datacoordServer.SetFileResourceObserver(s.fileResourceObserver)
		if err := s.datacoordServer.Init(); err != nil {
			mlog.Error(s.ctx, "dataCoord init failed", mlog.Err(err))
			return err
		}
		if err := s.datacoordServer.Start(); err != nil {
			mlog.Error(s.ctx, "dataCoord start failed", mlog.Err(err))
			return err
		}
		return nil
	})
	g.Go(func() error {
		s.queryCoordServer.SetFileResourceObserver(s.fileResourceObserver)
		if err := s.queryCoordServer.Init(); err != nil {
			mlog.Error(s.ctx, "queryCoord init failed", mlog.Err(err))
			return err
		}
		if err := s.queryCoordServer.Start(); err != nil {
			mlog.Error(s.ctx, "queryCoord start failed", mlog.Err(err))
			return err
		}
		return nil
	})
	if err := g.Wait(); err != nil {
		return err
	}

	s.fileResourceObserver.Start()
	return nil
}

func (s *mixCoordImpl) initKVCreator() {
	if s.metaKVCreator == nil {
		if Params.MetaStoreCfg.MetaStoreType.GetValue() == util.MetaStoreTypeTiKV {
			s.metaKVCreator = func() kv.MetaKv {
				return tikv.NewTiKV(s.tikvCli, Params.TiKVCfg.MetaRootPath.GetValue(),
					tikv.WithRequestTimeout(paramtable.Get().TiKVCfg.RequestTimeout.GetAsDuration(time.Millisecond)))
			}
		} else {
			s.metaKVCreator = func() kv.MetaKv {
				return etcdkv.NewEtcdKV(s.etcdCli, Params.EtcdCfg.MetaRootPath.GetValue(),
					etcdkv.WithRequestTimeout(paramtable.Get().EtcdCfg.RequestTimeout.GetAsDuration(time.Millisecond)))
			}
		}
	}
}

func (s *mixCoordImpl) Start() error {
	return nil
}

// startAndUpdateHealthy is the internal impl actually start mixcoord
// it could be invoked by:
// - `Start()` when enableActiveStandBy is false
// - `activateFunc` when enableActiveStandBy is true
func (s *mixCoordImpl) startAndUpdateHealthy() {
	s.UpdateStateCode(commonpb.StateCode_Healthy)
	s.startPosixCleanupTask()
	RegisterMgrRoute(s)
}

func (s *mixCoordImpl) IsServerActive(serverID int64) bool {
	return s.queryCoordServer.ServerExist(serverID) || s.datacoordServer.ServerExist(serverID)
}

func (s *mixCoordImpl) checkExpiredPOSIXDIR() {
	if !paramtable.Get().CommonCfg.EnablePosixMode.GetAsBool() {
		return
	}

	rootCachePath := pathutil.GetPath(pathutil.RootCachePath, 0)
	var entries []os.DirEntry
	var err error
	if entries, err = os.ReadDir(rootCachePath); err != nil {
		mlog.Warn(s.ctx, "failed to read root cache directory", mlog.String("path", rootCachePath), mlog.String("error", err.Error()))
		return
	}
	var subdirs []string
	var removedDirs []string

	for _, entry := range entries {
		if entry.IsDir() {
			subdirs = append(subdirs, entry.Name())
			if nodeID, err := strconv.ParseInt(entry.Name(), 10, 64); err != nil {
				mlog.Warn(s.ctx, "invalid node directory name", mlog.String("dirName", entry.Name()), mlog.String("error", err.Error()))
			} else {
				if !s.IsServerActive(nodeID) {
					expiredDirPath := filepath.Join(rootCachePath, entry.Name())
					if err := os.RemoveAll(expiredDirPath); err != nil {
						mlog.Error(s.ctx, "failed to remove expired node directory",
							mlog.String("path", expiredDirPath),
							mlog.FieldNodeID(nodeID),
							mlog.String("error", err.Error()))
					} else {
						mlog.Info(s.ctx, "removed expired node directory",
							mlog.String("path", expiredDirPath),
							mlog.FieldNodeID(nodeID))
						removedDirs = append(removedDirs, entry.Name())
					}
				}
			}
		}
	}
	if len(removedDirs) > 0 {
		mlog.Info(s.ctx, "root cache directory cleanup completed",
			mlog.String("path", rootCachePath),
			mlog.Strings("allSubdirectories", subdirs),
			mlog.Strings("removedDirectories", removedDirs),
			mlog.Int("totalDirs", len(subdirs)),
			mlog.Int("removedDirs", len(removedDirs)))
	}
}

func (s *mixCoordImpl) startPosixCleanupTask() {
	s.posixCleanupStartOnce.Do(func() {
		ctx, cancel := context.WithCancel(s.ctx) //nolint:gosec // cancel is stored and called in stopPosixCleanupTask
		s.posixCleanupCancel = cancel

		s.posixCleanupWg.Add(1)
		go s.posixCleanupLoop(ctx)
	})
}

func (s *mixCoordImpl) stopPosixCleanupTask() {
	s.posixCleanupStopOnce.Do(func() {
		if s.posixCleanupCancel != nil {
			s.posixCleanupCancel()
		}
		s.posixCleanupWg.Wait()
	})
}

func (s *mixCoordImpl) posixCleanupLoop(ctx context.Context) {
	defer s.posixCleanupWg.Done()

	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	mlog.Info(ctx, "POSIX directory cleanup task started")

	for {
		select {
		case <-ctx.Done():
			mlog.Info(ctx, "POSIX directory cleanup task stopped")
			return
		case <-ticker.C:
			s.checkExpiredPOSIXDIR()
		}
	}
}

func (s *mixCoordImpl) Stop() error {
	mlog.Info(s.ctx, "graceful stop")

	s.stopPosixCleanupTask()

	s.GracefulStop()
	mlog.Info(s.ctx, "graceful stop done")

	if err := s.queryCoordServer.Stop(); err != nil {
		mlog.Error(s.ctx, "Failed to stop queryCoord", mlog.Err(err))
	}

	if err := s.datacoordServer.Stop(); err != nil {
		mlog.Error(s.ctx, "Failed to stop dataCoord", mlog.Err(err))
	}

	if err := s.rootcoordServer.Stop(); err != nil {
		mlog.Error(s.ctx, "Failed to stop rootCoord", mlog.Err(err))
	}

	// All coordinators have stopped. Now stop the session.
	s.session.SetMixCoordMode(false)
	s.session.Stop()

	s.fileResourceObserver.Stop()
	s.cancel()
	return nil
}

func (s *mixCoordImpl) initStreamingCoord() {
	fMixcoord := syncutil.NewFuture[types.MixCoordClient]()
	fMixcoord.Set(s.mixCoordClient)

	s.streamingCoord = streamingcoord.NewServerBuilder().
		WithETCD(s.etcdCli).
		WithMetaKV(s.metaKVCreator()).
		WithSession(s.session).
		WithMixCoordClient(fMixcoord).
		Build()
}

func (s *mixCoordImpl) initSession() error {
	s.session = sessionutil.NewSession(s.ctx)
	s.session.Init(typeutil.MixCoordRole, s.address, true)
	s.session.SetEnableActiveStandBy(true)
	// Mark session as MixCoord mode so individual coordinator Stop() calls won't cancel it.
	// MixCoord owns the session lifecycle and stops it after all coordinators have stopped.
	s.session.SetMixCoordMode(true)
	s.rootcoordServer.SetSession(s.session)
	s.datacoordServer.SetSession(s.session)
	s.queryCoordServer.SetSession(s.session)

	return nil
}

func (s *mixCoordImpl) SetAddress(address string) {
	s.address = address
	s.rootcoordServer.SetAddress(address)
	s.datacoordServer.SetAddress(address)
	s.queryCoordServer.SetAddress(address)
}

func (s *mixCoordImpl) SetEtcdClient(client *clientv3.Client) {
	s.etcdCli = client
	s.rootcoordServer.SetEtcdClient(client)
	s.datacoordServer.SetEtcdClient(client)
	s.queryCoordServer.SetEtcdClient(client)
}

func (s *mixCoordImpl) SetTiKVClient(client *txnkv.Client) {
	s.tikvCli = client
	s.rootcoordServer.SetTiKVClient(client)
	s.datacoordServer.SetTiKVClient(client)
	s.queryCoordServer.SetTiKVClient(client)
}

func (s *mixCoordImpl) SetMixCoordClient(client types.MixCoordClient) {
	s.mixCoordClient = client
}

func (s *mixCoordImpl) GetServerID() int64 {
	return paramtable.GetNodeID()
}

func (s *mixCoordImpl) UpdateStateCode(code commonpb.StateCode) {
	s.stateCode.Store(int32(code))
}

func (s *mixCoordImpl) GetStateCode() commonpb.StateCode {
	return commonpb.StateCode(s.stateCode.Load())
}

func (s *mixCoordImpl) GracefulStop() {
	if s.streamingCoord != nil {
		s.streamingCoord.Stop()
		s.streamingCoord = nil
	}
}

// RootCoordServer
func (s *mixCoordImpl) CreateCollection(ctx context.Context, req *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreateCollection(ctx, req)
}

func (s *mixCoordImpl) DropCollection(ctx context.Context, req *milvuspb.DropCollectionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropCollection(ctx, req)
}

func (s *mixCoordImpl) HasCollection(ctx context.Context, req *milvuspb.HasCollectionRequest) (*milvuspb.BoolResponse, error) {
	return s.rootcoordServer.HasCollection(ctx, req)
}

func (s *mixCoordImpl) DescribeCollection(ctx context.Context, req *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) {
	return s.rootcoordServer.DescribeCollection(ctx, req)
}

func (s *mixCoordImpl) ShowCollections(ctx context.Context, req *milvuspb.ShowCollectionsRequest) (*milvuspb.ShowCollectionsResponse, error) {
	return s.rootcoordServer.ShowCollections(ctx, req)
}

func (s *mixCoordImpl) ShowCollectionIDs(ctx context.Context, req *rootcoordpb.ShowCollectionIDsRequest) (*rootcoordpb.ShowCollectionIDsResponse, error) {
	return s.rootcoordServer.ShowCollectionIDs(ctx, req)
}

func (s *mixCoordImpl) AlterCollection(ctx context.Context, req *milvuspb.AlterCollectionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AlterCollection(ctx, req)
}

func (s *mixCoordImpl) AlterCollectionField(ctx context.Context, req *milvuspb.AlterCollectionFieldRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AlterCollectionField(ctx, req)
}

func (s *mixCoordImpl) AlterCollectionSchema(ctx context.Context, req *milvuspb.AlterCollectionSchemaRequest) (*milvuspb.AlterCollectionSchemaResponse, error) {
	return s.rootcoordServer.AlterCollectionSchema(ctx, req)
}

func (s *mixCoordImpl) AddCollectionFunction(ctx context.Context, req *milvuspb.AddCollectionFunctionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AddCollectionFunction(ctx, req)
}

func (s *mixCoordImpl) AlterCollectionFunction(ctx context.Context, req *milvuspb.AlterCollectionFunctionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AlterCollectionFunction(ctx, req)
}

func (s *mixCoordImpl) DropCollectionFunction(ctx context.Context, req *milvuspb.DropCollectionFunctionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropCollectionFunction(ctx, req)
}

func (s *mixCoordImpl) CreatePartition(ctx context.Context, req *milvuspb.CreatePartitionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreatePartition(ctx, req)
}

func (s *mixCoordImpl) CreatePartitionV2(ctx context.Context, req *milvuspb.CreatePartitionRequest) (*rootcoordpb.CreatePartitionResponse, error) {
	return s.rootcoordServer.CreatePartitionV2(ctx, req)
}

func (s *mixCoordImpl) DropPartition(ctx context.Context, req *milvuspb.DropPartitionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropPartition(ctx, req)
}

func (s *mixCoordImpl) HasPartition(ctx context.Context, req *milvuspb.HasPartitionRequest) (*milvuspb.BoolResponse, error) {
	return s.rootcoordServer.HasPartition(ctx, req)
}

func (s *mixCoordImpl) ShowPartitions(ctx context.Context, req *milvuspb.ShowPartitionsRequest) (*milvuspb.ShowPartitionsResponse, error) {
	return s.rootcoordServer.ShowPartitions(ctx, req)
}

func (s *mixCoordImpl) ShowPartitionsInternal(ctx context.Context, req *milvuspb.ShowPartitionsRequest) (*milvuspb.ShowPartitionsResponse, error) {
	return s.rootcoordServer.ShowPartitionsInternal(ctx, req)
}

func (s *mixCoordImpl) AllocTimestamp(ctx context.Context, req *rootcoordpb.AllocTimestampRequest) (*rootcoordpb.AllocTimestampResponse, error) {
	return s.rootcoordServer.AllocTimestamp(ctx, req)
}

func (s *mixCoordImpl) AllocID(ctx context.Context, req *rootcoordpb.AllocIDRequest) (*rootcoordpb.AllocIDResponse, error) {
	return s.rootcoordServer.AllocID(ctx, req)
}

func (s *mixCoordImpl) UpdateChannelTimeTick(ctx context.Context, req *internalpb.ChannelTimeTickMsg) (*commonpb.Status, error) {
	return s.rootcoordServer.UpdateChannelTimeTick(ctx, req)
}

func (s *mixCoordImpl) ShowSegments(ctx context.Context, req *milvuspb.ShowSegmentsRequest) (*milvuspb.ShowSegmentsResponse, error) {
	return s.rootcoordServer.ShowSegments(ctx, req)
}

func (s *mixCoordImpl) GetPChannelInfo(ctx context.Context, req *rootcoordpb.GetPChannelInfoRequest) (*rootcoordpb.GetPChannelInfoResponse, error) {
	return s.rootcoordServer.GetPChannelInfo(ctx, req)
}

func (s *mixCoordImpl) InvalidateCollectionMetaCache(ctx context.Context, req *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.InvalidateCollectionMetaCache(ctx, req)
}

func (s *mixCoordImpl) ShowConfigurations(ctx context.Context, req *internalpb.ShowConfigurationsRequest) (*internalpb.ShowConfigurationsResponse, error) {
	return s.rootcoordServer.ShowConfigurations(ctx, req)
}

func (s *mixCoordImpl) CreateAlias(ctx context.Context, in *milvuspb.CreateAliasRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreateAlias(ctx, in)
}

func (s *mixCoordImpl) DescribeCollectionInternal(ctx context.Context, in *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) {
	return s.rootcoordServer.DescribeCollectionInternal(ctx, in)
}

func (s *mixCoordImpl) BackupEzk(ctx context.Context, in *internalpb.BackupEzkRequest) (*internalpb.BackupEzkResponse, error) {
	return s.rootcoordServer.BackupEzk(ctx, in)
}

// DropAlias drop collection alias
func (c *mixCoordImpl) DropAlias(ctx context.Context, in *milvuspb.DropAliasRequest) (*commonpb.Status, error) {
	return c.rootcoordServer.DropAlias(ctx, in)
}

// AlterAlias alter collection alias
func (c *mixCoordImpl) AlterAlias(ctx context.Context, in *milvuspb.AlterAliasRequest) (*commonpb.Status, error) {
	return c.rootcoordServer.AlterAlias(ctx, in)
}

// DescribeAlias describe collection alias
func (c *mixCoordImpl) DescribeAlias(ctx context.Context, in *milvuspb.DescribeAliasRequest) (*milvuspb.DescribeAliasResponse, error) {
	return c.rootcoordServer.DescribeAlias(ctx, in)
}

// ListAliases list aliases
func (c *mixCoordImpl) ListAliases(ctx context.Context, in *milvuspb.ListAliasesRequest) (*milvuspb.ListAliasesResponse, error) {
	return c.rootcoordServer.ListAliases(ctx, in)
}

func (c *mixCoordImpl) AddCollectionField(ctx context.Context, in *milvuspb.AddCollectionFieldRequest) (*commonpb.Status, error) {
	return c.rootcoordServer.AddCollectionField(ctx, in)
}

func (c *mixCoordImpl) AddCollectionStructField(ctx context.Context, in *milvuspb.AddCollectionStructFieldRequest) (*commonpb.Status, error) {
	return c.rootcoordServer.AddCollectionStructField(ctx, in)
}

func (s *mixCoordImpl) CreateCredential(ctx context.Context, req *internalpb.CredentialInfo) (*commonpb.Status, error) {
	return s.rootcoordServer.CreateCredential(ctx, req)
}

func (s *mixCoordImpl) GetCredential(ctx context.Context, req *rootcoordpb.GetCredentialRequest) (*rootcoordpb.GetCredentialResponse, error) {
	return s.rootcoordServer.GetCredential(ctx, req)
}

func (s *mixCoordImpl) UpdateCredential(ctx context.Context, req *internalpb.CredentialInfo) (*commonpb.Status, error) {
	return s.rootcoordServer.UpdateCredential(ctx, req)
}

func (s *mixCoordImpl) DeleteCredential(ctx context.Context, req *milvuspb.DeleteCredentialRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DeleteCredential(ctx, req)
}

func (s *mixCoordImpl) ListCredUsers(ctx context.Context, req *milvuspb.ListCredUsersRequest) (*milvuspb.ListCredUsersResponse, error) {
	return s.rootcoordServer.ListCredUsers(ctx, req)
}

func (s *mixCoordImpl) CreateRole(ctx context.Context, req *milvuspb.CreateRoleRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreateRole(ctx, req)
}

func (s *mixCoordImpl) AlterRole(ctx context.Context, req *milvuspb.AlterRoleRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AlterRole(ctx, req)
}

func (s *mixCoordImpl) DropRole(ctx context.Context, req *milvuspb.DropRoleRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropRole(ctx, req)
}

func (s *mixCoordImpl) OperateUserRole(ctx context.Context, req *milvuspb.OperateUserRoleRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.OperateUserRole(ctx, req)
}

func (s *mixCoordImpl) SelectRole(ctx context.Context, req *milvuspb.SelectRoleRequest) (*milvuspb.SelectRoleResponse, error) {
	return s.rootcoordServer.SelectRole(ctx, req)
}

func (s *mixCoordImpl) SelectUser(ctx context.Context, req *milvuspb.SelectUserRequest) (*milvuspb.SelectUserResponse, error) {
	return s.rootcoordServer.SelectUser(ctx, req)
}

func (s *mixCoordImpl) OperatePrivilege(ctx context.Context, req *milvuspb.OperatePrivilegeRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.OperatePrivilege(ctx, req)
}

func (s *mixCoordImpl) SelectGrant(ctx context.Context, req *milvuspb.SelectGrantRequest) (*milvuspb.SelectGrantResponse, error) {
	return s.rootcoordServer.SelectGrant(ctx, req)
}

func (s *mixCoordImpl) ListPolicy(ctx context.Context, req *internalpb.ListPolicyRequest) (*internalpb.ListPolicyResponse, error) {
	return s.rootcoordServer.ListPolicy(ctx, req)
}

func (s *mixCoordImpl) CheckHealth(ctx context.Context, req *milvuspb.CheckHealthRequest) (*milvuspb.CheckHealthResponse, error) {
	return s.rootcoordServer.CheckHealth(ctx, req)
}

func (s *mixCoordImpl) RenameCollection(ctx context.Context, req *milvuspb.RenameCollectionRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.RenameCollection(ctx, req)
}

func (s *mixCoordImpl) CreateDatabase(ctx context.Context, req *milvuspb.CreateDatabaseRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreateDatabase(ctx, req)
}

func (s *mixCoordImpl) DropDatabase(ctx context.Context, req *milvuspb.DropDatabaseRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropDatabase(ctx, req)
}

func (s *mixCoordImpl) ListDatabases(ctx context.Context, req *milvuspb.ListDatabasesRequest) (*milvuspb.ListDatabasesResponse, error) {
	return s.rootcoordServer.ListDatabases(ctx, req)
}

func (s *mixCoordImpl) DescribeDatabase(ctx context.Context, req *rootcoordpb.DescribeDatabaseRequest) (*rootcoordpb.DescribeDatabaseResponse, error) {
	return s.rootcoordServer.DescribeDatabase(ctx, req)
}

func (s *mixCoordImpl) AlterDatabase(ctx context.Context, req *rootcoordpb.AlterDatabaseRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AlterDatabase(ctx, req)
}

func (s *mixCoordImpl) BackupRBAC(ctx context.Context, req *milvuspb.BackupRBACMetaRequest) (*milvuspb.BackupRBACMetaResponse, error) {
	return s.rootcoordServer.BackupRBAC(ctx, req)
}

func (s *mixCoordImpl) RestoreRBAC(ctx context.Context, req *milvuspb.RestoreRBACMetaRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.RestoreRBAC(ctx, req)
}

func (s *mixCoordImpl) CreatePrivilegeGroup(ctx context.Context, req *milvuspb.CreatePrivilegeGroupRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.CreatePrivilegeGroup(ctx, req)
}

func (s *mixCoordImpl) DropPrivilegeGroup(ctx context.Context, req *milvuspb.DropPrivilegeGroupRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.DropPrivilegeGroup(ctx, req)
}

func (s *mixCoordImpl) ListPrivilegeGroups(ctx context.Context, req *milvuspb.ListPrivilegeGroupsRequest) (*milvuspb.ListPrivilegeGroupsResponse, error) {
	return s.rootcoordServer.ListPrivilegeGroups(ctx, req)
}

func (s *mixCoordImpl) OperatePrivilegeGroup(ctx context.Context, req *milvuspb.OperatePrivilegeGroupRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.OperatePrivilegeGroup(ctx, req)
}

// GetComponentStates get states of components
func (s *mixCoordImpl) GetComponentStates(ctx context.Context, req *milvuspb.GetComponentStatesRequest) (*milvuspb.ComponentStates, error) {
	code := s.GetStateCode()
	mlog.Debug(ctx, "Mix coord current state", mlog.String("StateCode", code.String()))

	nodeID := common.NotRegisteredID
	if s.session != nil && s.session.Registered() {
		nodeID = s.session.ServerID
	}

	return &milvuspb.ComponentStates{
		State: &milvuspb.ComponentInfo{
			NodeID:    nodeID,
			Role:      typeutil.MixCoordRole,
			StateCode: code,
			ExtraInfo: nil,
		},
		Status: merr.Success(),
		SubcomponentStates: []*milvuspb.ComponentInfo{
			{
				NodeID:    nodeID,
				Role:      typeutil.MixCoordRole,
				StateCode: code,
				ExtraInfo: nil,
			},
		},
	}, nil
}

// GetTimeTickChannel get timetick channel name
func (sc *mixCoordImpl) GetTimeTickChannel(ctx context.Context, req *internalpb.GetTimeTickChannelRequest) (*milvuspb.StringResponse, error) {
	return &milvuspb.StringResponse{
		Status: merr.Success(),
		Value:  Params.CommonCfg.RootCoordTimeTick.GetValue(),
	}, nil
}

// GetStatisticsChannel get statistics channel name
func (s *mixCoordImpl) GetStatisticsChannel(ctx context.Context, req *internalpb.GetStatisticsChannelRequest) (*milvuspb.StringResponse, error) {
	return &milvuspb.StringResponse{
		Status: merr.Success(),
		Value:  Params.CommonCfg.RootCoordStatistics.GetValue(),
	}, nil
}

// GetMetrics get metrics
func (s *mixCoordImpl) GetMetrics(ctx context.Context, in *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
	systemTopology := metricsinfo.SystemTopology{
		NodesInfo: make([]metricsinfo.SystemTopologyNode, 0),
	}

	// If a processing role is specified, the corresponding role will be used for processing
	ret := gjson.Parse(in.GetRequest())
	processRole, _ := metricsinfo.ParseMetricProcessInRole(ret)
	if len(processRole) > 0 && processRole == typeutil.QueryCoordRole {
		return s.GetQcMetrics(ctx, in)
	} else if len(processRole) > 0 && processRole == typeutil.DataCoordRole {
		return s.GetDcMetrics(ctx, in)
	}

	identifierMap := make(map[string]int)

	rootCoordResp, rootCoordErr := s.rootcoordServer.GetMetrics(ctx, in)
	var rootCoordTopology metricsinfo.RootCoordTopology
	rootCoordRoleName := ""
	if rootCoordErr == nil && rootCoordResp != nil {
		rootCoordRoleName = rootCoordResp.GetComponentName()
		rootCoordErr = metricsinfo.UnmarshalTopology(rootCoordResp.GetResponse(), &rootCoordTopology)
		identifierMap[rootCoordRoleName] = int(rootCoordTopology.Self.ID)
	}

	dataCoordResp, dataCoordErr := s.datacoordServer.GetMetrics(ctx, in)
	var dataCoordTopology metricsinfo.DataCoordTopology
	dataCoordRoleName := ""
	if dataCoordErr == nil && dataCoordResp != nil {
		dataCoordRoleName = dataCoordResp.GetComponentName()
		dataCoordErr = metricsinfo.UnmarshalTopology(dataCoordResp.GetResponse(), &dataCoordTopology)
		identifierMap[dataCoordRoleName] = int(dataCoordTopology.Cluster.Self.ID)
	}

	queryCoordResp, queryCoordErr := s.queryCoordServer.GetMetrics(ctx, in)
	var queryCoordTopology metricsinfo.QueryCoordTopology
	queryCoordRoleName := ""
	if queryCoordErr == nil && queryCoordResp != nil {
		queryCoordRoleName = queryCoordResp.GetComponentName()
		queryCoordErr = metricsinfo.UnmarshalTopology(queryCoordResp.GetResponse(), &queryCoordTopology)
		identifierMap[queryCoordRoleName] = int(queryCoordTopology.Cluster.Self.ID)
	}

	if rootCoordErr == nil && rootCoordResp != nil {
		rootCoordTopologyNode := metricsinfo.SystemTopologyNode{
			Identifier: identifierMap[rootCoordRoleName],
			Connected:  make([]metricsinfo.ConnectionEdge, 0),
			Infos:      &rootCoordTopology.Self,
		}

		if dataCoordErr == nil && dataCoordResp != nil {
			rootCoordTopologyNode.Connected = append(rootCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[dataCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.DataCoordRole,
			})
		}
		if queryCoordErr == nil && queryCoordResp != nil {
			rootCoordTopologyNode.Connected = append(rootCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[queryCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.QueryCoordRole,
			})
		}

		systemTopology.NodesInfo = append(systemTopology.NodesInfo, rootCoordTopologyNode)
	}

	if dataCoordErr == nil && dataCoordResp != nil {
		dataCoordTopologyNode := metricsinfo.SystemTopologyNode{
			Identifier: identifierMap[dataCoordRoleName],
			Connected:  make([]metricsinfo.ConnectionEdge, 0),
			Infos:      &dataCoordTopology.Cluster.Self,
		}

		if rootCoordErr == nil && rootCoordResp != nil {
			dataCoordTopologyNode.Connected = append(dataCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[rootCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.RootCoordRole,
			})
		}
		if queryCoordErr == nil && queryCoordResp != nil {
			dataCoordTopologyNode.Connected = append(dataCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[queryCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.QueryCoordRole,
			})
		}

		for _, dataNode := range dataCoordTopology.Cluster.ConnectedDataNodes {
			node := dataNode
			identifier := int(node.ID)
			identifierMap[dataNode.Name] = identifier
			dataNodeTopologyNode := metricsinfo.SystemTopologyNode{
				Identifier: identifier,
				Connected:  nil,
				Infos:      &node,
			}
			systemTopology.NodesInfo = append(systemTopology.NodesInfo, dataNodeTopologyNode)
			dataCoordTopologyNode.Connected = append(dataCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifier,
				Type:                metricsinfo.CoordConnectToNode,
				TargetType:          typeutil.DataNodeRole,
			})
		}

		systemTopology.NodesInfo = append(systemTopology.NodesInfo, dataCoordTopologyNode)
	}

	if queryCoordErr == nil && queryCoordResp != nil {
		queryCoordTopologyNode := metricsinfo.SystemTopologyNode{
			Identifier: identifierMap[queryCoordRoleName],
			Connected:  make([]metricsinfo.ConnectionEdge, 0),
			Infos:      &queryCoordTopology.Cluster.Self,
		}

		if rootCoordErr == nil && rootCoordResp != nil {
			queryCoordTopologyNode.Connected = append(queryCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[rootCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.RootCoordRole,
			})
		}
		if dataCoordErr == nil && dataCoordResp != nil {
			queryCoordTopologyNode.Connected = append(queryCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifierMap[dataCoordRoleName],
				Type:                metricsinfo.Forward,
				TargetType:          typeutil.DataCoordRole,
			})
		}

		for _, queryNode := range queryCoordTopology.Cluster.ConnectedNodes {
			node := queryNode
			identifier := int(node.ID)
			identifierMap[queryNode.Name] = identifier
			queryNodeTopologyNode := metricsinfo.SystemTopologyNode{
				Identifier: identifier,
				Connected:  nil,
				Infos:      &node,
			}
			systemTopology.NodesInfo = append(systemTopology.NodesInfo, queryNodeTopologyNode)
			queryCoordTopologyNode.Connected = append(queryCoordTopologyNode.Connected, metricsinfo.ConnectionEdge{
				ConnectedIdentifier: identifier,
				Type:                metricsinfo.CoordConnectToNode,
				TargetType:          node.Type,
			})
		}

		systemTopology.NodesInfo = append(systemTopology.NodesInfo, queryCoordTopologyNode)
	}

	resp, err := metricsinfo.MarshalTopology(systemTopology)
	if err != nil {
		return &milvuspb.GetMetricsResponse{
			Status:        merr.Status(err),
			Response:      "",
			ComponentName: metricsinfo.ConstructComponentName(typeutil.MixCoordRole, paramtable.GetNodeID()),
		}, nil
	}

	return &milvuspb.GetMetricsResponse{
		Status:        merr.Success(),
		Response:      resp,
		ComponentName: metricsinfo.ConstructComponentName(typeutil.MixCoordRole, paramtable.GetNodeID()),
	}, nil
}

// GetMetrics get metrics
func (s *mixCoordImpl) GetDcMetrics(ctx context.Context, in *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
	return s.datacoordServer.GetMetrics(ctx, in)
}

func (s *mixCoordImpl) GetQcMetrics(ctx context.Context, in *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
	return s.queryCoordServer.GetMetrics(ctx, in)
}

func (s *mixCoordImpl) GetDataCoordTopology(ctx context.Context, req *milvuspb.GetMetricsRequest) (*metricsinfo.DataCoordTopology, error) {
	return s.datacoordServer.GetDataCoordTopology(ctx, req)
}

func (s *mixCoordImpl) GetQueryCoordTopology(ctx context.Context, req *milvuspb.GetMetricsRequest) (*metricsinfo.QueryCoordTopology, error) {
	return s.queryCoordServer.GetQueryCoordTopology(ctx, req)
}

func (s *mixCoordImpl) ActivateChecker(ctx context.Context, req *querypb.ActivateCheckerRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.ActivateChecker(ctx, req)
}

func (s *mixCoordImpl) DeactivateChecker(ctx context.Context, req *querypb.DeactivateCheckerRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.DeactivateChecker(ctx, req)
}

func (s *mixCoordImpl) ListCheckers(ctx context.Context, req *querypb.ListCheckersRequest) (*querypb.ListCheckersResponse, error) {
	return s.queryCoordServer.ListCheckers(ctx, req)
}

func (s *mixCoordImpl) ShowLoadCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
	return s.queryCoordServer.ShowLoadCollections(ctx, req)
}

func (s *mixCoordImpl) LoadCollection(ctx context.Context, req *querypb.LoadCollectionRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.LoadCollection(ctx, req)
}

func (s *mixCoordImpl) ReleaseCollection(ctx context.Context, req *querypb.ReleaseCollectionRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.ReleaseCollection(ctx, req)
}

func (s *mixCoordImpl) ShowLoadPartitions(ctx context.Context, req *querypb.ShowPartitionsRequest) (*querypb.ShowPartitionsResponse, error) {
	return s.queryCoordServer.ShowLoadPartitions(ctx, req)
}

func (s *mixCoordImpl) LoadPartitions(ctx context.Context, req *querypb.LoadPartitionsRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.LoadPartitions(ctx, req)
}

func (s *mixCoordImpl) ReleasePartitions(ctx context.Context, req *querypb.ReleasePartitionsRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.ReleasePartitions(ctx, req)
}

func (s *mixCoordImpl) SyncNewCreatedPartition(ctx context.Context, req *querypb.SyncNewCreatedPartitionRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.SyncNewCreatedPartition(ctx, req)
}

func (s *mixCoordImpl) GetPartitionStates(ctx context.Context, req *querypb.GetPartitionStatesRequest) (*querypb.GetPartitionStatesResponse, error) {
	return s.queryCoordServer.GetPartitionStates(ctx, req)
}

func (s *mixCoordImpl) GetLoadSegmentInfo(ctx context.Context, req *querypb.GetSegmentInfoRequest) (*querypb.GetSegmentInfoResponse, error) {
	return s.queryCoordServer.GetLoadSegmentInfo(ctx, req)
}

func (s *mixCoordImpl) LoadBalance(ctx context.Context, req *querypb.LoadBalanceRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.LoadBalance(ctx, req)
}

func (s *mixCoordImpl) GetReplicas(ctx context.Context, req *milvuspb.GetReplicasRequest) (*milvuspb.GetReplicasResponse, error) {
	return s.queryCoordServer.GetReplicas(ctx, req)
}

func (s *mixCoordImpl) GetShardLeaders(ctx context.Context, req *querypb.GetShardLeadersRequest) (*querypb.GetShardLeadersResponse, error) {
	return s.queryCoordServer.GetShardLeaders(ctx, req)
}

func (s *mixCoordImpl) CreateResourceGroup(ctx context.Context, req *milvuspb.CreateResourceGroupRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.CreateResourceGroup(ctx, req)
}

func (s *mixCoordImpl) UpdateResourceGroups(ctx context.Context, req *querypb.UpdateResourceGroupsRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.UpdateResourceGroups(ctx, req)
}

func (s *mixCoordImpl) DropResourceGroup(ctx context.Context, req *milvuspb.DropResourceGroupRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.DropResourceGroup(ctx, req)
}

func (s *mixCoordImpl) TransferNode(ctx context.Context, req *milvuspb.TransferNodeRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.TransferNode(ctx, req)
}

func (s *mixCoordImpl) TransferReplica(ctx context.Context, req *querypb.TransferReplicaRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.TransferReplica(ctx, req)
}

func (s *mixCoordImpl) ListResourceGroups(ctx context.Context, req *milvuspb.ListResourceGroupsRequest) (*milvuspb.ListResourceGroupsResponse, error) {
	return s.queryCoordServer.ListResourceGroups(ctx, req)
}

func (s *mixCoordImpl) DescribeResourceGroup(ctx context.Context, req *querypb.DescribeResourceGroupRequest) (*querypb.DescribeResourceGroupResponse, error) {
	return s.queryCoordServer.DescribeResourceGroup(ctx, req)
}

func (s *mixCoordImpl) ListQueryNode(ctx context.Context, req *querypb.ListQueryNodeRequest) (*querypb.ListQueryNodeResponse, error) {
	return s.queryCoordServer.ListQueryNode(ctx, req)
}

func (s *mixCoordImpl) GetQueryNodeDistribution(ctx context.Context, req *querypb.GetQueryNodeDistributionRequest) (*querypb.GetQueryNodeDistributionResponse, error) {
	return s.queryCoordServer.GetQueryNodeDistribution(ctx, req)
}

func (s *mixCoordImpl) SuspendBalance(ctx context.Context, req *querypb.SuspendBalanceRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.SuspendBalance(ctx, req)
}

func (s *mixCoordImpl) ResumeBalance(ctx context.Context, req *querypb.ResumeBalanceRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.ResumeBalance(ctx, req)
}

func (s *mixCoordImpl) CheckBalanceStatus(ctx context.Context, req *querypb.CheckBalanceStatusRequest) (*querypb.CheckBalanceStatusResponse, error) {
	return s.queryCoordServer.CheckBalanceStatus(ctx, req)
}

func (s *mixCoordImpl) SuspendNode(ctx context.Context, req *querypb.SuspendNodeRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.SuspendNode(ctx, req)
}

func (s *mixCoordImpl) ResumeNode(ctx context.Context, req *querypb.ResumeNodeRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.ResumeNode(ctx, req)
}

func (s *mixCoordImpl) TransferSegment(ctx context.Context, req *querypb.TransferSegmentRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.TransferSegment(ctx, req)
}

func (s *mixCoordImpl) TransferChannel(ctx context.Context, req *querypb.TransferChannelRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.TransferChannel(ctx, req)
}

func (s *mixCoordImpl) CheckQueryNodeDistribution(ctx context.Context, req *querypb.CheckQueryNodeDistributionRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.CheckQueryNodeDistribution(ctx, req)
}

func (s *mixCoordImpl) UpdateLoadConfig(ctx context.Context, req *querypb.UpdateLoadConfigRequest) (*commonpb.Status, error) {
	return s.queryCoordServer.UpdateLoadConfig(ctx, req)
}

// DataCoordServer
func (s *mixCoordImpl) GetSegmentInfo(ctx context.Context, req *datapb.GetSegmentInfoRequest) (*datapb.GetSegmentInfoResponse, error) {
	return s.datacoordServer.GetSegmentInfo(ctx, req)
}

func (s *mixCoordImpl) Flush(ctx context.Context, req *datapb.FlushRequest) (*datapb.FlushResponse, error) {
	return s.datacoordServer.Flush(ctx, req)
}

func (s *mixCoordImpl) AssignSegmentID(ctx context.Context, req *datapb.AssignSegmentIDRequest) (*datapb.AssignSegmentIDResponse, error) {
	return s.datacoordServer.AssignSegmentID(ctx, req)
}

func (s *mixCoordImpl) GetSegmentStates(ctx context.Context, req *datapb.GetSegmentStatesRequest) (*datapb.GetSegmentStatesResponse, error) {
	return s.datacoordServer.GetSegmentStates(ctx, req)
}

func (s *mixCoordImpl) GetInsertBinlogPaths(ctx context.Context, req *datapb.GetInsertBinlogPathsRequest) (*datapb.GetInsertBinlogPathsResponse, error) {
	return s.datacoordServer.GetInsertBinlogPaths(ctx, req)
}

func (s *mixCoordImpl) GetCollectionStatistics(ctx context.Context, req *datapb.GetCollectionStatisticsRequest) (*datapb.GetCollectionStatisticsResponse, error) {
	return s.datacoordServer.GetCollectionStatistics(ctx, req)
}

func (s *mixCoordImpl) GetPartitionStatistics(ctx context.Context, req *datapb.GetPartitionStatisticsRequest) (*datapb.GetPartitionStatisticsResponse, error) {
	return s.datacoordServer.GetPartitionStatistics(ctx, req)
}

func (s *mixCoordImpl) GetSegmentInfoChannel(ctx context.Context, req *datapb.GetSegmentInfoChannelRequest) (*milvuspb.StringResponse, error) {
	return s.datacoordServer.GetSegmentInfoChannel(ctx, req)
}

func (s *mixCoordImpl) SaveBinlogPaths(ctx context.Context, req *datapb.SaveBinlogPathsRequest) (*commonpb.Status, error) {
	return s.datacoordServer.SaveBinlogPaths(ctx, req)
}

func (s *mixCoordImpl) GetRecoveryInfo(ctx context.Context, req *datapb.GetRecoveryInfoRequest) (*datapb.GetRecoveryInfoResponse, error) {
	return s.datacoordServer.GetRecoveryInfo(ctx, req)
}

func (s *mixCoordImpl) GetRecoveryInfoV2(ctx context.Context, req *datapb.GetRecoveryInfoRequestV2) (*datapb.GetRecoveryInfoResponseV2, error) {
	return s.datacoordServer.GetRecoveryInfoV2(ctx, req)
}

func (s *mixCoordImpl) GetChannelRecoveryInfo(ctx context.Context, req *datapb.GetChannelRecoveryInfoRequest) (*datapb.GetChannelRecoveryInfoResponse, error) {
	return s.datacoordServer.GetChannelRecoveryInfo(ctx, req)
}

func (s *mixCoordImpl) GetFlushedSegments(ctx context.Context, req *datapb.GetFlushedSegmentsRequest) (*datapb.GetFlushedSegmentsResponse, error) {
	return s.datacoordServer.GetFlushedSegments(ctx, req)
}

func (s *mixCoordImpl) GetSegmentsByStates(ctx context.Context, req *datapb.GetSegmentsByStatesRequest) (*datapb.GetSegmentsByStatesResponse, error) {
	return s.datacoordServer.GetSegmentsByStates(ctx, req)
}

func (s *mixCoordImpl) ManualCompaction(ctx context.Context, req *milvuspb.ManualCompactionRequest) (*milvuspb.ManualCompactionResponse, error) {
	return s.datacoordServer.ManualCompaction(ctx, req)
}

func (s *mixCoordImpl) GetCompactionState(ctx context.Context, req *milvuspb.GetCompactionStateRequest) (*milvuspb.GetCompactionStateResponse, error) {
	return s.datacoordServer.GetCompactionState(ctx, req)
}

func (s *mixCoordImpl) GetCompactionStateWithPlans(ctx context.Context, req *milvuspb.GetCompactionPlansRequest) (*milvuspb.GetCompactionPlansResponse, error) {
	return s.datacoordServer.GetCompactionStateWithPlans(ctx, req)
}

func (s *mixCoordImpl) WatchChannels(ctx context.Context, req *datapb.WatchChannelsRequest) (*datapb.WatchChannelsResponse, error) {
	return s.datacoordServer.WatchChannels(ctx, req)
}

func (s *mixCoordImpl) GetFlushState(ctx context.Context, req *datapb.GetFlushStateRequest) (*milvuspb.GetFlushStateResponse, error) {
	return s.datacoordServer.GetFlushState(ctx, req)
}

func (s *mixCoordImpl) GetFlushAllState(ctx context.Context, req *milvuspb.GetFlushAllStateRequest) (*milvuspb.GetFlushAllStateResponse, error) {
	return s.datacoordServer.GetFlushAllState(ctx, req)
}

func (s *mixCoordImpl) DropVirtualChannel(ctx context.Context, req *datapb.DropVirtualChannelRequest) (*datapb.DropVirtualChannelResponse, error) {
	return s.datacoordServer.DropVirtualChannel(ctx, req)
}

func (s *mixCoordImpl) SetSegmentState(ctx context.Context, req *datapb.SetSegmentStateRequest) (*datapb.SetSegmentStateResponse, error) {
	return s.datacoordServer.SetSegmentState(ctx, req)
}

func (s *mixCoordImpl) UpdateSegmentStatistics(ctx context.Context, req *datapb.UpdateSegmentStatisticsRequest) (*commonpb.Status, error) {
	return s.datacoordServer.UpdateSegmentStatistics(ctx, req)
}

func (s *mixCoordImpl) UpdateChannelCheckpoint(ctx context.Context, req *datapb.UpdateChannelCheckpointRequest) (*commonpb.Status, error) {
	return s.datacoordServer.UpdateChannelCheckpoint(ctx, req)
}

func (s *mixCoordImpl) MarkSegmentsDropped(ctx context.Context, req *datapb.MarkSegmentsDroppedRequest) (*commonpb.Status, error) {
	return s.datacoordServer.MarkSegmentsDropped(ctx, req)
}

func (s *mixCoordImpl) BroadcastAlteredCollection(ctx context.Context, req *datapb.AlterCollectionRequest) (*commonpb.Status, error) {
	return s.datacoordServer.BroadcastAlteredCollection(ctx, req)
}

func (s *mixCoordImpl) GcConfirm(ctx context.Context, req *datapb.GcConfirmRequest) (*datapb.GcConfirmResponse, error) {
	return s.datacoordServer.GcConfirm(ctx, req)
}

func (s *mixCoordImpl) CreateIndex(ctx context.Context, req *indexpb.CreateIndexRequest) (*commonpb.Status, error) {
	return s.datacoordServer.CreateIndex(ctx, req)
}

func (s *mixCoordImpl) AlterIndex(ctx context.Context, req *indexpb.AlterIndexRequest) (*commonpb.Status, error) {
	return s.datacoordServer.AlterIndex(ctx, req)
}

func (s *mixCoordImpl) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRequest) (*indexpb.GetIndexStateResponse, error) {
	return s.datacoordServer.GetIndexState(ctx, req)
}

func (s *mixCoordImpl) GetSegmentIndexState(ctx context.Context, req *indexpb.GetSegmentIndexStateRequest) (*indexpb.GetSegmentIndexStateResponse, error) {
	return s.datacoordServer.GetSegmentIndexState(ctx, req)
}

func (s *mixCoordImpl) GetIndexInfos(ctx context.Context, req *indexpb.GetIndexInfoRequest) (*indexpb.GetIndexInfoResponse, error) {
	return s.datacoordServer.GetIndexInfos(ctx, req)
}

func (s *mixCoordImpl) DescribeIndex(ctx context.Context, req *indexpb.DescribeIndexRequest) (*indexpb.DescribeIndexResponse, error) {
	return s.datacoordServer.DescribeIndex(ctx, req)
}

func (s *mixCoordImpl) GetIndexStatistics(ctx context.Context, req *indexpb.GetIndexStatisticsRequest) (*indexpb.GetIndexStatisticsResponse, error) {
	return s.datacoordServer.GetIndexStatistics(ctx, req)
}

func (s *mixCoordImpl) DropIndex(ctx context.Context, req *indexpb.DropIndexRequest) (*commonpb.Status, error) {
	return s.datacoordServer.DropIndex(ctx, req)
}

func (s *mixCoordImpl) GetIndexBuildProgress(ctx context.Context, req *indexpb.GetIndexBuildProgressRequest) (*indexpb.GetIndexBuildProgressResponse, error) {
	return s.datacoordServer.GetIndexBuildProgress(ctx, req)
}

func (s *mixCoordImpl) ReportDataNodeTtMsgs(ctx context.Context, req *datapb.ReportDataNodeTtMsgsRequest) (*commonpb.Status, error) {
	return s.datacoordServer.ReportDataNodeTtMsgs(ctx, req)
}

func (s *mixCoordImpl) GcControl(ctx context.Context, req *datapb.GcControlRequest) (*commonpb.Status, error) {
	return s.datacoordServer.GcControl(ctx, req)
}

func (s *mixCoordImpl) ImportV2(ctx context.Context, req *internalpb.ImportRequestInternal) (*internalpb.ImportResponse, error) {
	return s.datacoordServer.ImportV2(ctx, req)
}

func (s *mixCoordImpl) GetImportProgress(ctx context.Context, req *internalpb.GetImportProgressRequest) (*internalpb.GetImportProgressResponse, error) {
	return s.datacoordServer.GetImportProgress(ctx, req)
}

func (s *mixCoordImpl) ListImports(ctx context.Context, req *internalpb.ListImportsRequestInternal) (*internalpb.ListImportsResponse, error) {
	return s.datacoordServer.ListImports(ctx, req)
}

func (s *mixCoordImpl) CommitImport(ctx context.Context, req *datapb.CommitImportRequest) (*commonpb.Status, error) {
	return s.datacoordServer.CommitImport(ctx, req)
}

func (s *mixCoordImpl) AbortImport(ctx context.Context, req *datapb.AbortImportRequest) (*commonpb.Status, error) {
	return s.datacoordServer.AbortImport(ctx, req)
}

func (s *mixCoordImpl) HandleCommitVchannel(ctx context.Context, req *datapb.HandleCommitVchannelRequest) (*commonpb.Status, error) {
	return s.datacoordServer.HandleCommitVchannel(ctx, req)
}

func (s *mixCoordImpl) ListIndexes(ctx context.Context, req *indexpb.ListIndexesRequest) (*indexpb.ListIndexesResponse, error) {
	return s.datacoordServer.ListIndexes(ctx, req)
}

func (s *mixCoordImpl) AllocSegment(ctx context.Context, req *datapb.AllocSegmentRequest) (*datapb.AllocSegmentResponse, error) {
	return s.datacoordServer.AllocSegment(ctx, req)
}

func (s *mixCoordImpl) NotifyDropPartition(ctx context.Context, channel string, partitionIDs []int64) error {
	return s.datacoordServer.NotifyDropPartition(ctx, channel, partitionIDs)
}

// RegisterStreamingCoordGRPCService registers the grpc service of streaming coordinator.
func (s *mixCoordImpl) RegisterStreamingCoordGRPCService(server *grpc.Server) {
	s.streamingCoord.RegisterGRPCService(server)
}

func (s *mixCoordImpl) GetQuotaMetrics(ctx context.Context, req *internalpb.GetQuotaMetricsRequest) (*internalpb.GetQuotaMetricsResponse, error) {
	return s.rootcoordServer.GetQuotaMetrics(ctx, req)
}

func (s *mixCoordImpl) ListLoadedSegments(ctx context.Context, req *querypb.ListLoadedSegmentsRequest) (*querypb.ListLoadedSegmentsResponse, error) {
	return s.queryCoordServer.ListLoadedSegments(ctx, req)
}

func (s *mixCoordImpl) RunAnalyzer(ctx context.Context, req *querypb.RunAnalyzerRequest) (*milvuspb.RunAnalyzerResponse, error) {
	return s.queryCoordServer.RunAnalyzer(ctx, req)
}

func (s *mixCoordImpl) ValidateAnalyzer(ctx context.Context, req *querypb.ValidateAnalyzerRequest) (*querypb.ValidateAnalyzerResponse, error) {
	return s.queryCoordServer.ValidateAnalyzer(ctx, req)
}

func (s *mixCoordImpl) ComputePhraseMatchSlop(ctx context.Context, req *querypb.ComputePhraseMatchSlopRequest) (*querypb.ComputePhraseMatchSlopResponse, error) {
	return s.queryCoordServer.ComputePhraseMatchSlop(ctx, req)
}

func (s *mixCoordImpl) ClearReadTaskQueue(ctx context.Context, req *internalpb.ClearReadTaskQueueRequest) (*internalpb.ClearReadTaskQueueResponse, error) {
	resp := &internalpb.ClearReadTaskQueueResponse{Status: merr.Success()}
	if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
		resp.Status = merr.Status(err)
		return resp, nil
	}

	var errs []error
	proxyResp, err := s.rootcoordServer.ClearReadTaskQueue(ctx, req)
	if err != nil {
		errs = append(errs, err)
	} else if proxyResp != nil {
		resp.Results = append(resp.Results, proxyResp.GetResults()...)
		resp.ProxyQueuedCleared += proxyResp.GetProxyQueuedCleared()
		if !merr.Ok(proxyResp.GetStatus()) {
			errs = append(errs, merr.Error(proxyResp.GetStatus()))
		}
	}

	queryNodeResp, err := s.queryCoordServer.ClearReadTaskQueue(ctx, req)
	if err != nil {
		errs = append(errs, err)
	} else if queryNodeResp != nil {
		resp.Results = append(resp.Results, queryNodeResp.GetResults()...)
		resp.QuerynodeQueuedCleared += queryNodeResp.GetQuerynodeQueuedCleared()
		resp.QueuedNqCleared += queryNodeResp.GetQueuedNqCleared()
		if !merr.Ok(queryNodeResp.GetStatus()) {
			errs = append(errs, merr.Error(queryNodeResp.GetStatus()))
		}
	}

	if err := merr.Combine(errs...); err != nil {
		resp.Status = merr.Status(err)
	}

	mlog.Info(ctx, "cleared cluster read task queues",
		mlog.String("taskType", req.GetTaskType()),
		mlog.String("reason", req.GetReason()),
		mlog.Int64("proxyQueuedCleared", resp.GetProxyQueuedCleared()),
		mlog.Int64("queryNodeQueuedCleared", resp.GetQuerynodeQueuedCleared()),
		mlog.Int64("queuedNQCleared", resp.GetQueuedNqCleared()),
		mlog.Int("results", len(resp.GetResults())),
		mlog.Err(merr.Error(resp.GetStatus())))
	return resp, nil
}

func (s *mixCoordImpl) FlushAll(ctx context.Context, req *datapb.FlushAllRequest) (*datapb.FlushAllResponse, error) {
	return s.datacoordServer.FlushAll(ctx, req)
}

// AddFileResource add file resource
func (s *mixCoordImpl) AddFileResource(ctx context.Context, req *milvuspb.AddFileResourceRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.AddFileResource(ctx, req)
}

// RemoveFileResource remove file resource
func (s *mixCoordImpl) RemoveFileResource(ctx context.Context, req *milvuspb.RemoveFileResourceRequest) (*commonpb.Status, error) {
	return s.rootcoordServer.RemoveFileResource(ctx, req)
}

// ListFileResources list file resources
func (s *mixCoordImpl) ListFileResources(ctx context.Context, req *milvuspb.ListFileResourcesRequest) (*milvuspb.ListFileResourcesResponse, error) {
	return s.rootcoordServer.ListFileResources(ctx, req)
}

// TruncateCollection truncate collection
func (s *mixCoordImpl) TruncateCollection(ctx context.Context, req *milvuspb.TruncateCollectionRequest) (*milvuspb.TruncateCollectionResponse, error) {
	return s.rootcoordServer.TruncateCollection(ctx, req)
}

// DropSegmentsByTime drop segments by time for TruncateCollection
func (s *mixCoordImpl) DropSegmentsByTime(ctx context.Context, collectionID int64, flushTsList map[string]uint64) error {
	return s.datacoordServer.DropSegmentsByTime(ctx, collectionID, flushTsList)
}

// ManualUpdateCurrentTarget manually update current target for TruncateCollection
func (s *mixCoordImpl) ManualUpdateCurrentTarget(ctx context.Context, collectionID int64) error {
	return s.queryCoordServer.ManualUpdateCurrentTarget(ctx, collectionID)
}

func (s *mixCoordImpl) CreateSnapshot(ctx context.Context, req *datapb.CreateSnapshotRequest) (*commonpb.Status, error) {
	return s.datacoordServer.CreateSnapshot(ctx, req)
}

func (s *mixCoordImpl) DropSnapshot(ctx context.Context, req *datapb.DropSnapshotRequest) (*commonpb.Status, error) {
	return s.datacoordServer.DropSnapshot(ctx, req)
}

func (s *mixCoordImpl) DescribeSnapshot(ctx context.Context, req *datapb.DescribeSnapshotRequest) (*datapb.DescribeSnapshotResponse, error) {
	return s.datacoordServer.DescribeSnapshot(ctx, req)
}

func (s *mixCoordImpl) RestoreSnapshot(ctx context.Context, req *datapb.RestoreSnapshotRequest) (*datapb.RestoreSnapshotResponse, error) {
	return s.datacoordServer.RestoreSnapshot(ctx, req)
}

func (s *mixCoordImpl) ExportSnapshot(ctx context.Context, req *datapb.ExportSnapshotRequest) (*datapb.ExportSnapshotResponse, error) {
	return s.datacoordServer.ExportSnapshot(ctx, req)
}

func (s *mixCoordImpl) GetRestoreSnapshotState(ctx context.Context, req *datapb.GetRestoreSnapshotStateRequest) (*datapb.GetRestoreSnapshotStateResponse, error) {
	return s.datacoordServer.GetRestoreSnapshotState(ctx, req)
}

func (s *mixCoordImpl) ListRestoreSnapshotJobs(ctx context.Context, req *datapb.ListRestoreSnapshotJobsRequest) (*datapb.ListRestoreSnapshotJobsResponse, error) {
	return s.datacoordServer.ListRestoreSnapshotJobs(ctx, req)
}

func (s *mixCoordImpl) ListSnapshots(ctx context.Context, req *datapb.ListSnapshotsRequest) (*datapb.ListSnapshotsResponse, error) {
	return s.datacoordServer.ListSnapshots(ctx, req)
}

func (s *mixCoordImpl) PinSnapshotData(ctx context.Context, req *datapb.PinSnapshotDataRequest) (*datapb.PinSnapshotDataResponse, error) {
	return s.datacoordServer.PinSnapshotData(ctx, req)
}

func (s *mixCoordImpl) UnpinSnapshotData(ctx context.Context, req *datapb.UnpinSnapshotDataRequest) (*commonpb.Status, error) {
	return s.datacoordServer.UnpinSnapshotData(ctx, req)
}

func (s *mixCoordImpl) BatchUpdateManifest(ctx context.Context, req *datapb.BatchUpdateManifestRequest) (*commonpb.Status, error) {
	return s.datacoordServer.BatchUpdateManifest(ctx, req)
}

func (s *mixCoordImpl) CommitBackfillResult(ctx context.Context, req *datapb.CommitBackfillResultRequest) (*datapb.CommitBackfillResultResponse, error) {
	return s.datacoordServer.CommitBackfillResult(ctx, req)
}

// Client Telemetry methods - forwarded to rootcoord

func (s *mixCoordImpl) ClientHeartbeat(ctx context.Context, req *milvuspb.ClientHeartbeatRequest) (*milvuspb.ClientHeartbeatResponse, error) {
	return s.rootcoordServer.ClientHeartbeat(ctx, req)
}

func (s *mixCoordImpl) GetClientTelemetry(ctx context.Context, req *milvuspb.GetClientTelemetryRequest) (*milvuspb.GetClientTelemetryResponse, error) {
	return s.rootcoordServer.GetClientTelemetry(ctx, req)
}

func (s *mixCoordImpl) PushClientCommand(ctx context.Context, req *milvuspb.PushClientCommandRequest) (*milvuspb.PushClientCommandResponse, error) {
	return s.rootcoordServer.PushClientCommand(ctx, req)
}

func (s *mixCoordImpl) DeleteClientCommand(ctx context.Context, req *milvuspb.DeleteClientCommandRequest) (*milvuspb.DeleteClientCommandResponse, error) {
	return s.rootcoordServer.DeleteClientCommand(ctx, req)
}

func (s *mixCoordImpl) RefreshExternalCollection(ctx context.Context, req *datapb.RefreshExternalCollectionRequest) (*datapb.RefreshExternalCollectionResponse, error) {
	return s.datacoordServer.RefreshExternalCollection(ctx, req)
}

func (s *mixCoordImpl) GetRefreshExternalCollectionProgress(ctx context.Context, req *datapb.GetRefreshExternalCollectionProgressRequest) (*datapb.GetRefreshExternalCollectionProgressResponse, error) {
	return s.datacoordServer.GetRefreshExternalCollectionProgress(ctx, req)
}

func (s *mixCoordImpl) ListRefreshExternalCollectionJobs(ctx context.Context, req *datapb.ListRefreshExternalCollectionJobsRequest) (*datapb.ListRefreshExternalCollectionJobsResponse, error) {
	return s.datacoordServer.ListRefreshExternalCollectionJobs(ctx, req)
}
