// Initializes the from node, given the parsed select expression
func (s *selectNode) initFrom(p *planner, parsed *parser.Select) *roachpb.Error {
	scan := &scanNode{planner: p, txn: p.txn}

	from := parsed.From
	switch len(from) {
	case 0:
		// Nothing to do

	case 1:
		ate, ok := from[0].(*parser.AliasedTableExpr)
		if !ok {
			return roachpb.NewErrorf("TODO(pmattis): unsupported FROM: %s", from)
		}
		s.pErr = scan.initTableExpr(p, ate)
		if s.pErr != nil {
			return s.pErr
		}
	default:
		s.pErr = roachpb.NewErrorf("TODO(pmattis): unsupported FROM: %s", from)
		return s.pErr
	}

	s.pErr = scan.init(parsed)
	if s.pErr != nil {
		return s.pErr
	}
	s.from = scan
	return nil
}

// getDescriptor looks up the descriptor for `plainKey`, validates it,
// and unmarshals it into `descriptor`.
func (p *planner) getDescriptor(plainKey descriptorKey, descriptor descriptorProto) *roachpb.Error {
	gr, err := p.txn.Get(plainKey.Key())
	if err != nil {
		return err
	}
	if !gr.Exists() {
		return roachpb.NewUErrorf("%s %q does not exist", descriptor.TypeName(), plainKey.Name())
	}

	descKey := MakeDescMetadataKey(ID(gr.ValueInt()))
	desc := &Descriptor{}
	if pErr := p.txn.GetProto(descKey, desc); pErr != nil {
		return pErr
	}

	switch t := descriptor.(type) {
	case *TableDescriptor:
		table := desc.GetTable()
		if table == nil {
			return roachpb.NewErrorf("%q is not a table", plainKey.Name())
		}
		*t = *table
	case *DatabaseDescriptor:
		database := desc.GetDatabase()
		if database == nil {
			return roachpb.NewErrorf("%q is not a database", plainKey.Name())
		}
		*t = *database
	}

	return roachpb.NewError(descriptor.Validate())
}

// getDescriptorFromTargetList examines a TargetList and fetches the
// appropriate descriptor.
// TODO(marc): support multiple targets.
func (p *planner) getDescriptorFromTargetList(targets parser.TargetList) (descriptorProto, *roachpb.Error) {
	if targets.Databases != nil {
		if len(targets.Databases) == 0 {
			return nil, roachpb.NewError(errNoDatabase)
		} else if len(targets.Databases) != 1 {
			return nil, roachpb.NewErrorf("TODO(marc): multiple targets not implemented")
		}
		descriptor, err := p.getDatabaseDesc(targets.Databases[0])
		if err != nil {
			return nil, err
		}
		return descriptor, nil
	}

	if len(targets.Tables) == 0 {
		return nil, roachpb.NewError(errNoTable)
	} else if len(targets.Tables) != 1 {
		return nil, roachpb.NewErrorf("TODO(marc): multiple targets not implemented")
	}
	descriptor, err := p.getTableDesc(targets.Tables[0])
	if err != nil {
		return nil, err
	}
	return descriptor, nil
}

// expandTableGlob expands wildcards from the end of `expr` and
// returns the list of matching tables.
// `expr` is possibly modified to be qualified with the database it refers to.
// `expr` is assumed to be of one of several forms:
// 		database.table
// 		table
// 		*
func (p *planner) expandTableGlob(expr *parser.QualifiedName) (
	parser.QualifiedNames, *roachpb.Error) {
	if len(expr.Indirect) == 0 {
		return parser.QualifiedNames{expr}, nil
	}

	if err := expr.QualifyWithDatabase(p.session.Database); err != nil {
		return nil, roachpb.NewError(err)
	}
	// We must have a single indirect: either .table or .*
	if len(expr.Indirect) != 1 {
		return nil, roachpb.NewErrorf("invalid table glob: %s", expr)
	}

	switch expr.Indirect[0].(type) {
	case parser.NameIndirection:
		return parser.QualifiedNames{expr}, nil
	case parser.StarIndirection:
		dbDesc, pErr := p.getDatabaseDesc(string(expr.Base))
		if pErr != nil {
			return nil, pErr
		}
		tableNames, pErr := p.getTableNames(dbDesc)
		if pErr != nil {
			return nil, pErr
		}
		return tableNames, nil
	default:
		return nil, roachpb.NewErrorf("invalid table glob: %s", expr)
	}
}

