<commit_before>// Copyright 2014, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package vtgate

// This is a V3 file. Do not intermix with V2.

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sync"

	log "github.com/golang/glog"
	"golang.org/x/net/context"

	"github.com/youtube/vitess/go/acl"
	"github.com/youtube/vitess/go/cache"
	"github.com/youtube/vitess/go/vt/topo"
	"github.com/youtube/vitess/go/vt/vtgate/engine"
	"github.com/youtube/vitess/go/vt/vtgate/planbuilder"
	"github.com/youtube/vitess/go/vt/vtgate/vindexes"
)

// Planner is used to compute the plan. It contains
// the vschema, and has a cache of previous computed plans.
type Planner struct {
	serv    topo.SrvTopoServer
	cell    string
	mu      sync.Mutex
	vschema *vindexes.VSchema
	plans   *cache.LRUCache
}

var once sync.Once

// NewPlanner creates a new planner for VTGate.
// It will watch the vschema in the topology until the ctx is closed.
func NewPlanner(ctx context.Context, serv topo.SrvTopoServer, cell string, cacheSize int) *Planner {
	plr := &Planner{
		serv:  serv,
		cell:  cell,
		plans: cache.NewLRUCache(int64(cacheSize)),
	}
	plr.WatchVSchema(ctx)
	once.Do(func() {
		http.Handle("/debug/query_plans", plr)
		http.Handle("/debug/vschema", plr)
	})
	return plr
}

// WatchVSchema watches the VSchema from the topo. The function does
// not return an error. It instead logs warnings on failure.
// We get the list of keyspaces to watch at the beginning (because it's easier)
// and we don't watch that. So when adding a new keyspace, for now,
// vtgate needs to be restarted (after having rebuilt the serving graph).
// This could be fixed by adding a WatchSrvKeyspaceNames API to topo.Server,
// and when it triggers we'd diff the values, and terminate the missing
// SrvKeyspaces, and start watching the new SrvKeyspaces.
func (plr *Planner) WatchVSchema(ctx context.Context) {
	keyspaces, err := plr.serv.GetSrvKeyspaceNames(ctx, plr.cell)
	if err != nil {
		log.Warningf("Error loading vschema: could not read keyspaces: %v", err)
		return
	}

	// we will wait until we get the first value for each keyspace
	// before returning
	var wg sync.WaitGroup

	// mu protects formal
	var mu sync.Mutex
	formal := &vindexes.VSchemaFormal{
		Keyspaces: make(map[string]vindexes.KeyspaceFormal),
	}
	processKeyspace := func(keyspace, kschema string) {
		// Note we use a closure here to allow for defer()
		// to work properly, and also to let the caller
		// do more things even if this fails.

		// unpack the new VSchema, or skip it
		var kformal vindexes.KeyspaceFormal
		if err := json.Unmarshal([]byte(kschema), &kformal); err != nil {
			log.Warningf("Error unmarshalling vschema for keyspace %s: %v", keyspace, err)
			return
		}

		// rebuild the new component
		mu.Lock()
		defer mu.Unlock()
		formal.Keyspaces[keyspace] = kformal
		vschema, err := vindexes.BuildVSchema(formal)
		if err != nil {
			log.Warningf("Error creating VSchema: %v", err)
			return
		}

		plr.mu.Lock()
		plr.vschema = vschema
		plr.mu.Unlock()
		plr.plans.Clear()
	}

	for _, keyspace := range keyspaces {
		wg.Add(1)
		go func(keyspace string) {
			gotFirstValue := false

			notifications, err := plr.serv.WatchVSchema(ctx, keyspace)
			if err != nil {
				log.Warningf("Error watching vschema for keyspace %s, will not watch it: %v", keyspace, err)
				return
			}

			for kschema := range notifications {
				processKeyspace(keyspace, kschema)

				if !gotFirstValue {
					gotFirstValue = true
					wg.Done()
				}
			}
		}(keyspace)
	}

	wg.Wait()
}

// VSchema returns the VSchema.
func (plr *Planner) VSchema() *vindexes.VSchema {
	plr.mu.Lock()
	defer plr.mu.Unlock()
	return plr.vschema
}

// GetPlan computes the plan for the given query. If one is in
// the cache, it reuses it.
func (plr *Planner) GetPlan(sql, keyspace string) (*engine.Plan, error) {<commit_msg>Only log v3 schema errors after all keyspaces.<commit_after>// Copyright 2014, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package vtgate

// This is a V3 file. Do not intermix with V2.

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sync"

	log "github.com/golang/glog"
	"golang.org/x/net/context"

	"github.com/youtube/vitess/go/acl"
	"github.com/youtube/vitess/go/cache"
	"github.com/youtube/vitess/go/vt/topo"
	"github.com/youtube/vitess/go/vt/vtgate/engine"
	"github.com/youtube/vitess/go/vt/vtgate/planbuilder"
	"github.com/youtube/vitess/go/vt/vtgate/vindexes"
)

