// TestStoreRangeSplitInsideRow verifies an attempt to split a range inside of
// a table row will cause a split at a boundary between rows.
func TestStoreRangeSplitInsideRow(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer config.TestingDisableTableSplits()()
	store, stopper, _ := createTestStore(t)
	defer stopper.Stop()

	// Manually create some the column keys corresponding to the table:
	//
	//   CREATE TABLE t (id STRING PRIMARY KEY, col1 INT, col2 INT)
	tableKey := keys.MakeTablePrefix(keys.MaxReservedDescID + 1)
	rowKey := roachpb.Key(encoding.EncodeVarintAscending(append([]byte(nil), tableKey...), 1))
	rowKey = encoding.EncodeStringAscending(encoding.EncodeVarintAscending(rowKey, 1), "a")
	col1Key := keys.MakeColumnKey(append([]byte(nil), rowKey...), 1)
	col2Key := keys.MakeColumnKey(append([]byte(nil), rowKey...), 2)

	// We don't care about the value, so just store any old thing.
	if pErr := store.DB().Put(col1Key, "column 1"); pErr != nil {
		t.Fatal(pErr)
	}
	if pErr := store.DB().Put(col2Key, "column 2"); pErr != nil {
		t.Fatal(pErr)
	}

	// Split between col1Key and col2Key by splitting before col2Key.
	args := adminSplitArgs(col2Key, col2Key)
	_, pErr := client.SendWrapped(rg1(store), nil, &args)
	if pErr != nil {
		t.Fatalf("%s: split unexpected error: %s", col1Key, pErr)
	}

	rng1 := store.LookupReplica(col1Key, nil)
	rng2 := store.LookupReplica(col2Key, nil)
	// Verify the two columns are still on the same range.
	if !reflect.DeepEqual(rng1, rng2) {
		t.Fatalf("%s: ranges differ: %+v vs %+v", roachpb.Key(col1Key), rng1, rng2)
	}
	// Verify we split on a row key.
	if startKey := rng1.Desc().StartKey; !startKey.Equal(rowKey) {
		t.Fatalf("%s: expected split on %s, but found %s",
			roachpb.Key(col1Key), roachpb.Key(rowKey), startKey)
	}

	// Verify the previous range was split on a row key.
	rng3 := store.LookupReplica(tableKey, nil)
	if endKey := rng3.Desc().EndKey; !endKey.Equal(rowKey) {
		t.Fatalf("%s: expected split on %s, but found %s",
			roachpb.Key(col1Key), roachpb.Key(rowKey), endKey)
	}
}