// TestTxnDBBasics verifies that a simple transaction can be run and
// either committed or aborted. On commit, mutations are visible; on
// abort, mutations are never visible. During the txn, verify that
// uncommitted writes cannot be read outside of the txn but can be
// read from inside the txn.
func TestTxnDBBasics(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := createTestDB(t)
	defer s.Stop()
	value := []byte("value")

	for _, commit := range []bool{true, false} {
		key := []byte(fmt.Sprintf("key-%t", commit))

		pErr := s.DB.Txn(func(txn *client.Txn) *roachpb.Error {
			// Use snapshot isolation so non-transactional read can always push.
			if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
				return roachpb.NewError(err)
			}

			// Put transactional value.
			if pErr := txn.Put(key, value); pErr != nil {
				return pErr
			}

			// Attempt to read outside of txn.
			if gr, pErr := s.DB.Get(key); pErr != nil {
				return pErr
			} else if gr.Exists() {
				return roachpb.NewErrorf("expected nil value; got %v", gr.Value)
			}

			// Read within the transaction.
			if gr, pErr := txn.Get(key); pErr != nil {
				return pErr
			} else if !gr.Exists() || !bytes.Equal(gr.ValueBytes(), value) {
				return roachpb.NewErrorf("expected value %q; got %q", value, gr.Value)
			}

			if !commit {
				return roachpb.NewErrorf("purposefully failing transaction")
			}
			return nil
		})

		if commit != (pErr == nil) {
			t.Errorf("expected success? %t; got %s", commit, pErr)
		} else if !commit && !testutils.IsPError(pErr, "purposefully failing transaction") {
			t.Errorf("unexpected failure with !commit: %s", pErr)
		}

		// Verify the value is now visible on commit == true, and not visible otherwise.
		gr, pErr := s.DB.Get(key)
		if commit {
			if pErr != nil || !gr.Exists() || !bytes.Equal(gr.ValueBytes(), value) {
				t.Errorf("expected success reading value: %+v, %s", gr.ValueBytes(), pErr)
			}
		} else {
			if pErr != nil || gr.Exists() {
				t.Errorf("expected success and nil value: %s, %s", gr, pErr)
			}
		}
	}
}

// TestClientRunTransaction verifies some simple transaction isolation
// semantics.
func TestClientRunTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := server.StartTestServer(t)
	defer s.Stop()
	defer setTxnRetryBackoff(1 * time.Millisecond)()
	db := createTestClient(t, s.Stopper(), s.ServingAddr())

	for _, commit := range []bool{true, false} {
		value := []byte("value")
		key := []byte(fmt.Sprintf("%s/key-%t", testUser, commit))

		// Use snapshot isolation so non-transactional read can always push.
		pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
			if pErr := txn.SetIsolation(roachpb.SNAPSHOT); pErr != nil {
				return pErr
			}

			// Put transactional value.
			if pErr := txn.Put(key, value); pErr != nil {
				return pErr
			}
			// Attempt to read outside of txn.
			if gr, pErr := db.Get(key); pErr != nil {
				return pErr
			} else if gr.Value != nil {
				return roachpb.NewErrorf("expected nil value; got %+v", gr.Value)
			}
			// Read within the transaction.
			if gr, pErr := txn.Get(key); pErr != nil {
				return pErr
			} else if gr.Value == nil || !bytes.Equal(gr.ValueBytes(), value) {
				return roachpb.NewErrorf("expected value %q; got %q", value, gr.ValueBytes())
			}
			if !commit {
				return roachpb.NewErrorf("purposefully failing transaction")
			}
			return nil
		})

		if commit != (pErr == nil) {
			t.Errorf("expected success? %t; got %s", commit, pErr)
		} else if !commit && !testutils.IsPError(pErr, "purposefully failing transaction") {
			t.Errorf("unexpected failure with !commit: %s", pErr)
		}

		// Verify the value is now visible on commit == true, and not visible otherwise.
		gr, pErr := db.Get(key)
		if commit {
			if pErr != nil || gr.Value == nil || !bytes.Equal(gr.ValueBytes(), value) {
				t.Errorf("expected success reading value: %+v, %s", gr.Value, pErr)
			}
		} else {
			if pErr != nil || gr.Value != nil {
				t.Errorf("expected success and nil value: %+v, %s", gr.Value, pErr)
			}
		}
	}
}

