// copy phase:
//	- copy the data from source tablets to destination masters (with replication on)
// Assumes that the schema has already been created on each destination tablet
// (probably from vtctl's CopySchemaShard)
507282
func (scw *SplitCloneWorker) clone(ctx context.Context, state StatusWorkerState) error {
	if state != WorkerStateCloneOnline && state != WorkerStateCloneOffline {
		panic(fmt.Sprintf("invalid state passed to clone(): %v", state))
	}
	scw.setState(state)
	start := time.Now()
	defer func() {
		statsStateDurationsNs.Set(string(state), time.Since(start).Nanoseconds())
	}()

	var firstSourceTablet, err = scw.findFirstSourceTablet(ctx, state)
	if err != nil {
		return err
	}

	statsCounters, tableStatusList := scw.getCounters(state)

	// The throttlers exist only for the duration of this clone() call.
	// That means a SplitClone invocation with both online and offline phases
	// will create throttlers for each phase.
	if err := scw.createThrottlers(); err != nil {
		return err
	}
	defer scw.closeThrottlers()

	sourceSchemaDefinition, err := scw.getSourceSchema(ctx, firstSourceTablet)
	if err != nil {
		return err
	}
	scw.wr.Logger().Infof("Source tablet 0 has %v tables to copy", len(sourceSchemaDefinition.TableDefinitions))
	tableStatusList.initialize(sourceSchemaDefinition)

	var firstError error

	ctx, cancelCopy := context.WithCancel(ctx)
	defer cancelCopy()
	processError := func(format string, args ...interface{}) {
		// in theory we could have two threads see firstError as null and both write to the variable
		// that should not cause any problems though - canceling and logging is concurrently safe,
		// and overwriting the variable will not cause any problems
		scw.wr.Logger().Errorf(format, args...)
		if firstError == nil {
			firstError = vterrors.Errorf(vtrpc.Code_INTERNAL, format, args...)
			cancelCopy()
		}
	}

	// NOTE: Code below this point must *not* use "return" to exit this Go routine
	// early. Instead, "processError" must be called to cancel the context. This
	// way all launched Go routines will terminate as well.
	// (However, "return" in a new Go routine, i.e. not this one, is fine.)
	// If Go routines have already been started, make sure that Wait() on the
	// respective WaitGroup is called as well e.g. "destinationWaitGroup.Wait()"
	// must always be reached. Waiting for the Go routines is important to avoid
	// races between "defer throttler.ThreadFinished()" (must be executed first)
	// and "defer scw.closeThrottlers()". Otherwise, vtworker will panic.

	// In parallel, setup the channels to send SQL data chunks to for each destination tablet:
	insertChannels := make([]chan string, len(scw.destinationShards))
	destinationWaitGroup := sync.WaitGroup{}
	for shardIndex, si := range scw.destinationShards {
		// We create one channel per destination tablet. It is sized to have a
		// buffer of a maximum of destinationWriterCount * 2 items, to hopefully
		// always have data. We then have destinationWriterCount go routines reading
		// from it.
		insertChannels[shardIndex] = make(chan string, scw.destinationWriterCount*2)

		for j := 0; j < scw.destinationWriterCount; j++ {
			destinationWaitGroup.Add(1)
			go scw.startExecutor(ctx, &destinationWaitGroup, si.Keyspace(), si.ShardName(), insertChannels[shardIndex], j, processError)
		}
	}

	// Now for each table, read data chunks and send them to all
	// insertChannels
	readers := sync.WaitGroup{}

	err = scw.startCloningData(ctx, state, sourceSchemaDefinition, processError, firstSourceTablet, tableStatusList, start, statsCounters, insertChannels, &readers)
	if err != nil {
		return vterrors.Wrap(err, "failed to startCloningData")
	}
	readers.Wait()

	for shardIndex := range scw.destinationShards {
		close(insertChannels[shardIndex])
	}
	destinationWaitGroup.Wait()

	return firstError
}
575385
func (body Body) Copy() Body {
	cpy := make(Body, len(body))
	for i := range body {
		cpy[i] = body[i].Copy()
	}
	return cpy
}
412356
func (s *Schema) Clone() *Schema {
	cols := make([]*Column, 0, s.Len())
	keys := make([]KeyInfo, 0, len(s.Keys))
	for _, col := range s.Columns {
		cols = append(cols, col.Clone().(*Column))
	}
	for _, key := range s.Keys {
		keys = append(keys, key.Clone())
	}
	schema := NewSchema(cols...)
	schema.SetUniqueKeys(keys)
	for id, cols := range s.TblID2Handle {
		schema.TblID2Handle[id] = make([]*Column, 0, len(cols))
		for _, col := range cols {
			var inColumns = false
			for i, colInColumns := range s.Columns {
				if col == colInColumns {
					schema.TblID2Handle[id] = append(schema.TblID2Handle[id], schema.Columns[i])
					inColumns = true
					break
				}
			}
			if !inColumns {
				schema.TblID2Handle[id] = append(schema.TblID2Handle[id], col.Clone().(*Column))
			}
		}
	}
	return schema
}
1198143
func (q *Query) Copy() *Query {
	var modelValues map[string]*queryParamsAppender
	if len(q.modelValues) > 0 {
		modelValues = make(map[string]*queryParamsAppender, len(q.modelValues))
		for k, v := range q.modelValues {
			modelValues[k] = v
		}
	}

	copy := &Query{
		ctx:       q.ctx,
		db:        q.db,
		stickyErr: q.stickyErr,

		model:         q.model,
		implicitModel: q.implicitModel,
		deleted:       q.deleted,

		with:        q.with[:len(q.with):len(q.with)],
		tables:      q.tables[:len(q.tables):len(q.tables)],
		columns:     q.columns[:len(q.columns):len(q.columns)],
		set:         q.set[:len(q.set):len(q.set)],
		modelValues: modelValues,
		where:       q.where[:len(q.where):len(q.where)],
		updWhere:    q.updWhere[:len(q.updWhere):len(q.updWhere)],
		joins:       q.joins[:len(q.joins):len(q.joins)],
		group:       q.group[:len(q.group):len(q.group)],
		having:      q.having[:len(q.having):len(q.having)],
		order:       q.order[:len(q.order):len(q.order)],
		onConflict:  q.onConflict,
		returning:   q.returning[:len(q.returning):len(q.returning)],
		limit:       q.limit,
		offset:      q.offset,
		selFor:      q.selFor,
	}

	return copy
}
228132
func Copy(s *Snapshot) *Snapshot {
	cs := *(s.ChangeSet)

	return &Snapshot{
		ChangeSet: &cs,
		Version:   s.Version,
	}
}
283462
func (p *PublishPacket) Copy() *PublishPacket {
	newP := NewControlPacket(Publish).(*PublishPacket)
	newP.TopicName = p.TopicName
	newP.Payload = p.Payload

	return newP
}
833038
func (h *Header) Copy() *Header {
	return &Header{h.Version, h.Type, h.Length, h.Transaction}
}
639666
func (r *Store) Copy(ctx context.Context, from, to string) error {
	return r.move(ctx, from, to, false)
}
808729
func (n *Node) Copy(c *Node) {
	n.State = c.State
	n.Priority = c.Priority
	n.Host.Copy(&c.Host)
	n.VserversEnabled = c.VserversEnabled
}
369025
func (c *CheckRestart) Copy() *CheckRestart {
	if c == nil {
		return nil
	}

	nc := new(CheckRestart)
	nc.Limit = c.Limit
	if c.Grace != nil {
		g := *c.Grace
		nc.Grace = &g
	}
	nc.IgnoreWarnings = c.IgnoreWarnings
	return nc
}
941405
func (qrs *Rules) Copy() (newqrs *Rules) {
	newqrs = New()
	if qrs.rules != nil {
		newqrs.rules = make([]*Rule, 0, len(qrs.rules))
		for _, qr := range qrs.rules {
			newqrs.rules = append(newqrs.rules, qr.Copy())
		}
	}
	return newqrs
}
785376
func (b *Builder) copySchemaTables(dbName string) *model.DBInfo {
	oldSchemaTables := b.is.schemaMap[dbName]
	newSchemaTables := &schemaTables{
		dbInfo: oldSchemaTables.dbInfo.Copy(),
		tables: make(map[string]table.Table, len(oldSchemaTables.tables)),
	}
	for k, v := range oldSchemaTables.tables {
		newSchemaTables.tables[k] = v
	}
	b.is.schemaMap[dbName] = newSchemaTables
	return newSchemaTables.dbInfo
}
326521
func (d *EphemeralDisk) Copy() *EphemeralDisk {
	ld := new(EphemeralDisk)
	*ld = *d
	return ld
}
596784
func (td *TrianglesData) Copy() Triangles {
	copyTd := MakeTrianglesData(td.Len())
	copyTd.Update(td)
	return copyTd
}
806437
func (vs *ValueMap) Copy() *ValueMap {
	if vs == nil {
		return nil
	}
	cpy := NewValueMap()
	cpy.hashMap = vs.hashMap.Copy()
	return cpy
}
177494
func (d *DriverNetwork) Copy() *DriverNetwork {
	if d == nil {
		return nil
	}
	pm := make(map[string]int, len(d.PortMap))
	for k, v := range d.PortMap {
		pm[k] = v
	}
	return &DriverNetwork{
		PortMap:       pm,
		IP:            d.IP,
		AutoAdvertise: d.AutoAdvertise,
	}
}
724887
func (dbcfgs *DBConfigs) Copy() *DBConfigs {
	result := &DBConfigs{userConfigs: make(map[string]*userConfig)}
	for k, u := range dbcfgs.userConfigs {
		newu := *u
		result.userConfigs[k] = &newu
	}
	result.DBName.Set(dbcfgs.DBName.Get())
	result.SidecarDBName.Set(dbcfgs.SidecarDBName.Get())
	return result
}