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

Code before:
	// WatchSubscription represents a subscription to watch a query.
	WatchSubscription struct {
		Subscribe bool                    // true for subscribe, false for unsubscribe
		Query     DSQLQuery               // query to watch
		ClientFD  int                     // client file descriptor
		CacheChan chan *[]dstore.KeyValue // channel to receive cache data for this query
	}

	// AdhocQueryResult represents the result of an adhoc query.
	AdhocQueryResult struct {
		Result      *[]dstore.DSQLQueryResultRow
		Fingerprint string
		Err         error
	}

	// 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)

Review comment (suggestion):
The fingerprinting mechanism is probably internal to the query watcher, we can make it private.
Also maybe a better name for this is `fingerprintToQueryMap`

Code after:
		Query     DSQLQuery               // query to watch
		ClientFD  int                     // client file descriptor
		CacheChan chan *[]dstore.KeyValue // channel to receive cache data for this query
	}

	// AdhocQueryResult represents the result of an adhoc query.
	AdhocQueryResult struct {
		Result      *[]dstore.DSQLQueryResultRow
		Fingerprint string
		Err         error
	}

	// AdhocQuery represents an adhoc query request.
	AdhocQuery struct {
		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