// SetUserPriority sets the transaction's user priority. Transactions default to
// normal user priority. The user priority must be set before any operations are
// performed on the transaction.
func (txn *Txn) SetUserPriority(userPriority roachpb.UserPriority) *roachpb.Error {
	if txn.UserPriority != userPriority {
		if txn.Proto.IsInitialized() {
			return roachpb.NewErrorf("cannot change the user priority of a running transaction")
		}
		if userPriority < roachpb.MinUserPriority || userPriority > roachpb.MaxUserPriority {
			return roachpb.NewErrorf("the given user priority %f is out of the allowed range [%f, %f]", userPriority, roachpb.MinUserPriority, roachpb.MaxUserPriority)
		}
		txn.UserPriority = userPriority
	}
	return nil
}

func (sc *SchemaChanger) findTableWithLease(txn *client.Txn, lease TableDescriptor_SchemaChangeLease) (*TableDescriptor, *roachpb.Error) {
	tableDesc, err := getTableDescFromID(txn, sc.tableID)
	if err != nil {
		return nil, err
	}
	if tableDesc.Lease == nil {
		return nil, roachpb.NewErrorf("no lease present for tableID: %d", sc.tableID)
	}
	if *tableDesc.Lease != lease {
		return nil, roachpb.NewErrorf("table: %d has lease: %v, expected: %v", sc.tableID, tableDesc.Lease, lease)
	}
	return tableDesc, nil
}

// getAliasedTableLease looks up the table descriptor for an alias table
// expression.
func (p *planner) getAliasedTableLease(n parser.TableExpr) (*TableDescriptor, *roachpb.Error) {
	ate, ok := n.(*parser.AliasedTableExpr)
	if !ok {
		return nil, roachpb.NewErrorf("TODO(pmattis): unsupported FROM: %s", n)
	}
	table, ok := ate.Expr.(*parser.QualifiedName)
	if !ok {
		return nil, roachpb.NewErrorf("TODO(pmattis): unsupported FROM: %s", n)
	}
	desc, pErr := p.getTableLease(table)
	if pErr != nil {
		return nil, pErr
	}
	return &desc, nil
}

func makeKeyVals(desc *TableDescriptor, columnIDs []ColumnID) ([]parser.Datum, *roachpb.Error) {
	vals := make([]parser.Datum, len(columnIDs))
	for i, id := range columnIDs {
		col, pErr := desc.FindColumnByID(id)
		if pErr != nil {
			return nil, pErr
		}
		switch col.Type.Kind {
		case ColumnType_BOOL:
			vals[i] = parser.DummyBool
		case ColumnType_INT:
			vals[i] = parser.DummyInt
		case ColumnType_FLOAT:
			vals[i] = parser.DummyFloat
		case ColumnType_DECIMAL:
			vals[i] = parser.DummyDecimal
		case ColumnType_STRING:
			vals[i] = parser.DummyString
		case ColumnType_BYTES:
			vals[i] = parser.DummyBytes
		case ColumnType_DATE:
			vals[i] = parser.DummyDate
		case ColumnType_TIMESTAMP:
			vals[i] = parser.DummyTimestamp
		case ColumnType_INTERVAL:
			vals[i] = parser.DummyInterval
		default:
			return nil, roachpb.NewErrorf("TODO(pmattis): decoded index key: %s", col.Type.Kind)
		}
	}
	return vals, nil
}

func (t *tableState) release(lease *LeaseState, store LeaseStore) *roachpb.Error {
	t.mu.Lock()
	defer t.mu.Unlock()

	s := t.active.find(lease.Version, lease.expiration)
	if s == nil {
		return roachpb.NewErrorf("table %d version %d not found", lease.ID, lease.Version)
	}
	s.refcount--
	if log.V(3) {
		log.Infof("release: descID=%d version=%d refcount=%d", s.ID, s.Version, s.refcount)
	}
	if s.refcount == 0 {
		n := t.active.findNewest(0)
		if s != n {
			if s.Version < n.Version {
				// TODO(pmattis): If an active transaction is releasing the lease for
				// an older version, hold on to it for a few seconds in anticipation of
				// another operation being performed within the transaction. If we
				// release the lease immediately the transaction will necessarily abort
				// on the next operation due to not being able to get the lease.
			}
			t.active.remove(s)
			return t.releaseNodeLease(s, store)
		}
	}
	return nil
}

