func (m *sqlExecutionStore) ReadHistoryBranch(
	request *p.InternalReadHistoryBranchRequest,
) (*p.InternalReadHistoryBranchResponse, error) {
	ctx, cancel := newExecutionContext()
	defer cancel()
	branchIDBytes, err := primitives.ParseUUID(request.BranchID)
	if err != nil {
		return nil, err
	}
	treeIDBytes, err := primitives.ParseUUID(request.TreeID)
	if err != nil {
		return nil, err
	}
	var token historyNodePaginationToken
	if len(request.NextPageToken) == 0 {
		token = newHistoryNodePaginationToken(request.MinNodeID, MinTxnID)
	} else if len(request.NextPageToken) == 8 {
		lastNodeID, err := deserializePageToken(request.NextPageToken)
		if err != nil {
			return nil, err
		}
		token = newHistoryNodePaginationToken(lastNodeID+1, MinTxnID)
	} else {
		token, err = deserializeHistoryNodePaginationToken(request.NextPageToken)
		if err != nil {
			return nil, err
		}
	}
	rows, err := m.Db.RangeSelectFromHistoryNode(ctx, sqlplugin.HistoryNodeSelectFilter{
		ShardID:      request.ShardID,
		TreeID:       treeIDBytes,
		BranchID:     branchIDBytes,
		MinNodeID:    token.LastNodeID,
		MinTxnID:     token.LastTxnID,
		MaxNodeID:    request.MaxNodeID,
		PageSize:     request.PageSize,
		MetadataOnly: request.MetadataOnly,
	})
	switch err {
	case nil:
	case sql.ErrNoRows:
	default:
		return nil, err
	}
	nodes := make([]p.InternalHistoryNode, 0, len(rows))
	for _, row := range rows {
		nodes = append(nodes, p.InternalHistoryNode{
			NodeID:            row.NodeID,
			PrevTransactionID: row.PrevTxnID,
			TransactionID:     row.TxnID,
			Events:            p.NewDataBlob(row.Data, row.DataEncoding),
		})
	}
	var pagingToken []byte
	if len(rows) < request.PageSize {
		pagingToken = nil
	} else {
		lastRow := rows[len(rows)-1]
		pagingToken, err = serializeHistoryNodePaginationToken(
			newHistoryNodePaginationToken(lastRow.NodeID, lastRow.TxnID),
		)
		if err != nil {
			return nil, err
		}
	}
	return &p.InternalReadHistoryBranchResponse{
		Nodes:         nodes,
		NextPageToken: pagingToken,
	}, nil
}
ReadHistoryBranch returns history node data for a branch
package sql
import (
	"database/sql"
	"fmt"
	"math"
	commonpb "go.temporal.io/api/common/v1"
	"go.temporal.io/api/serviceerror"
	p "go.temporal.io/server/common/persistence"
	"go.temporal.io/server/common/persistence/sql/sqlplugin"
	"go.temporal.io/server/common/primitives"
)
const (
	MinTxnID = math.MaxInt64
)
func (m *sqlExecutionStore) AppendHistoryNodes(
	request *p.InternalAppendHistoryNodesRequest,
) error {
	ctx, cancel := newExecutionContext()
	defer cancel()
	branchInfo := request.BranchInfo
	node := request.Node
	treeIDBytes, err := primitives.ParseUUID(branchInfo.GetTreeId())
	if err != nil {
		return err
	}
	branchIDBytes, err := primitives.ParseUUID(branchInfo.GetBranchId())
	if err != nil {
		return err
	}
	nodeRow := &sqlplugin.HistoryNodeRow{
		TreeID:       treeIDBytes,
		BranchID:     branchIDBytes,
		NodeID:       node.NodeID,
		PrevTxnID:    node.PrevTransactionID,
		TxnID:        node.TransactionID,
		Data:         node.Events.Data,
		DataEncoding: node.Events.EncodingType.String(),
		ShardID:      request.ShardID,
	}
	if !request.IsNewBranch {
		_, err = m.Db.InsertIntoHistoryNode(ctx, nodeRow)
		if err != nil {
			if m.Db.IsDupEntryError(err) {
				return &p.ConditionFailedError{Msg: fmt.Sprintf("AppendHistoryNodes: row already exist: %v", err)}
			}
			return serviceerror.NewUnavailable(fmt.Sprintf("AppendHistoryNodes: %v", err))
		}
		return nil
	}
	treeInfoBlob := request.TreeInfo
	treeRow := &sqlplugin.HistoryTreeRow{
		ShardID:      request.ShardID,
		TreeID:       treeIDBytes,
		BranchID:     branchIDBytes,
		Data:         treeInfoBlob.Data,
		DataEncoding: treeInfoBlob.EncodingType.String(),
	}
	return m.txExecute(ctx, "AppendHistoryNodes", func(tx sqlplugin.Tx) error {
		result, err := tx.InsertIntoHistoryNode(ctx, nodeRow)
		if err != nil {
			return err
		}
		rowsAffected, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rowsAffected != 1 {
			return fmt.Errorf("expected 1 row to be affected for node table, got %v", rowsAffected)
		}
		result, err = tx.InsertIntoHistoryTree(ctx, treeRow)
		if err != nil {
			return err
		}
		rowsAffected, err = result.RowsAffected()
		if err != nil {
			return err
		}
		if !(rowsAffected == 1 || rowsAffected == 2) {
			return fmt.Errorf("expected 1 or 2 rows to be affected for tree table as we allow upserts, got %v", rowsAffected)
		}
		return nil
	})
}
func (m *sqlExecutionStore) DeleteHistoryNodes(
	request *p.InternalDeleteHistoryNodesRequest,
) error {
	ctx, cancel := newExecutionContext()
	defer cancel()
	branchInfo := request.BranchInfo
	nodeID := request.NodeID
	txnID := request.TransactionID
	shardID := request.ShardID
	if nodeID < p.GetBeginNodeID(branchInfo) {
		return &p.InvalidPersistenceRequestError{
			Msg: fmt.Sprintf("cannot append to ancestors' nodes"),
		}
	}
	treeIDBytes, err := primitives.ParseUUID(branchInfo.GetTreeId())
	if err != nil {
		return err
	}
	branchIDBytes, err := primitives.ParseUUID(branchInfo.GetBranchId())
	if err != nil {
		return err
	}
	nodeRow := &sqlplugin.HistoryNodeRow{
		TreeID:   treeIDBytes,
		BranchID: branchIDBytes,
		NodeID:   nodeID,
		TxnID:    txnID,
		ShardID:  shardID,
	}
	_, err = m.Db.DeleteFromHistoryNode(ctx, nodeRow)
	if err != nil {
		return serviceerror.NewUnavailable(fmt.Sprintf("DeleteHistoryNodes: %v", err))
	}
	return nil
}