// Planner is used to compute the plan. It contains
// the vschema, and has a cache of previous computed plans.
type Planner struct {
	serv    topo.SrvTopoServer
	cell    string
	mu      sync.Mutex
	vschema *vindexes.VSchema
	plans   *cache.LRUCache
}

var once sync.Once

// NewPlanner creates a new planner for VTGate.
// It will watch the vschema in the topology until the ctx is closed.
func NewPlanner(ctx context.Context, serv topo.SrvTopoServer, cell string, cacheSize int) *Planner {
	plr := &Planner{
		serv:  serv,
		cell:  cell,
		plans: cache.NewLRUCache(int64(cacheSize)),
	}
	plr.WatchVSchema(ctx)
	once.Do(func() {
		http.Handle("/debug/query_plans", plr)
		http.Handle("/debug/vschema", plr)
	})
	return plr
}

// WatchVSchema watches the VSchema from the topo. The function does
// not return an error. It instead logs warnings on failure.
// We get the list of keyspaces to watch at the beginning (because it's easier)
// and we don't watch that. So when adding a new keyspace, for now,
// vtgate needs to be restarted (after having rebuilt the serving graph).
// This could be fixed by adding a WatchSrvKeyspaceNames API to topo.Server,
// and when it triggers we'd diff the values, and terminate the missing
// SrvKeyspaces, and start watching the new SrvKeyspaces.
func (plr *Planner) WatchVSchema(ctx context.Context) {
	keyspaces, err := plr.serv.GetSrvKeyspaceNames(ctx, plr.cell)
	if err != nil {
		log.Warningf("Error loading vschema: could not read keyspaces: %v", err)
		return
	}

	// we will wait until we get the first value for each keyspace
	// before returning
	var wg sync.WaitGroup

	// mu protects formal
	var mu sync.Mutex
	formal := &vindexes.VSchemaFormal{
		Keyspaces: make(map[string]vindexes.KeyspaceFormal),
	}
	processKeyspace := func(keyspace, kschema string) error {
		// Note we use a closure here to allow for defer()
		// to work properly, and also to let the caller
		// do more things even if this fails.

		// unpack the new VSchema, or skip it
		var kformal vindexes.KeyspaceFormal
		if err := json.Unmarshal([]byte(kschema), &kformal); err != nil {
			return fmt.Errorf("Error unmarshalling vschema for keyspace %s: %v", keyspace, err)
		}

		// rebuild the new component
		mu.Lock()
		defer mu.Unlock()
		formal.Keyspaces[keyspace] = kformal
		vschema, err := vindexes.BuildVSchema(formal)
		if err != nil {
			return fmt.Errorf("Error creating VSchema: %v", err)
		}

		plr.mu.Lock()
		plr.vschema = vschema
		plr.mu.Unlock()
		plr.plans.Clear()

		return nil
	}

	lastValueIndex := len(keyspaces) - 1
	for i, keyspace := range keyspaces {
		wg.Add(1)
		go func(i int, keyspace string) {
			gotFirstValue := false

			notifications, err := plr.serv.WatchVSchema(ctx, keyspace)
			if err != nil {
				log.Warningf("Error watching vschema for keyspace %s, will not watch it: %v", keyspace, err)
				wg.Done()
				return
			}

			for kschema := range notifications {
				err := processKeyspace(keyspace, kschema)

				if !gotFirstValue {
					gotFirstValue = true

					// only log the last error, as other
					// errors may need the next values
					if err != nil && i == lastValueIndex {
						log.Warningf("%v", err)
					}

					wg.Done()
				} else {
					// always log a refresh that fails
					if err != nil {
						log.Warningf("%v", err)
					}
				}
			}
		}(i, keyspace)
	}

	wg.Wait()
}

// VSchema returns the VSchema.
func (plr *Planner) VSchema() *vindexes.VSchema {
	plr.mu.Lock()
	defer plr.mu.Unlock()
	return plr.vschema
}

// GetPlan computes the plan for the given query. If one is in
// the cache, it reuses it.
func (plr *Planner) GetPlan(sql, keyspace string) (*engine.Plan, error) {
{'fraction_non_alphanumeric': 0.1119942196531792, 'fraction_numerical': 0.0024084778420038534, 'mean_word_length': 3.368753287743293, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7623751', 'n_tokens_mistral': 3110, 'n_tokens_neox': 2821, 'n_words': 927}