// insertEventRecord inserts a single event into the event log as part of the
// provided transaction.
func (ev EventLogger) insertEventRecord(txn *client.Txn, eventType EventLogType, targetID, reportingID int32, info interface{}) *roachpb.Error {
	const insertEventTableStmt = `
INSERT INTO system.eventlog (
  timestamp, eventType, targetID, reportingID, info
)
VALUES(
  $1, $2, $3, $4, $5
)
`
	args := []interface{}{
		ev.selectEventTimestamp(txn.Proto.Timestamp),
		eventType,
		targetID,
		reportingID,
		nil, // info
	}
	if info != nil {
		infoBytes, err := json.Marshal(info)
		if err != nil {
			return roachpb.NewError(err)
		}
		args[4] = string(infoBytes)
	}

	rows, err := ev.ExecuteStatementInTransaction(txn, insertEventTableStmt, args...)
	if err != nil {
		return err
	}
	if rows != 1 {
		return roachpb.NewErrorf("%d rows affected by log insertion; expected exactly one row affected.", rows)
	}
	return nil
}

// getTableID retrieves the table ID for the specified table. It uses the
// descriptor cache to perform lookups, falling back to the KV store when
// necessary.
func (p *planner) getTableID(qname *parser.QualifiedName) (ID, *roachpb.Error) {
	if err := qname.NormalizeTableName(p.session.Database); err != nil {
		return 0, roachpb.NewError(err)
	}

	dbID, pErr := p.getDatabaseID(qname.Database())
	if pErr != nil {
		return 0, pErr
	}

	// Lookup the ID of the table in the cache. The use of the cache might cause
	// the usage of a recently renamed table, but that's a race that could occur
	// anyways.
	nameKey := tableKey{dbID, qname.Table()}
	key := nameKey.Key()
	if nameVal := p.systemConfig.GetValue(key); nameVal != nil {
		id, err := nameVal.GetInt()
		return ID(id), roachpb.NewError(err)
	}

	gr, pErr := p.txn.Get(key)
	if pErr != nil {
		return 0, pErr
	}
	if !gr.Exists() {
		return 0, roachpb.NewErrorf("table %q does not exist", nameKey.Name())
	}
	return ID(gr.ValueInt()), nil
}

// waitForOneVersion returns once there are no unexpired leases on the
// previous version of the table descriptor. It returns the current version.
// After returning there can only be versions of the descriptor >= to the
// returned verson. Lease acquisition (see acquire()) maintains the
// invariant that no new leases for desc.Version-1 will be granted once
// desc.Version exists.
func (s LeaseStore) waitForOneVersion(tableID ID, retryOpts retry.Options) (DescriptorVersion, *roachpb.Error) {
	desc := &Descriptor{}
	descKey := MakeDescMetadataKey(tableID)
	var tableDesc *TableDescriptor
	for r := retry.Start(retryOpts); r.Next(); {
		// Get the current version of the table descriptor non-transactionally.
		//
		// TODO(pmattis): Do an inconsistent read here?
		if pErr := s.db.GetProto(descKey, desc); pErr != nil {
			return 0, pErr
		}
		tableDesc = desc.GetTable()
		if tableDesc == nil {
			return 0, roachpb.NewErrorf("ID %d is not a table", tableID)
		}
		// Check to see if there are any leases that still exist on the previous
		// version of the descriptor.
		now := s.clock.Now()
		count, pErr := s.countLeases(tableDesc.ID, tableDesc.Version-1, now.GoTime())
		if pErr != nil {
			return 0, pErr
		}
		if count == 0 {
			break
		}
		log.Infof("publish (count leases): descID=%d version=%d count=%d",
			tableDesc.ID, tableDesc.Version-1, count)
	}
	return tableDesc.Version, nil
}

func markDebug(plan planNode, mode explainMode) (planNode, *roachpb.Error) {
	switch t := plan.(type) {
	case *selectNode:
		return markDebug(t.from, mode)

	case *scanNode:
		// Mark the node as being explained.
		t.columns = []column{
			{name: "RowIdx", typ: parser.DummyInt},
			{name: "Key", typ: parser.DummyString},
			{name: "Value", typ: parser.DummyString},
			{name: "Output", typ: parser.DummyBool},
		}
		t.explain = mode
		return t, nil

	case *indexJoinNode:
		return markDebug(t.index, mode)

	case *sortNode:
		return markDebug(t.plan, mode)

	default:
		return nil, roachpb.NewErrorf("TODO(pmattis): unimplemented %T", plan)
	}
}