// MakeNameMetadataKey returns the key for the name. Pass name == "" in order
// to generate the prefix key to use to scan over all of the names for the
// specified parentID.
func MakeNameMetadataKey(parentID ID, name string) roachpb.Key {
	name = NormalizeName(name)
	k := keys.MakeTablePrefix(uint32(namespaceTable.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(namespaceTable.PrimaryIndex.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(parentID))
	if name != "" {
		k = encoding.EncodeBytesAscending(k, []byte(name))
		k = keys.MakeColumnKey(k, uint32(namespaceTable.Columns[2].ID))
	}
	return k
}

// MakeDescMetadataKey returns the key for the descriptor.
func MakeDescMetadataKey(descID ID) roachpb.Key {
	k := keys.MakeTablePrefix(uint32(descriptorTable.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(descriptorTable.PrimaryIndex.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(descID))
	return keys.MakeColumnKey(k, uint32(descriptorTable.Columns[1].ID))
}

// MakeZoneKey returns the key for 'id's entry in the system.zones table.
func MakeZoneKey(id ID) roachpb.Key {
	k := keys.MakeTablePrefix(uint32(zonesTable.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(zonesTable.PrimaryIndex.ID))
	k = encoding.EncodeUvarintAscending(k, uint64(id))
	return keys.MakeColumnKey(k, uint32(zonesTable.Columns[1].ID))
}

// insertRow adds to the batch the kv operations necessary to insert a table row
// with the given values.
func (ri *rowInserter) insertRow(b *client.Batch, values []parser.Datum) error {
	if len(values) != len(ri.insertCols) {
		return util.Errorf("got %d values but expected %d", len(values), len(ri.insertCols))
	}

	// Encode the values to the expected column type. This needs to
	// happen before index encoding because certain datum types (i.e. tuple)
	// cannot be used as index values.
	for i, val := range values {
		// Make sure the value can be written to the column before proceeding.
		var err error
		if ri.marshalled[i], err = sqlbase.MarshalColumnValue(ri.insertCols[i], val); err != nil {
			return err
		}
	}

	primaryIndexKey, secondaryIndexEntries, err := ri.helper.encodeIndexes(ri.insertColIDtoRowIndex, values)
	if err != nil {
		return err
	}

	// Write the row sentinel. We want to write the sentinel first in case
	// we are trying to insert a duplicate primary key: if we write the
	// secondary indexes first, we may get an error that looks like a
	// uniqueness violation on a non-unique index.
	ri.key = keys.MakeNonColumnKey(primaryIndexKey)
	if log.V(2) {
		log.Infof("CPut %s -> NULL", ri.key)
	}
	// Each sentinel value needs a distinct RawBytes field as the computed
	// checksum includes the key the value is associated with.
	ri.sentinelValue.SetBytes([]byte{})
	b.CPut(&ri.key, &ri.sentinelValue, nil)
	ri.key = nil

	for _, secondaryIndexEntry := range secondaryIndexEntries {
		if log.V(2) {
			log.Infof("CPut %s -> %v", secondaryIndexEntry.Key, secondaryIndexEntry.Value)
		}
		ri.key = secondaryIndexEntry.Key
		b.CPut(&ri.key, secondaryIndexEntry.Value, nil)
	}
	ri.key = nil

	// Write the row columns.
	for i, val := range values {
		col := ri.insertCols[i]

		if ri.helper.columnInPK(col.ID) {
			// Skip primary key columns as their values are encoded in the row
			// sentinel key which is guaranteed to exist for as long as the row
			// exists.
			continue
		}

		if ri.marshalled[i].RawBytes != nil {
			// We only output non-NULL values. Non-existent column keys are
			// considered NULL during scanning and the row sentinel ensures we know
			// the row exists.

			ri.key = keys.MakeColumnKey(primaryIndexKey, uint32(col.ID))
			if log.V(2) {
				log.Infof("CPut %s -> %v", ri.key, val)
			}

			b.CPut(&ri.key, &ri.marshalled[i], nil)
			ri.key = nil
		}
	}

	return nil
}

// updateRow adds to the batch the kv operations necessary to update a table row
// with the given values.
//
// The row corresponding to oldValues is updated with the ones in updateValues.
// Note that updateValues only contains the ones that are changing.
//
// The return value is only good until the next call to UpdateRow.
func (ru *rowUpdater) updateRow(
	b *client.Batch,
	oldValues []parser.Datum,
	updateValues []parser.Datum,
) ([]parser.Datum, error) {
	if len(oldValues) != len(ru.fetchCols) {
		return nil, util.Errorf("got %d values but expected %d", len(oldValues), len(ru.fetchCols))
	}
	if len(updateValues) != len(ru.updateCols) {
		return nil, util.Errorf("got %d values but expected %d", len(updateValues), len(ru.updateCols))
	}

	primaryIndexKey, secondaryIndexEntries, err := ru.helper.encodeIndexes(ru.fetchColIDtoRowIndex, oldValues)
	if err != nil {
		return nil, err
	}

	// Check that the new value types match the column types. This needs to
	// happen before index encoding because certain datum types (i.e. tuple)
	// cannot be used as index values.
	for i, val := range updateValues {
		if ru.marshalled[i], err = sqlbase.MarshalColumnValue(ru.updateCols[i], val); err != nil {
			return nil, err
		}
	}

	// Update the row values.
	copy(ru.newValues, oldValues)
	for i, updateCol := range ru.updateCols {
		ru.newValues[ru.fetchColIDtoRowIndex[updateCol.ID]] = updateValues[i]
	}

	newPrimaryIndexKey := primaryIndexKey
	rowPrimaryKeyChanged := false
	var newSecondaryIndexEntries []sqlbase.IndexEntry
	if ru.primaryKeyColChange {
		newPrimaryIndexKey, newSecondaryIndexEntries, err = ru.helper.encodeIndexes(ru.fetchColIDtoRowIndex, ru.newValues)
		if err != nil {
			return nil, err
		}
		rowPrimaryKeyChanged = !bytes.Equal(primaryIndexKey, newPrimaryIndexKey)
	} else {
		newSecondaryIndexEntries, err = sqlbase.EncodeSecondaryIndexes(
			ru.helper.tableDesc.ID, ru.helper.indexes, ru.fetchColIDtoRowIndex, ru.newValues)
		if err != nil {
			return nil, err
		}
	}

	if rowPrimaryKeyChanged {
		err := ru.rd.deleteRow(b, oldValues)
		if err != nil {
			return nil, err
		}
		err = ru.ri.insertRow(b, ru.newValues)
		return ru.newValues, err
	}

	// Update secondary indexes.
	for i, newSecondaryIndexEntry := range newSecondaryIndexEntries {
		secondaryIndexEntry := secondaryIndexEntries[i]
		secondaryKeyChanged := !bytes.Equal(newSecondaryIndexEntry.Key, secondaryIndexEntry.Key)
		if secondaryKeyChanged {
			if log.V(2) {
				log.Infof("Del %s", secondaryIndexEntry.Key)
			}
			b.Del(secondaryIndexEntry.Key)
			// Do not update Indexes in the DELETE_ONLY state.
			if _, ok := ru.deleteOnlyIndex[i]; !ok {
				if log.V(2) {
					log.Infof("CPut %s -> %v", newSecondaryIndexEntry.Key, newSecondaryIndexEntry.Value)
				}
				b.CPut(newSecondaryIndexEntry.Key, newSecondaryIndexEntry.Value, nil)
			}
		}
	}

	// Add the new values.
	for i, val := range updateValues {
		col := ru.updateCols[i]

		if ru.helper.columnInPK(col.ID) {
			// Skip primary key columns as their values are encoded in the row
			// sentinel key which is guaranteed to exist for as long as the row
			// exists.
			continue
		}

		ru.key = keys.MakeColumnKey(newPrimaryIndexKey, uint32(col.ID))
		if ru.marshalled[i].RawBytes != nil {
			// We only output non-NULL values. Non-existent column keys are
			// considered NULL during scanning and the row sentinel ensures we know
			// the row exists.
//.........这里部分代码省略.........

func (sc *SchemaChanger) truncateAndBackfillColumnsChunk(
	added []sqlbase.ColumnDescriptor,
	dropped []sqlbase.ColumnDescriptor,
	nonNullableColumn string,
	defaultExprs []parser.TypedExpr,
	evalCtx parser.EvalContext,
	sp sqlbase.Span,
) (roachpb.Key, bool, error) {
	var curSentinel roachpb.Key
	done := false
	err := sc.db.Txn(func(txn *client.Txn) error {
		tableDesc, err := getTableDescFromID(txn, sc.tableID)
		if err != nil {
			return err
		}
		// Short circuit the backfill if the table has been deleted.
		if tableDesc.Deleted {
			done = true
			return nil
		}

		// Run a scan across the table using the primary key. Running
		// the scan and applying the changes in many transactions is
		// fine because the schema change is in the correct state to
		// handle intermediate OLTP commands which delete and add
		// values during the scan.
		b := &client.Batch{}
		b.Scan(sp.Start, sp.End, ColumnTruncateAndBackfillChunkSize)
		if err := txn.Run(b); err != nil {
			return err
		}

		// Use a different batch to truncate/backfill columns.
		writeBatch := &client.Batch{}
		marshalled := make([]roachpb.Value, len(defaultExprs))
		done = true
		for _, result := range b.Results {
			var sentinelKey roachpb.Key
			for _, kv := range result.Rows {
				// Still processing table.
				done = false
				if nonNullableColumn != "" {
					return fmt.Errorf("column %s contains null values", nonNullableColumn)
				}

				if sentinelKey == nil || !bytes.HasPrefix(kv.Key, sentinelKey) {
					// Sentinel keys have a 0 suffix indicating 0 bytes of
					// column ID. Strip off that suffix to determine the
					// prefix shared with the other keys for the row.
					sentinelKey = sqlbase.StripColumnIDLength(kv.Key)
					// Store away key for the next table row as the point from
					// which to start from.
					curSentinel = sentinelKey

					// Delete the entire dropped columns. This used to use SQL
					// UPDATE in the past to update the dropped column to
					// NULL; but a column in the process of being dropped is
					// placed in the table descriptor mutations, and a SQL
					// UPDATE of a column in mutations will fail.
					for _, columnDesc := range dropped {
						// Delete the dropped column.
						colKey := keys.MakeColumnKey(sentinelKey, uint32(columnDesc.ID))
						if log.V(2) {
							log.Infof("Del %s", colKey)
						}
						writeBatch.Del(colKey)
					}

					// Add the new columns and backfill the values.
					for i, expr := range defaultExprs {
						if expr == nil {
							continue
						}
						col := added[i]
						colKey := keys.MakeColumnKey(sentinelKey, uint32(col.ID))
						d, err := expr.Eval(evalCtx)
						if err != nil {
							return err
						}
						marshalled[i], err = sqlbase.MarshalColumnValue(col, d)
						if err != nil {
							return err
						}

						if log.V(2) {
							log.Infof("Put %s -> %v", colKey, d)
						}
						// Insert default value into the column. If this row
						// was recently added the default value might have
						// already been populated, because the
						// ColumnDescriptor is in the WRITE_ONLY state.
						// Reinserting the default value is not a big deal.
						//
						// Note: a column in the WRITE_ONLY state cannot be
						// populated directly through SQL. A SQL INSERT cannot
						// directly reference the column, and the INSERT
						// populates the column with the default value.
						writeBatch.Put(colKey, &marshalled[i])
					}
				}
//.........这里部分代码省略.........

func (p *planner) backfillBatch(b *client.Batch, oldTableDesc *TableDescriptor, mutationID MutationID) *roachpb.Error {
	var droppedColumnDescs []ColumnDescriptor
	var droppedIndexDescs []IndexDescriptor
	var newIndexDescs []IndexDescriptor
	// Collect the elements that are part of the mutation.
	for _, m := range oldTableDesc.Mutations {
		if m.MutationID != mutationID {
			// Mutations are applied in a FIFO order. Only apply the first set of
			// mutations if they have the mutation ID we're looking for.
			break
		}
		switch m.Direction {
		case DescriptorMutation_ADD:
			switch t := m.Descriptor_.(type) {
			case *DescriptorMutation_Column:
				// TODO(vivek): Add column to new columns and use it
				// to fill in default values.

			case *DescriptorMutation_Index:
				newIndexDescs = append(newIndexDescs, *t.Index)
			}

		case DescriptorMutation_DROP:
			switch t := m.Descriptor_.(type) {
			case *DescriptorMutation_Column:
				droppedColumnDescs = append(droppedColumnDescs, *t.Column)

			case *DescriptorMutation_Index:
				droppedIndexDescs = append(droppedIndexDescs, *t.Index)
			}
		}
	}

	// TODO(vivek): Break these backfill operations into chunks. All of them
	// will fail on big tables (see #3274).

	// Delete the entire dropped columns.
	// This used to use SQL UPDATE in the past to update the dropped
	// column to NULL; but a column in the process of being
	// dropped is placed in the table descriptor mutations, and
	// a SQL UPDATE of a column in mutations will fail.
	if len(droppedColumnDescs) > 0 {
		// Run a scan across the table using the primary key.
		start := roachpb.Key(MakeIndexKeyPrefix(oldTableDesc.ID, oldTableDesc.PrimaryIndex.ID))
		// Use a different batch to perform the scan.
		batch := &client.Batch{}
		batch.Scan(start, start.PrefixEnd(), 0)
		if pErr := p.txn.Run(batch); pErr != nil {
			return pErr
		}
		for _, result := range batch.Results {
			var sentinelKey roachpb.Key
			for _, kv := range result.Rows {
				if sentinelKey == nil || !bytes.HasPrefix(kv.Key, sentinelKey) {
					// Sentinel keys have a 0 suffix indicating 0 bytes of column
					// ID. Strip off that suffix to determine the prefix shared with the
					// other keys for the row.
					sentinelKey = stripColumnIDLength(kv.Key)
					for _, columnDesc := range droppedColumnDescs {
						// Delete the dropped column.
						colKey := keys.MakeColumnKey(sentinelKey, uint32(columnDesc.ID))
						if log.V(2) {
							log.Infof("Del %s", colKey)
						}
						b.Del(colKey)
					}
				}
			}
		}
	}

	for _, indexDescriptor := range droppedIndexDescs {
		indexPrefix := MakeIndexKeyPrefix(oldTableDesc.ID, indexDescriptor.ID)

		// Delete the index.
		indexStartKey := roachpb.Key(indexPrefix)
		indexEndKey := indexStartKey.PrefixEnd()
		if log.V(2) {
			log.Infof("DelRange %s - %s", indexStartKey, indexEndKey)
		}
		b.DelRange(indexStartKey, indexEndKey)
	}

	if len(newIndexDescs) > 0 {
		// Get all the rows affected.
		// TODO(vivek): Avoid going through Select.
		// TODO(tamird): Support partial indexes?
		// Use a scanNode with SELECT to pass in a TableDescriptor
		// to the SELECT without needing to use a parser.QualifiedName,
		// because we want to run schema changes from a gossip feed of
		// table IDs.
		scan := &scanNode{
			planner: p,
			txn:     p.txn,
			desc:    oldTableDesc,
		}
		scan.initDescDefaults()
		rows, pErr := p.selectIndex(&selectNode{}, scan, nil, false)
		if pErr != nil {
			return pErr
//.........这里部分代码省略.........

func (p *planner) backfillBatch(b *client.Batch, oldTableDesc, newTableDesc *TableDescriptor) error {
	var droppedColumnDescs []ColumnDescriptor
	var droppedIndexDescs []IndexDescriptor
	var newIndexDescs []IndexDescriptor
	for _, m := range oldTableDesc.Mutations {
		switch m.Direction {
		case DescriptorMutation_ADD:
			switch t := m.Descriptor_.(type) {
			case *DescriptorMutation_Column:
				// TODO(vivek): Add column to new columns and use it
				// to fill in default values.

			case *DescriptorMutation_Index:
				newIndexDescs = append(newIndexDescs, *t.Index)
			}

		case DescriptorMutation_DROP:
			switch t := m.Descriptor_.(type) {
			case *DescriptorMutation_Column:
				droppedColumnDescs = append(droppedColumnDescs, *t.Column)

			case *DescriptorMutation_Index:
				droppedIndexDescs = append(droppedIndexDescs, *t.Index)
			}
		}
	}

	// TODO(vivek): Break these backfill operations into chunks. All of them
	// will fail on big tables.

	// Delete the entire dropped columns.
	// This used to use SQL UPDATE in the past to update the dropped
	// column to NULL; but a column in the process of being
	// dropped is placed in the table descriptor mutations, and
	// a SQL UPDATE of a column in mutations will fail.
	if len(droppedColumnDescs) > 0 {
		// Run a scan across the table using the primary key.
		start := roachpb.Key(MakeIndexKeyPrefix(newTableDesc.ID, newTableDesc.PrimaryIndex.ID))
		// Use a different batch to perform the scan.
		batch := &client.Batch{}
		batch.Scan(start, start.PrefixEnd(), 0)
		if err := p.txn.Run(batch); err != nil {
			return err
		}
		for _, result := range batch.Results {
			var sentinelKey roachpb.Key
			for _, kv := range result.Rows {
				if sentinelKey == nil || !bytes.HasPrefix(kv.Key, sentinelKey) {
					// Sentinel keys have a 0 suffix indicating 0 bytes of column
					// ID. Strip off that suffix to determine the prefix shared with the
					// other keys for the row.
					sentinelKey = stripColumnIDLength(kv.Key)
					for _, columnDesc := range droppedColumnDescs {
						// Delete the dropped column.
						colKey := keys.MakeColumnKey(sentinelKey, uint32(columnDesc.ID))
						if log.V(2) {
							log.Infof("Del %s", colKey)
						}
						b.Del(colKey)
					}
				}
			}
		}
	}

	for _, indexDescriptor := range droppedIndexDescs {
		indexPrefix := MakeIndexKeyPrefix(newTableDesc.ID, indexDescriptor.ID)

		// Delete the index.
		indexStartKey := roachpb.Key(indexPrefix)
		indexEndKey := indexStartKey.PrefixEnd()
		if log.V(2) {
			log.Infof("DelRange %s - %s", indexStartKey, indexEndKey)
		}
		b.DelRange(indexStartKey, indexEndKey)
	}

	if len(newIndexDescs) > 0 {
		// Get all the rows affected.
		// TODO(vivek): Avoid going through Select.
		// TODO(tamird): Support partial indexes?
		// Use a scanNode with SELECT to pass in a TableDescriptor
		// to the SELECT without needing to use a parser.QualifiedName,
		// because we want to run schema changes from a gossip feed of
		// table IDs.
		scan := &scanNode{
			planner: p,
			txn:     p.txn,
			desc:    oldTableDesc,
		}
		scan.initDescDefaults()
		rows, err := p.selectWithScan(scan, &parser.Select{Exprs: oldTableDesc.allColumnsSelector()})
		if err != nil {
			return err
		}

		// Construct a map from column ID to the index the value appears at within a
		// row.
		colIDtoRowIndex, err := makeColIDtoRowIndex(rows, oldTableDesc)
		if err != nil {
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		}

		if p.prepareOnly {
			continue
		}

		primaryIndexKey, _, eErr := encodeIndexKey(
			&primaryIndex, colIDtoRowIndex, rowVals, primaryIndexKeyPrefix)
		if eErr != nil {
			return nil, roachpb.NewError(eErr)
		}

		// Write the secondary indexes.
		indexes := tableDesc.Indexes
		// Also include the secondary indexes in mutation state WRITE_ONLY.
		for _, m := range tableDesc.Mutations {
			if m.State == DescriptorMutation_WRITE_ONLY {
				if index := m.GetIndex(); index != nil {
					indexes = append(indexes, *index)
				}
			}
		}
		secondaryIndexEntries, eErr := encodeSecondaryIndexes(
			tableDesc.ID, indexes, colIDtoRowIndex, rowVals)
		if eErr != nil {
			return nil, roachpb.NewError(eErr)
		}

		for _, secondaryIndexEntry := range secondaryIndexEntries {
			if log.V(2) {
				log.Infof("CPut %s -> %v", secondaryIndexEntry.key,
					secondaryIndexEntry.value)
			}
			b.CPut(secondaryIndexEntry.key, secondaryIndexEntry.value, nil)
		}

		// Write the row sentinel.
		sentinelKey := keys.MakeNonColumnKey(primaryIndexKey)
		if log.V(2) {
			log.Infof("CPut %s -> NULL", roachpb.Key(sentinelKey))
		}
		// This is subtle: An interface{}(nil) deletes the value, so we pass in
		// []byte{} as a non-nil value.
		b.CPut(sentinelKey, []byte{}, nil)

		// Write the row columns.
		for i, val := range rowVals {
			col := cols[i]
			if _, ok := primaryKeyCols[col.ID]; ok {
				// Skip primary key columns as their values are encoded in the row
				// sentinel key which is guaranteed to exist for as long as the row
				// exists.
				continue
			}

			if marshalled[i] != nil {
				// We only output non-NULL values. Non-existent column keys are
				// considered NULL during scanning and the row sentinel ensures we know
				// the row exists.

				key := keys.MakeColumnKey(primaryIndexKey, uint32(col.ID))
				if log.V(2) {
					log.Infof("CPut %s -> %v", roachpb.Key(key), val)
				}

				b.CPut(key, marshalled[i], nil)
			}
		}

		if err := rh.append(rowVals); err != nil {
			return nil, roachpb.NewError(err)
		}
	}
	if pErr := rows.PErr(); pErr != nil {
		return nil, pErr
	}

	if p.prepareOnly {
		// Return the result column types.
		return rh.getResults(), nil
	}

	if isSystemConfigID(tableDesc.GetID()) {
		// Mark transaction as operating on the system DB.
		p.txn.SetSystemConfigTrigger()
	}

	if autoCommit {
		// An auto-txn can commit the transaction with the batch. This is an
		// optimization to avoid an extra round-trip to the transaction
		// coordinator.
		pErr = p.txn.CommitInBatch(b)
	} else {
		pErr = p.txn.Run(b)
	}
	if pErr != nil {
		return nil, convertBatchError(&tableDesc, *b, pErr)
	}
	return rh.getResults(), nil
}

func (sc *SchemaChanger) truncateAndBackfillColumns(
	lease *TableDescriptor_SchemaChangeLease,
	added []ColumnDescriptor,
	dropped []ColumnDescriptor,
	version DescriptorVersion,
) *roachpb.Error {
	evalCtx := parser.EvalContext{}
	// Set the eval context timestamps.
	pTime := timeutil.Now()
	evalCtx.SetTxnTimestamp(pTime)
	evalCtx.SetStmtTimestamp(pTime)
	defaultExprs, err := makeDefaultExprs(added, &parser.Parser{}, evalCtx)
	if err != nil {
		return roachpb.NewError(err)
	}

	// Remember any new non nullable column with no default value.
	nonNullableColumn := ""
	for _, columnDesc := range added {
		if columnDesc.DefaultExpr == nil && !columnDesc.Nullable {
			nonNullableColumn = columnDesc.Name
		}
	}

	// Add or Drop a column.
	if len(dropped) > 0 || nonNullableColumn != "" || len(defaultExprs) > 0 {
		// First extend the schema change lease.
		l, pErr := sc.ExtendLease(*lease)
		if pErr != nil {
			return pErr
		}
		*lease = l

		pErr = sc.db.Txn(func(txn *client.Txn) *roachpb.Error {
			tableDesc, pErr := getTableDescAtVersion(txn, sc.tableID, version)
			if pErr != nil {
				return pErr
			}

			// Run a scan across the table using the primary key.
			start := roachpb.Key(MakeIndexKeyPrefix(tableDesc.ID, tableDesc.PrimaryIndex.ID))
			b := &client.Batch{}
			b.Scan(start, start.PrefixEnd(), 0)
			if pErr := txn.Run(b); pErr != nil {
				return pErr
			}

			if nonNullableColumn != "" {
				for _, result := range b.Results {
					if len(result.Rows) > 0 {
						return roachpb.NewErrorf("column %s contains null values", nonNullableColumn)
					}
				}
			}

			// Use a different batch to truncate/backfill columns.
			writeBatch := &client.Batch{}
			for _, result := range b.Results {
				var sentinelKey roachpb.Key
				for _, kv := range result.Rows {
					if sentinelKey == nil || !bytes.HasPrefix(kv.Key, sentinelKey) {
						// Sentinel keys have a 0 suffix indicating 0 bytes of column
						// ID. Strip off that suffix to determine the prefix shared with the
						// other keys for the row.
						sentinelKey = stripColumnIDLength(kv.Key)

						// Delete the entire dropped columns.
						// This used to use SQL UPDATE in the past to update the dropped
						// column to NULL; but a column in the process of being
						// dropped is placed in the table descriptor mutations, and
						// a SQL UPDATE of a column in mutations will fail.
						for _, columnDesc := range dropped {
							// Delete the dropped column.
							colKey := keys.MakeColumnKey(sentinelKey, uint32(columnDesc.ID))
							if log.V(2) {
								log.Infof("Del %s", colKey)
							}
							writeBatch.Del(colKey)
						}

						// Add the new columns and backfill the values.
						for i, expr := range defaultExprs {
							if expr == nil {
								continue
							}
							col := added[i]
							colKey := keys.MakeColumnKey(sentinelKey, uint32(col.ID))
							d, err := expr.Eval(evalCtx)
							if err != nil {
								return roachpb.NewError(err)
							}
							val, err := marshalColumnValue(col, d, evalCtx.Args)
							if err != nil {
								return roachpb.NewError(err)
							}

							if log.V(2) {
								log.Infof("Put %s -> %v", colKey, val)
							}
							// Insert default value into the column. If this row
//.........这里部分代码省略.........

//.........这里部分代码省略.........
	result := &valuesNode{}
	for rows.Next() {
		rowVals := rows.Values()
		result.rows = append(result.rows, parser.DTuple(nil))

		primaryIndexKey, _, err := encodeIndexKey(
			&primaryIndex, colIDtoRowIndex, rowVals, primaryIndexKeyPrefix)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		// Compute the current secondary index key:value pairs for this row.
		secondaryIndexEntries, err := encodeSecondaryIndexes(
			tableDesc.ID, indexes, colIDtoRowIndex, rowVals)
		if err != nil {
			return nil, roachpb.NewError(err)
		}

		// Our updated value expressions occur immediately after the plain
		// columns in the output.
		newVals := rowVals[len(tableDesc.Columns):]
		// Update the row values.
		for i, col := range cols {
			val := newVals[i]
			if !col.Nullable && val == parser.DNull {
				return nil, roachpb.NewUErrorf("null value in column %q violates not-null constraint", col.Name)
			}
			rowVals[colIDtoRowIndex[col.ID]] = val
		}

		// Check that the new value types match the column types. This needs to
		// happen before index encoding because certain datum types (i.e. tuple)
		// cannot be used as index values.
		for i, val := range newVals {
			var mErr error
			if marshalled[i], mErr = marshalColumnValue(cols[i], val, p.evalCtx.Args); mErr != nil {
				return nil, roachpb.NewError(mErr)
			}
		}

		// Compute the new secondary index key:value pairs for this row.
		newSecondaryIndexEntries, eErr := encodeSecondaryIndexes(
			tableDesc.ID, indexes, colIDtoRowIndex, rowVals)
		if eErr != nil {
			return nil, roachpb.NewError(eErr)
		}

		// Update secondary indexes.
		for i, newSecondaryIndexEntry := range newSecondaryIndexEntries {
			secondaryIndexEntry := secondaryIndexEntries[i]
			if !bytes.Equal(newSecondaryIndexEntry.key, secondaryIndexEntry.key) {
				// Do not update Indexes in the DELETE_ONLY state.
				if _, ok := deleteOnlyIndex[i]; !ok {
					if log.V(2) {
						log.Infof("CPut %s -> %v", newSecondaryIndexEntry.key,
							newSecondaryIndexEntry.value)
					}
					b.CPut(newSecondaryIndexEntry.key, newSecondaryIndexEntry.value, nil)
				}
				if log.V(2) {
					log.Infof("Del %s", secondaryIndexEntry.key)
				}
				b.Del(secondaryIndexEntry.key)
			}
		}

		// Add the new values.
		for i, val := range newVals {
			col := cols[i]

			key := keys.MakeColumnKey(primaryIndexKey, uint32(col.ID))
			if marshalled[i] != nil {
				// We only output non-NULL values. Non-existent column keys are
				// considered NULL during scanning and the row sentinel ensures we know
				// the row exists.
				if log.V(2) {
					log.Infof("Put %s -> %v", key, val)
				}

				b.Put(key, marshalled[i])
			} else {
				// The column might have already existed but is being set to NULL, so
				// delete it.
				if log.V(2) {
					log.Infof("Del %s", key)
				}

				b.Del(key)
			}
		}
	}

	if pErr := rows.PErr(); pErr != nil {
		return nil, pErr
	}
	if pErr := p.txn.Run(&b); pErr != nil {
		return nil, convertBatchError(tableDesc, b, pErr)
	}

	return result, nil
}

//.........这里部分代码省略.........
			if log.V(2) {
				log.Infof("DelRange %s - %s", rowStartKey, rowEndKey)
			}
			b.DelRange(rowStartKey, rowEndKey, false)

			// Delete all the old secondary indexes.
			for _, secondaryIndexEntry := range secondaryIndexEntries {
				if log.V(2) {
					log.Infof("Del %s", secondaryIndexEntry.key)
				}
				b.Del(secondaryIndexEntry.key)
			}

			// Write the new row sentinel. We want to write the sentinel first in case
			// we are trying to insert a duplicate primary key: if we write the
			// secondary indexes first, we may get an error that looks like a
			// uniqueness violation on a non-unique index.
			sentinelKey := keys.MakeNonColumnKey(newPrimaryIndexKey)
			if log.V(2) {
				log.Infof("CPut %s -> NULL", roachpb.Key(sentinelKey))
			}
			// This is subtle: An interface{}(nil) deletes the value, so we pass in
			// []byte{} as a non-nil value.
			b.CPut(sentinelKey, []byte{}, nil)

			// Write any fields from the old row that were not modified by the UPDATE.
			for i, col := range tableDesc.Columns {
				if _, ok := colIDSet[col.ID]; ok {
					continue
				}
				if _, ok := primaryKeyCols[col.ID]; ok {
					continue
				}
				key := keys.MakeColumnKey(newPrimaryIndexKey, uint32(col.ID))
				val := rowVals[i]
				marshalledVal, mErr := marshalColumnValue(col, val, p.evalCtx.Args)
				if mErr != nil {
					return nil, roachpb.NewError(mErr)
				}

				if log.V(2) {
					log.Infof("Put %s -> %v", roachpb.Key(key), val)
				}
				b.Put(key, marshalledVal)
			}
			// At this point, we've deleted the old row and associated index data and
			// written the sentinel keys and column keys for non-updated columns. Fall
			// through to below where the index keys and updated column keys will be
			// written.
		}

		// Update secondary indexes.
		for i, newSecondaryIndexEntry := range newSecondaryIndexEntries {
			secondaryIndexEntry := secondaryIndexEntries[i]
			secondaryKeyChanged := !bytes.Equal(newSecondaryIndexEntry.key, secondaryIndexEntry.key)
			if secondaryKeyChanged {
				if log.V(2) {
					log.Infof("Del %s", secondaryIndexEntry.key)
				}
				b.Del(secondaryIndexEntry.key)
			}
			if rowPrimaryKeyChanged || secondaryKeyChanged {
				// Do not update Indexes in the DELETE_ONLY state.
				if _, ok := deleteOnlyIndex[i]; !ok {
					if log.V(2) {
						log.Infof("CPut %s -> %v", newSecondaryIndexEntry.key,