func (ls *logStorage) readDupeLeaves(ctx context.Context, logID int64, dupes map[string][]int, results []*trillian.QueuedLogLeaf) error {
	numDupes := len(dupes)
	if numDupes == 0 {
		return nil
	}
	glog.V(2).Infof("dupe rowsToRead: %v", numDupes)
	ids := make([][]byte, 0, numDupes)
	for k := range dupes {
		ids = append(ids, []byte(k))
	}
	dupesRead := 0
	tx := ls.ts.client.Single()
	err := readLeaves(ctx, tx, logID, ids, func(l *trillian.LogLeaf) {
		glog.V(2).Infof("Found already exists dupe: %v", l)
		dupesRead++
		indices := dupes[string(l.LeafIdentityHash)]
		glog.V(2).Infof("Indices %v", indices)
		if len(indices) == 0 {
			glog.Warningf("Logic error: Spanner returned a leaf %x, but it matched no requested index", l.LeafIdentityHash)
			return
		}
		for _, i := range indices {
			leaf := l
			results[i] = &trillian.QueuedLogLeaf{
				Leaf:   leaf,
				Status: status.Newf(codes.AlreadyExists, "leaf already exists: %v", l.LeafIdentityHash).Proto(),
			}
		}
	})
	tx.Close()
	if err != nil {
		return err
	}
	if got, want := dupesRead, numDupes; got != want {
		return fmt.Errorf("read unexpected number of dupe rows %d, want %d", got, want)
	}
	return nil
}
readDupeLeaves reads the leaves whose ids are passed as keys in the dupes map,
and stores them in results.
package cloudspanner
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"sort"
	"sync"
	"time"
	"cloud.google.com/go/spanner"
	"github.com/golang/glog"
	"github.com/google/trillian"
	"github.com/google/trillian/merkle/rfc6962"
	"github.com/google/trillian/storage"
	"github.com/google/trillian/storage/cache"
	"github.com/google/trillian/storage/cloudspanner/spannerpb"
	"github.com/google/trillian/types"
	"go.opencensus.io/trace"
	"golang.org/x/sync/semaphore"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/timestamppb"
)
const (
	leafDataTbl            = "LeafData"
	seqDataByMerkleHashIdx = "SequenceByMerkleHash"
	seqDataTbl             = "SequencedLeafData"
	unseqTable             = "Unsequenced"
	getActiveLogIDsSQL = `SELECT t.TreeID FROM TreeRoots t
WHERE (t.TreeType = 1 OR t.TreeType = 3)
AND (t.TreeState = 1 OR t.TreeState = 5)
AND t.Deleted=false`
)
type LogStorageOptions struct {
	TreeStorageOptions
	DequeueAcrossMerkleBuckets bool
	DequeueAcrossMerkleBucketsRangeFraction float64
}
var (
	colLeafIdentityHash    = "LeafIdentityHash"
	colLeafValue           = "LeafValue"
	colExtraData           = "ExtraData"
	colMerkleLeafHash      = "MerkleLeafHash"
	colSequenceNumber      = "SequenceNumber"
	colQueueTimestampNanos = "QueueTimestampNanos"
)
type leafDataCols struct {
	TreeID              int64
	LeafIdentityHash    []byte
	LeafValue           []byte
	ExtraData           []byte
	QueueTimestampNanos int64
}
type sequencedLeafDataCols struct {
	TreeID                  int64
	SequenceNumber          int64
	LeafIdentityHash        []byte
	MerkleLeafHash          []byte
	IntegrateTimestampNanos int64
}
type unsequencedCols struct {
	TreeID              int64
	Bucket              int64
	QueueTimestampNanos int64
	MerkleLeafHash      []byte
	LeafIdentityHash    []byte
}
func NewLogStorage(client *spanner.Client) storage.LogStorage {
	return NewLogStorageWithOpts(client, LogStorageOptions{})
}
func NewLogStorageWithOpts(client *spanner.Client, opts LogStorageOptions) storage.LogStorage {
	if got := opts.DequeueAcrossMerkleBucketsRangeFraction; got <= 0 || got > 1.0 {
		opts.DequeueAcrossMerkleBucketsRangeFraction = 1.0
	}
	return &logStorage{
		ts: newTreeStorageWithOpts(client, opts.TreeStorageOptions),
		writeSem: semaphore.NewWeighted(128),
		opts:     opts,
	}
}
type logStorage struct {
	ts *treeStorage
	writeSem *semaphore.Weighted
	opts LogStorageOptions
}
func (ls *logStorage) CheckDatabaseAccessible(ctx context.Context) error {
	return checkDatabaseAccessible(ctx, ls.ts.client)
}
func (ls *logStorage) readOnlyTX() *spanner.ReadOnlyTransaction {
	var staleness spanner.TimestampBound
	if ls.opts.ReadOnlyStaleness > 0 {
		staleness = spanner.ExactStaleness(ls.opts.ReadOnlyStaleness)
	} else {
		staleness = spanner.StrongRead()
	}
	return ls.ts.client.ReadOnlyTransaction().WithTimestampBound(staleness)
}
func (ls *logStorage) GetActiveLogIDs(ctx context.Context) ([]int64, error) {
	ids := []int64{}
	stmt := spanner.NewStatement(getActiveLogIDsSQL)
	rows := ls.readOnlyTX().Query(ctx, stmt)
	if err := rows.Do(func(r *spanner.Row) error {
		var id int64
		if err := r.Columns(&id); err != nil {
			return err
		}
		ids = append(ids, id)
		return nil
	}); err != nil {
		glog.Warningf("GetActiveLogIDs: %v", err)
		return nil, fmt.Errorf("problem executing getActiveLogIDsSQL: %v", err)
	}
	return ids, nil
}
func newLogCache(tree *trillian.Tree) (*cache.SubtreeCache, error) {
	return cache.NewLogSubtreeCache(rfc6962.DefaultHasher), nil
}
func (ls *logStorage) begin(ctx context.Context, tree *trillian.Tree, readonly bool, stx spanRead) (*logTX, error) {
	tx, err := ls.ts.begin(ctx, tree, newLogCache, stx)
	if err != nil {
		return nil, err
	}
	ltx := &logTX{
		ls:       ls,
		dequeued: make(map[string]*QueuedEntry),
		treeTX:   tx,
	}
	if err := ltx.getLatestRoot(ctx); err == storage.ErrTreeNeedsInit {
		return ltx, err
	} else if err != nil {
		tx.Close()
		return nil, err
	}
	return ltx, nil
}
func (ls *logStorage) BeginForTree(ctx context.Context, treeID int64) (storage.LogTreeTX, error) {
	return nil, ErrNotImplemented
}
func (ls *logStorage) ReadWriteTransaction(ctx context.Context, tree *trillian.Tree, f storage.LogTXFunc) error {
	_, err := ls.ts.client.ReadWriteTransaction(ctx, func(ctx context.Context, stx *spanner.ReadWriteTransaction) error {
		tx, err := ls.begin(ctx, tree, false  , stx)
		if err != nil && err != storage.ErrTreeNeedsInit {
			return err
		}
		if err := f(ctx, tx); err != nil {
			return err
		}
		return tx.flushSubtrees(ctx)
	})
	return err
}
func (ls *logStorage) SnapshotForTree(ctx context.Context, tree *trillian.Tree) (storage.ReadOnlyLogTreeTX, error) {
	return ls.begin(ctx, tree, true  , ls.ts.client.ReadOnlyTransaction())
}
func (ls *logStorage) QueueLeaves(ctx context.Context, tree *trillian.Tree, leaves []*trillian.LogLeaf, qTimestamp time.Time) ([]*trillian.QueuedLogLeaf, error) {
	_, treeConfig, err := ls.ts.getTreeAndConfig(ctx, tree)
	if err != nil {
		return nil, err
	}
	config, ok := treeConfig.(*spannerpb.LogStorageConfig)
	if !ok {
		return nil, status.Errorf(codes.Internal, "got unexpected config type for Log operation: %T", treeConfig)
	}
	now := time.Now().UTC().Unix()
	bucketPrefix := (now % config.NumUnseqBuckets) << 8
	results := make([]*trillian.QueuedLogLeaf, len(leaves))
	writeDupes := make(map[string][]int)
	qTS := qTimestamp.UnixNano()
	var wg sync.WaitGroup
	for i, l := range leaves {
		wg.Add(1)
		i := i
		l := l
		go func() {
			defer wg.Done()
			m1, err := spanner.InsertStruct(leafDataTbl, leafDataCols{
				TreeID:              tree.TreeId,
				LeafIdentityHash:    l.LeafIdentityHash,
				LeafValue:           l.LeafValue,
				ExtraData:           l.ExtraData,
				QueueTimestampNanos: qTS,
			})
			if err != nil {
				results[i] = &trillian.QueuedLogLeaf{Status: status.Convert(err).Proto()}
				return
			}
			b := bucketPrefix | int64(l.MerkleLeafHash[0])
			m2, err := spanner.InsertStruct(unseqTable, unsequencedCols{
				TreeID:              tree.TreeId,
				Bucket:              b,
				QueueTimestampNanos: qTS,
				MerkleLeafHash:      l.MerkleLeafHash,
				LeafIdentityHash:    l.LeafIdentityHash,
			})
			if err != nil {
				results[i] = &trillian.QueuedLogLeaf{Status: status.Convert(err).Proto()}
				return
			}
			_, err = ls.ts.client.Apply(ctx, []*spanner.Mutation{m1, m2})
			if spanner.ErrCode(err) == codes.AlreadyExists {
				k := string(l.LeafIdentityHash)
				writeDupes[k] = append(writeDupes[k], i)
			} else if err != nil {
				results[i] = &trillian.QueuedLogLeaf{Status: status.Convert(err).Proto()}
			} else {
				results[i] = &trillian.QueuedLogLeaf{Leaf: l}  
			}
		}()
	}
	wg.Wait()
	err = ls.readDupeLeaves(ctx, tree.TreeId, writeDupes, results)
	if err != nil {
		return nil, err
	}
	return results, nil
}
func (ls *logStorage) AddSequencedLeaves(ctx context.Context, tree *trillian.Tree, leaves []*trillian.LogLeaf, ts time.Time) ([]*trillian.QueuedLogLeaf, error) {
	ctx, span := trace.StartSpan(ctx, "AddSequencedLeaves")
	defer span.End()
	okProto := status.New(codes.OK, "OK").Proto()
	_, span = trace.StartSpan(ctx, "insert")
	defer span.End()
	res := make([]*trillian.QueuedLogLeaf, len(leaves))
	errs := make(chan error, 1)
	var wg sync.WaitGroup
	for i, l := range leaves {
		l.QueueTimestamp = timestamppb.New(ts)
		if err := l.QueueTimestamp.CheckValid(); err != nil {
			return nil, fmt.Errorf("got invalid queue timestamp: %w", err)
		}
		i, l := i, l
		res[i] = &trillian.QueuedLogLeaf{Status: okProto}
		wg.Add(1)
		var err error
		m1, err := spanner.InsertStruct(leafDataTbl, leafDataCols{
			TreeID:              tree.TreeId,
			LeafIdentityHash:    l.LeafIdentityHash,
			LeafValue:           l.LeafValue,
			ExtraData:           l.ExtraData,
			QueueTimestampNanos: ts.UnixNano(),
		})
		if err != nil {
			return nil, err
		}
		m2, err := spanner.InsertStruct(seqDataTbl, sequencedLeafDataCols{
			TreeID:                  tree.TreeId,
			SequenceNumber:          l.LeafIndex,
			LeafIdentityHash:        l.LeafIdentityHash,
			MerkleLeafHash:          l.MerkleLeafHash,
			IntegrateTimestampNanos: 0,
		})
		if err != nil {
			return nil, err
		}
		m := []*spanner.Mutation{m1, m2}
		doneFunc := func(err error) {
			defer wg.Done()
			if err != nil {
				if status.Code(err) == codes.AlreadyExists {
					glog.Infof("Found already exists: index=%v, id=%v", l.LeafIndex, l.LeafIdentityHash)
					res[i].Status = status.New(codes.FailedPrecondition, "conflicting LeafIndex or LeafIdentityHash").Proto()
					return
				}
				select {
				case errs <- err:
				default:  
				}
			}
		}
		if err := ls.writeSem.Acquire(ctx, 1); err != nil {
			doneFunc(err)
		} else {
			go func() {
				defer ls.writeSem.Release(1)
				doneFunc(func() error {
					_, err := ls.ts.client.Apply(ctx, m)
					return err
				}())
			}()
		}
	}
	span.End()
	_, span = trace.StartSpan(ctx, "wait")
	wg.Wait()
	span.End()
	select {
	case err := <-errs:
		return nil, err
	default:  
	}
	return res, nil
}