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

Code before:
			log.Warnf("Fingerprint not found in cacheStore: %s", fingerPrint)
			return true
		}

		clients := value.(*sync.Map)
		/*
			For all the clients that are watching this fingerprint, we need to find the queries they had requested to run, and then send them responses based on the queries they had fired.
		*/
		clients.Range(func(clientKey, _ interface{}) bool {
			clientFd := clientKey.(int)

			query := queries[clientFd]
			if !regex.WildCardMatch(query.KeyRegex, event.Key) {
				return true
			}

			w.updateQueryCache(fingerPrint.(string), event)
			queryResult, err := w.runQuery(query)
			if err != nil {
				log.Error(err)
				return true
			}

			w.notifyClient(query, clientFd, queryResult)
			return true
		})

		return true
	})
}

// updateQueryCache updates the query cache based on the watch event.
func (w *QWatchManager) updateQueryCache(queryFingerprint string, event dstore.WatchEvent) {
	w.QueryCacheMu.Lock()
	defer w.QueryCacheMu.Unlock()

	store, ok := w.QueryCache.Get(queryFingerprint)
	if !ok {
		log.Warnf("Fingerprint not found in cacheStore: %s", queryFingerprint)
		return
	}

	switch event.Operation {
	case constants.Set:
		((*swiss.Map[string, *dstore.Obj])(store)).Put(event.Key, event.Value)
	case constants.Del:
		((*swiss.Map[string, *dstore.Obj])(store)).Delete(event.Key)
	default:
		log.Warnf("Unknown operation: %s", event.Operation)
	}
}

Review comment (performance):
I'm a little confused about this looping mechanism here.

Shouldn't it be structured as follows:
```golang
w.WatchList.Range(
    // Looping over all queries.

    query := key.(DSQLQuery)
    clients := value.(*sync.Map)

    if !regex.WildCardMatch() {
        // continue if no match
    }

    // 1. If you reach here it means we are interested in this query.
    
    // 2. Fetch this query's fingerprint, and update the cache (you could maintain a local map keeping track of which fingerprints' cache we have updated, to avoid dual updates - though the operations are idempotent so it may not be necessary)

    // 3. Execute the query and fetch the result.
    queryResult, err := w.runQuery(&query)

    // 4. Notify clients
    w.notifyClients(&query, clients, queryResult)
)
```

I think the current approach will lead to high number of query re-executions.

Code after:
			return true
		}

		w.updateQueryCache(query.Fingerprint, event)

		queryResult, err := w.runQuery(&query)
		if err != nil {
			log.Error(err)
			return true
		}

		w.notifyClients(&query, clients, queryResult)
		return true
	})
}

// updateQueryCache updates the query cache based on the watch event.
func (w *QueryManager) updateQueryCache(queryFingerprint string, event dstore.WatchEvent) {
	w.QueryCacheMu.Lock()
	defer w.QueryCacheMu.Unlock()

	store, ok := w.QueryCache.Get(queryFingerprint)
	if !ok {
		log.Warnf("Fingerprint not found in cacheStore: %s", queryFingerprint)
		return
	}

	switch event.Operation {
	case dstore.Set:
		((*swiss.Map[string, *dstore.Obj])(store)).Put(event.Key, event.Value)
	case dstore.Del:
		((*swiss.Map[string, *dstore.Obj])(store)).Delete(event.Key)
	default:
		log.Warnf("Unknown operation: %s", event.Operation)
	}
}

// notifyClient notifies all clients watching a query about the new result.
func (w *QueryManager) notifyClients(query *DSQLQuery, clients *sync.Map, queryResult *[]dstore.DSQLQueryResultRow) {
	encodedResult := clientio.Encode(CreatePushResponse(query, queryResult), false)
	// Send the result to the client
	clients.Range(func(clientKey, _ interface{}) bool {
		clientFD := clientKey.(int)
		_, err := syscall.Write(clientFD, encodedResult)
		if err != nil {
			w.removeWatcher(query, clientFD)
		}
		return true
	})
}