func markDebug(plan planNode, mode explainMode) (planNode, *roachpb.Error) {
	switch t := plan.(type) {
	case *selectNode:
		t.explain = mode

		if _, ok := t.from.node.(*indexJoinNode); ok {
			// We will replace the indexJoinNode with the index node; we cannot
			// process filters anymore (we don't have all the values).
			t.filter = nil
		}
		// Mark the from node as debug (and potentially replace it).
		newNode, err := markDebug(t.from.node, mode)
		t.from.node = newNode.(fromNode)
		return t, err

	case *scanNode:
		t.explain = mode
		return t, nil

	case *indexJoinNode:
		// Replace the indexJoinNode with the index node.
		return markDebug(t.index, mode)

	case *sortNode:
		// Replace the sort node with the node it wraps.
		return markDebug(t.plan, mode)

	case *groupNode:
		// Replace the group node with the node it wraps.
		return markDebug(t.plan, mode)

	default:
		return nil, roachpb.NewErrorf("TODO(pmattis): unimplemented %T", plan)
	}
}

// send runs the specified calls synchronously in a single batch and returns
// any errors. Returns a nil response for empty input (no requests).
func (db *DB) send(maxScanResults int64, readConsistency roachpb.ReadConsistencyType,
	reqs ...roachpb.Request) (*roachpb.BatchResponse, *roachpb.Error) {
	if len(reqs) == 0 {
		return nil, nil
	}

	if readConsistency == roachpb.INCONSISTENT {
		for _, req := range reqs {
			if req.Method() != roachpb.Get && req.Method() != roachpb.Scan &&
				req.Method() != roachpb.ReverseScan {
				return nil, roachpb.NewErrorf("method %s not allowed with INCONSISTENT batch", req.Method)
			}
		}
	}

	ba := roachpb.BatchRequest{}
	ba.Add(reqs...)

	ba.MaxScanResults = maxScanResults
	if db.userPriority != 1 {
		ba.UserPriority = db.userPriority
	}
	ba.ReadConsistency = readConsistency

	tracing.AnnotateTrace()

	br, pErr := db.sender.Send(context.TODO(), ba)
	if pErr != nil {
		if log.V(1) {
			log.Infof("failed batch: %s", pErr)
		}
		return nil, pErr
	}
	return br, nil
}

func TestTxnAbortCount(t *testing.T) {
	defer leaktest.AfterTest(t)()
	_, sender, cleanupFn := setupMetricsTest(t)
	defer cleanupFn()

	value := []byte("value")
	db := client.NewDB(sender)

	intentionalErrText := "intentional error to cause abort"
	// Test aborted transaction.
	if pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
		key := []byte("key-abort")

		if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
			return roachpb.NewError(err)
		}

		if pErr := txn.Put(key, value); pErr != nil {
			t.Fatal(pErr)
		}

		return roachpb.NewErrorf(intentionalErrText)
	}); !testutils.IsPError(pErr, intentionalErrText) {
		t.Fatalf("unexpected error: %s", pErr)
	}
	teardownHeartbeats(sender)
	checkTxnMetrics(t, sender, "abort txn", 0, 0, 1, 0)
}

func (s *Store) insertRangeLogEvent(txn *client.Txn, event rangeLogEvent) *roachpb.Error {
	const insertEventTableStmt = `
INSERT INTO system.rangelog (
  timestamp, rangeID, eventType, storeID, otherRangeID
)
VALUES(
  $1, $2, $3, $4, $5
)
`
	args := []interface{}{
		event.timestamp,
		event.rangeID,
		event.eventType,
		event.storeID,
		nil, //otherRangeID
	}
	if event.otherRangeID != nil {
		args[4] = *event.otherRangeID
	}

	rows, err := s.ctx.SQLExecutor.ExecuteStatementInTransaction(txn, insertEventTableStmt, args...)
	if err != nil {
		return err
	}
	if rows != 1 {
		return roachpb.NewErrorf("%d rows affected by log insertion; expected exactly one row affected.", rows)
	}
	return nil
}

// send runs the specified calls synchronously in a single batch and returns
// any errors. Returns (nil, nil) for an empty batch.
func (db *DB) send(ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
	if len(ba.Requests) == 0 {
		return nil, nil
	}
	if ba.ReadConsistency == roachpb.INCONSISTENT {
		for _, ru := range ba.Requests {
			req := ru.GetInner()
			if req.Method() != roachpb.Get && req.Method() != roachpb.Scan &&
				req.Method() != roachpb.ReverseScan {
				return nil, roachpb.NewErrorf("method %s not allowed with INCONSISTENT batch", req.Method)
			}
		}
	}

	if db.ctx.UserPriority != 1 {
		ba.UserPriority = db.ctx.UserPriority
	}

	tracing.AnnotateTrace()

	br, pErr := db.sender.Send(context.TODO(), ba)
	if pErr != nil {
		if log.V(1) {
			log.Infof("failed batch: %s", pErr)
		}
		return nil, pErr
	}
	return br, nil
}