Repository: DiceDB/dice
File: internal/querywatcher/query_watcher.go (Go)

Code before:

	// AdhocQuery represents an adhoc query request.
	AdhocQuery struct {
		Query        DSQLQuery
		ResponseChan chan AdhocQueryResult
	}

	// QWatchManager watches for changes in keys and notifies clients.
	QWatchManager struct {
		WatchList        sync.Map                               // WatchList is a map of fingerprints to their respective clients, type: map[string]*sync.Map[int]struct{}
		QueryCache       *swiss.Map[string, cacheStore]         // QueryCache is a map of fingerprints to their respective data caches
		FingerPrintCache *swiss.Map[string, map[int]*DSQLQuery] // FingerPrintCache is a map of fingerprint to the respective query with the ClientFD
		QueryCacheMu     sync.RWMutex
	}
)

var (
	// WatchSubscriptionChan is the channel to receive updates about query subscriptions.
	WatchSubscriptionChan chan WatchSubscription

	// AdhocQueryChan is the channel to receive adhoc queries.
	AdhocQueryChan chan AdhocQuery
)

// NewQWatchManager initializes a new QWatchManager.
func NewQWatchManager() *QWatchManager {
	WatchSubscriptionChan = make(chan WatchSubscription)
	AdhocQueryChan = make(chan AdhocQuery, 1000)
	return &QWatchManager{
		WatchList:        sync.Map{},
		QueryCache:       swiss.New[string, cacheStore](0),
		FingerPrintCache: swiss.New[string, map[int]*DSQLQuery](0),
	}
}

func newCacheStore() cacheStore {
	return swiss.New[string, *dstore.Obj](0)
}

// Run starts the QWatchManager's main loops.
func (w *QWatchManager) Run(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	var wg sync.WaitGroup

	wg.Add(3)
	go func() {
		defer wg.Done()
		w.listenForSubscriptions(ctx)
	}()

	go func() {
		defer wg.Done()

Review comment (suggestion):
I think a better name might be QueryManager. This component could be used to serve ad-hoc queries in the future (query execution without subscriptions) so tying the name to a particular command may not be a good idea.

Code after:
		Query        DSQLQuery
		ResponseChan chan AdhocQueryResult
	}

	// QueryManager watches for changes in keys and notifies clients.
	QueryManager struct {
		WatchList    sync.Map                       // WatchList is a map of query string to their respective clients, type: map[string]*sync.Map[int]struct{}
		QueryCache   *swiss.Map[string, cacheStore] // QueryCache is a map of fingerprints to their respective data caches
		QueryCacheMu sync.RWMutex
	}
)

var (
	// WatchSubscriptionChan is the channel to receive updates about query subscriptions.
	WatchSubscriptionChan chan WatchSubscription

	// AdhocQueryChan is the channel to receive adhoc queries.
	AdhocQueryChan chan AdhocQuery
)

// NewQueryManager initializes a new QueryManager.
func NewQueryManager() *QueryManager {
	WatchSubscriptionChan = make(chan WatchSubscription)
	AdhocQueryChan = make(chan AdhocQuery, 1000)
	return &QueryManager{
		WatchList:  sync.Map{},
		QueryCache: swiss.New[string, cacheStore](0),
	}
}

func newCacheStore() cacheStore {
	return swiss.New[string, *dstore.Obj](0)
}

// Run starts the QueryManager's main loops.
func (w *QueryManager) Run(ctx context.Context, watchChan <-chan dstore.WatchEvent) {
	var wg sync.WaitGroup

	wg.Add(3)
	go func() {
		defer wg.Done()
		w.listenForSubscriptions(ctx)
	}()

	go func() {
		defer wg.Done()
		w.watchKeys(ctx, watchChan)
	}()

	go func() {
		defer wg.Done()