text:
package temptable

import (
	"context"
	"math"
	"testing"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/parser/model"
	"github.com/pingcap/tidb/store/driver/txn"
	"github.com/pingcap/tidb/tablecodec"
	"github.com/pingcap/tidb/util/codec"
	"github.com/pingcap/tidb/util/mock"
	"github.com/stretchr/testify/require"
)

func incLastByte(key kv.Key) kv.Key {
	key = append([]byte{}, key...)
	key[len(key)-1] += 1
	return key
}

func decLastByte(key kv.Key) kv.Key {
	key = append([]byte{}, key...)
	key[len(key)-1] -= 1
	return key
}

func encodeTableKey(tblID int64, suffix ...byte) kv.Key {
	key := tablecodec.EncodeTablePrefix(tblID)
	key = append(key, suffix...)
	return key
}

func TestGetKeyAccessedTableID(t *testing.T) {
	t.Parallel()
	tbPrefix := tablecodec.TablePrefix()
	prefix0 := encodeTableKey(0)
	prefixMax := encodeTableKey(math.MaxInt64)
	prefixNegative := encodeTableKey(-1)
	prefix1 := encodeTableKey(1)
	prefixA := encodeTableKey(math.MaxInt64 / 2)
	prefixB := encodeTableKey(math.MaxInt64 - 1)

	cases := []struct {
		name       string
		key        kv.Key
		ok         bool
		testSuffix bool
		tbID       int64
	}{
		{name: "empty", key: []byte{}, ok: false},
		{name: "replace1", key: incLastByte(tbPrefix), ok: false},
		{name: "replace2", key: decLastByte(tbPrefix), ok: false},
		// key with not enough id len should not be regard as a valid table id
		{name: "tbPrefix", key: tbPrefix, ok: false},
		{name: "back1", key: prefix1[:len(prefix1)-1], tbID: 1, ok: false},
		{name: "back2", key: prefix1[:len(tbPrefix)+1], tbID: 1, ok: false},
		// table with an id 0 should not be regard as a valid table id
		{name: "prefix0", key: prefix0, testSuffix: true, ok: false},
		// table with id math.MaxInt64 should not regard as a valid table id
		{name: "prefixMax", key: prefixMax, testSuffix: true, ok: false},
		// table with id negative should not regard as a valid table id
		{name: "prefixNegative", key: prefixNegative, testSuffix: true, ok: false},
		// table with id > 0 && id < math.MaxInt64 regard as a valid table id
		{name: "prefix1", key: prefix1, tbID: 1, testSuffix: true, ok: true},
		{name: "prefixA", key: prefixA, tbID: math.MaxInt64 / 2, testSuffix: true, ok: true},
		{name: "prefixB", key: prefixB, tbID: math.MaxInt64 - 1, testSuffix: true, ok: true},
	}

	for _, c := range cases {
		keys := []kv.Key{c.key}
		if c.testSuffix {
			for _, s := range [][]byte{
				{0},
				{1},
				{0xFF},
				codec.EncodeInt(nil, 0),
				codec.EncodeInt(nil, math.MaxInt64/2),
				codec.EncodeInt(nil, math.MaxInt64),
			} {
				newKey := append([]byte{}, c.key...)
				newKey = append(newKey, s...)
				keys = append(keys, newKey)
			}
		}

		for i, key := range keys {
			tblID, ok := getKeyAccessedTableID(key)
			require.Equal(t, c.ok, ok, "%s %d", c.name, i)
			if c.ok {
				require.Equal(t, c.tbID, tblID, "%s %d", c.name, i)
			} else {
				require.Equal(t, int64(0), tblID, "%s %d", c.name, i)
			}
		}
	}
}

func TestGetRangeAccessedTableID(t *testing.T) {
	t.Parallel()
	cases := []struct {
		start kv.Key
		end   kv.Key
		ok    bool
		tbID  int64
	}{
		{
			start: encodeTableKey(1),
			end:   encodeTableKey(1),
			ok:    true,
			tbID:  1,
		},
		{
			start: encodeTableKey(1),
			end:   append(encodeTableKey(1), 0),
			ok:    true,
			tbID:  1,
		},
		{
			start: encodeTableKey(1),
			end:   append(encodeTableKey(1), 0xFF),
			ok:    true,
			tbID:  1,
		},
		{
			start: encodeTableKey(1),
			end:   encodeTableKey(2),
			ok:    true,
			tbID:  1,
		},
		{
			start: tablecodec.TablePrefix(),
			end:   encodeTableKey(1),
			ok:    false,
		},
		{
			start: tablecodec.TablePrefix(),
			end:   nil,
			ok:    false,
		},
		{
			start: encodeTableKey(0),
			end:   encodeTableKey(1),
			ok:    false,
		},
		{
			start: encodeTableKey(0),
			end:   nil,
			ok:    false,
		},
		{
			start: encodeTableKey(1),
			end:   encodeTableKey(5),
			ok:    false,
		},
		{
			start: encodeTableKey(1),
			end:   nil,
			ok:    false,
		},
		{
			start: encodeTableKey(1),
			end:   incLastByte(tablecodec.TablePrefix()),
			ok:    false,
		},
		{
			start: encodeTableKey(1),
			end:   encodeTableKey(1)[:len(encodeTableKey(1))-1],
			ok:    false,
		},
		{
			start: encodeTableKey(1)[:len(encodeTableKey(1))-1],
			end:   encodeTableKey(1),
			ok:    false,
		},
		{
			start: encodeTableKey(math.MaxInt64),
			end:   encodeTableKey(math.MaxInt64, 0),
			ok:    false,
		},
		{
			start: nil,
			end:   nil,
			ok:    false,
		},
		{
			start: nil,
			end:   encodeTableKey(2),
			ok:    false,
		},
	}

	for _, c := range cases {
		tblID, ok := getRangeAccessedTableID(c.start, c.end)
		require.Equal(t, c.ok, ok)
		if ok {
			require.Equal(t, c.tbID, tblID)
		} else {
			require.Equal(t, int64(0), tblID)
		}
	}
}

func TestNotTableRange(t *testing.T) {
	t.Parallel()
	falseCases := [][]kv.Key{
		{nil, nil},
		{nil, encodeTableKey(1, 0)},
		{nil, encodeTableKey(1, 1)},
		{encodeTableKey(1), nil},
		{encodeTableKey(1, 0), nil},
		{encodeTableKey(1), encodeTableKey(1)},
		{encodeTableKey(1), encodeTableKey(1, 0)},
		{encodeTableKey(1), encodeTableKey(1, 1)},
		{encodeTableKey(1), encodeTableKey(2)},
		{encodeTableKey(1), encodeTableKey(2, 0)},
		{encodeTableKey(1), encodeTableKey(2, 1)},
		{encodeTableKey(1, 0), encodeTableKey(1, 1)},
		{encodeTableKey(1), incLastByte(tablecodec.TablePrefix())},
	}

	tablePrefix := tablecodec.TablePrefix()
	trueCases := [][]kv.Key{
		{nil, decLastByte(tablePrefix)},
		{decLastByte(tablePrefix), append(decLastByte(tablePrefix), 1)},
		{incLastByte(tablePrefix), nil},
		{incLastByte(tablePrefix), append(incLastByte(tablePrefix), 1)},
	}

	for _, c := range falseCases {
		require.False(t, notTableRange(c[0], c[1]))
	}

	for _, c := range trueCases {
		require.True(t, notTableRange(c[0], c[1]))
	}
}

func TestGetSessionTemporaryTableKey(t *testing.T) {
	t.Parallel()
	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}

	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)

	normalTb, ok := is.TableByID(1)
	require.True(t, ok)
	require.Equal(t, model.TempTableNone, normalTb.Meta().TempTableType)
	globalTb, ok := is.TableByID(3)
	require.True(t, ok)
	require.Equal(t, model.TempTableGlobal, globalTb.Meta().TempTableType)
	localTb, ok := is.TableByID(5)
	require.True(t, ok)
	require.Equal(t, model.TempTableLocal, localTb.Meta().TempTableType)

	retriever := newMockedRetriever(t).SetAllowedMethod("Get").SetData(localTempTableData)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	// test local temporary table should read from session
	cases := append(append([]*kv.Entry{}, localTempTableData...), &kv.Entry{
		// also add a test case for key not exist in retriever
		Key: encodeTableKey(5, 'n'), Value: []byte("non-exist-key"),
	})
	for i, c := range cases {
		val, err := getSessionKey(ctx, localTb.Meta(), retriever, c.Key)
		if len(c.Value) == 0 || string(c.Value) == "non-exist-key" {
			require.True(t, kv.ErrNotExist.Equal(err), i)
			require.Nil(t, val, i)
		} else {
			require.NoError(t, err, i)
			require.Equal(t, c.Value, val, i)
		}
		invokes := retriever.GetInvokes()
		require.Equal(t, 1, len(invokes), i)
		require.Equal(t, "Get", invokes[0].Method, i)
		require.Equal(t, []interface{}{ctx, c.Key}, invokes[0].Args)
		retriever.ResetInvokes()

		// test for nil session
		val, err = getSessionKey(ctx, localTb.Meta(), nil, c.Key)
		require.True(t, kv.ErrNotExist.Equal(err), i)
		require.Nil(t, val, i)
		require.Equal(t, 0, len(retriever.GetInvokes()), i)
	}

	// test global temporary table should return empty data directly
	val, err := getSessionKey(ctx, globalTb.Meta(), retriever, encodeTableKey(3))
	require.True(t, kv.ErrNotExist.Equal(err))
	require.Nil(t, val)
	require.Equal(t, 0, len(retriever.GetInvokes()))

	// test normal table should not be allowed
	val, err = getSessionKey(ctx, normalTb.Meta(), retriever, encodeTableKey(1))
	require.Error(t, err, "Cannot get normal table key from session")
	require.Nil(t, val)
	require.Equal(t, 0, len(retriever.GetInvokes()))

	// test for other errors
	injectedErr := errors.New("err")
	retriever.InjectMethodError("Get", injectedErr)
	val, err = getSessionKey(ctx, localTb.Meta(), retriever, encodeTableKey(5))
	require.Nil(t, val)
	require.Equal(t, injectedErr, err)
}

func TestInterceptorTemporaryTableInfoByID(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1, 5).
		AddTable(model.TempTableGlobal, 2, 6).
		AddTable(model.TempTableLocal, 3, 7)

	interceptor := NewTemporaryTableSnapshotInterceptor(is, newMockedRetriever(t))

	// normal table should return nil
	tblInfo, ok := interceptor.temporaryTableInfoByID(1)
	require.False(t, ok)
	require.Nil(t, tblInfo)

	tblInfo, ok = interceptor.temporaryTableInfoByID(5)
	require.False(t, ok)
	require.Nil(t, tblInfo)

	// global temporary table
	tblInfo, ok = interceptor.temporaryTableInfoByID(2)
	require.True(t, ok)
	require.Equal(t, "tb2", tblInfo.Name.O)
	require.Equal(t, model.TempTableGlobal, tblInfo.TempTableType)

	tblInfo, ok = interceptor.temporaryTableInfoByID(6)
	require.True(t, ok)
	require.Equal(t, "tb6", tblInfo.Name.O)
	require.Equal(t, model.TempTableGlobal, tblInfo.TempTableType)

	// local temporary table
	tblInfo, ok = interceptor.temporaryTableInfoByID(3)
	require.True(t, ok)
	require.Equal(t, "tb3", tblInfo.Name.O)
	require.Equal(t, model.TempTableLocal, tblInfo.TempTableType)

	tblInfo, ok = interceptor.temporaryTableInfoByID(7)
	require.True(t, ok)
	require.Equal(t, "tb7", tblInfo.Name.O)
	require.Equal(t, model.TempTableLocal, tblInfo.TempTableType)

	// non exists table
	tblInfo, ok = interceptor.temporaryTableInfoByID(4)
	require.False(t, ok)
	require.Nil(t, tblInfo)

	tblInfo, ok = interceptor.temporaryTableInfoByID(8)
	require.False(t, ok)
	require.Nil(t, tblInfo)
}

func TestInterceptorOnGet(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)
	noTempTableData := []*kv.Entry{
		// normal table data
		{Key: encodeTableKey(1), Value: []byte("v1")},
		{Key: encodeTableKey(1, 1), Value: []byte("v11")},
		// no exist table data
		{Key: encodeTableKey(2), Value: []byte("v2")},
		{Key: encodeTableKey(2, 1), Value: []byte("v21")},
		// other data
		{Key: kv.Key("s"), Value: []byte("vs")},
		{Key: kv.Key("s0"), Value: []byte("vs0")},
		{Key: kv.Key("u"), Value: []byte("vu")},
		{Key: kv.Key("u0"), Value: []byte("vu0")},
		{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
		{Key: encodeTableKey(0), Value: []byte("v0")},
		{Key: encodeTableKey(0, 1), Value: []byte("v01")},
		{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
		{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
	}

	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}

	snap := newMockedSnapshot(newMockedRetriever(t).SetAllowedMethod("Get").SetData(noTempTableData))
	retriever := newMockedRetriever(t).SetAllowedMethod("Get").SetData(localTempTableData)
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	// test normal table and no table key should read from snapshot
	cases := append(append([]*kv.Entry{}, noTempTableData...), []*kv.Entry{
		// also add a test case for key not exist in snap
		{Key: encodeTableKey(1, 'n'), Value: []byte("non-exist-key")},
		{Key: encodeTableKey(2, 'n'), Value: []byte("non-exist-key")},
		{Key: kv.Key("sn"), Value: []byte("non-exist-key")},
		{Key: kv.Key("un"), Value: []byte("non-exist-key")},
	}...)
	for i, c := range cases {
		for _, emptyRetriever := range []bool{false, true} {
			inter := interceptor
			if emptyRetriever {
				inter = emptyRetrieverInterceptor
			}
			val, err := inter.OnGet(ctx, snap, c.Key)
			if string(c.Value) == "non-exist-key" {
				require.True(t, kv.ErrNotExist.Equal(err), i)
				require.Nil(t, val, i)
			} else {
				require.NoError(t, err, i)
				require.Equal(t, c.Value, val, i)
			}
			require.Equal(t, 0, len(retriever.GetInvokes()))
			invokes := snap.GetInvokes()
			require.Equal(t, 1, len(invokes), i)
			require.Equal(t, "Get", invokes[0].Method, i)
			require.Equal(t, []interface{}{ctx, c.Key}, invokes[0].Args)
			snap.ResetInvokes()
		}
	}

	// test global temporary table should return kv.ErrNotExist
	val, err := interceptor.OnGet(ctx, snap, encodeTableKey(3))
	require.True(t, kv.ErrNotExist.Equal(err))
	require.Nil(t, val)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 0, len(snap.GetInvokes()))

	val, err = interceptor.OnGet(ctx, snap, encodeTableKey(3, 1))
	require.True(t, kv.ErrNotExist.Equal(err))
	require.Nil(t, val)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 0, len(snap.GetInvokes()))

	val, err = emptyRetrieverInterceptor.OnGet(ctx, snap, encodeTableKey(3, 1))
	require.True(t, kv.ErrNotExist.Equal(err))
	require.Nil(t, val)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 0, len(snap.GetInvokes()))

	// test local temporary table should read from session
	cases = append(append([]*kv.Entry{}, localTempTableData...), &kv.Entry{
		// also add a test case for key not exist in retriever
		Key: encodeTableKey(5, 'n'), Value: []byte("non-exist-key"),
	})
	for i, c := range cases {
		val, err = interceptor.OnGet(ctx, snap, c.Key)
		if len(c.Value) == 0 || string(c.Value) == "non-exist-key" {
			require.True(t, kv.ErrNotExist.Equal(err), i)
			require.Nil(t, val, i)
		} else {
			require.NoError(t, err, i)
			require.Equal(t, c.Value, val, i)
		}
		require.Equal(t, 0, len(snap.GetInvokes()), i)
		invokes := retriever.GetInvokes()
		require.Equal(t, 1, len(invokes), i)
		require.Equal(t, "Get", invokes[0].Method, i)
		require.Equal(t, []interface{}{ctx, c.Key}, invokes[0].Args)
		retriever.ResetInvokes()

		val, err = emptyRetrieverInterceptor.OnGet(ctx, snap, c.Key)
		require.True(t, kv.ErrNotExist.Equal(err))
		require.Nil(t, val)
		require.Equal(t, 0, len(snap.GetInvokes()), i)
		require.Equal(t, 0, len(retriever.GetInvokes()), i)
	}

	// test error cases
	injectedErr := errors.New("err1")
	snap.InjectMethodError("Get", injectedErr)
	val, err = interceptor.OnGet(ctx, snap, encodeTableKey(1))
	require.Nil(t, val)
	require.Equal(t, injectedErr, err)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 1, len(snap.GetInvokes()))

	val, err = interceptor.OnGet(ctx, snap, kv.Key("s"))
	require.Nil(t, val)
	require.Equal(t, injectedErr, err)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 2, len(snap.GetInvokes()))
	snap.ResetInvokes()
	require.Equal(t, 0, len(snap.GetInvokes()))

	injectedErr = errors.New("err2")
	retriever.InjectMethodError("Get", injectedErr)
	val, err = interceptor.OnGet(ctx, snap, encodeTableKey(5))
	require.Nil(t, val)
	require.Equal(t, injectedErr, err)
	require.Equal(t, 0, len(snap.GetInvokes()))
	require.Equal(t, 1, len(retriever.GetInvokes()))
	retriever.ResetInvokes()
	require.Equal(t, 0, len(retriever.GetInvokes()))
}

func TestInterceptorBatchGetTemporaryTableKeys(t *testing.T) {
	t.Parallel()
	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
		{Key: encodeTableKey(8), Value: []byte("v8")},
		{Key: encodeTableKey(8, 0), Value: []byte("v80")},
		{Key: encodeTableKey(8, 1), Value: []byte("v81")},
		{Key: encodeTableKey(8, 0, 1), Value: []byte("v801")},
		{Key: encodeTableKey(8, 2), Value: []byte("")},
		{Key: encodeTableKey(8, 3), Value: nil},
	}

	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1, 4).
		AddTable(model.TempTableGlobal, 3, 6).
		AddTable(model.TempTableLocal, 5, 8)
	retriever := newMockedRetriever(t).SetAllowedMethod("Get").SetData(localTempTableData)
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	cases := []struct {
		keys       []kv.Key
		snapKeys   []kv.Key
		nilSession bool
		result     map[string][]byte
	}{
		{
			keys:     nil,
			snapKeys: nil,
			result:   nil,
		},
		{
			keys: []kv.Key{
				encodeTableKey(3),
				encodeTableKey(3, 1),
				encodeTableKey(6),
			},
			snapKeys: nil,
			result:   nil,
		},
		{
			keys: []kv.Key{
				encodeTableKey(3),
				encodeTableKey(5, 'n'),
			},
			snapKeys: nil,
			result:   nil,
		},
		{
			keys: []kv.Key{
				encodeTableKey(5, 'n'),
			},
			snapKeys: nil,
			result:   nil,
		},
		{
			keys: []kv.Key{
				encodeTableKey(0),
				encodeTableKey(1),
				encodeTableKey(2),
				encodeTableKey(math.MaxInt64),
				tablecodec.TablePrefix(),
				kv.Key("s"),
				kv.Key("v"),
			},
			snapKeys: []kv.Key{
				encodeTableKey(0),
				encodeTableKey(1),
				encodeTableKey(2),
				encodeTableKey(math.MaxInt64),
				tablecodec.TablePrefix(),
				kv.Key("s"),
				kv.Key("v"),
			},
			result: nil,
		},
		{
			keys: []kv.Key{
				encodeTableKey(5),
				encodeTableKey(5, 2),
				encodeTableKey(5, 'n'),
				encodeTableKey(8, 1),
			},
			snapKeys: nil,
			result: map[string][]byte{
				string(encodeTableKey(5)):    []byte("v5"),
				string(encodeTableKey(8, 1)): []byte("v81"),
			},
		},
		{
			keys: []kv.Key{
				encodeTableKey(5),
				encodeTableKey(1),
				encodeTableKey(5, 'n'),
				encodeTableKey(8, 1),
			},
			snapKeys: []kv.Key{encodeTableKey(1)},
			result: map[string][]byte{
				string(encodeTableKey(5)):    []byte("v5"),
				string(encodeTableKey(8, 1)): []byte("v81"),
			},
		},
		{
			keys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(5),
				encodeTableKey(1),
				encodeTableKey(5, 2),
				encodeTableKey(5, 'n'),
				encodeTableKey(8, 1),
			},
			snapKeys: []kv.Key{tablecodec.TablePrefix(), encodeTableKey(1)},
			result: map[string][]byte{
				string(encodeTableKey(5)):    []byte("v5"),
				string(encodeTableKey(8, 1)): []byte("v81"),
			},
		},
		{
			keys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(5),
				encodeTableKey(1),
				encodeTableKey(5, 2),
				encodeTableKey(5, 'n'),
				encodeTableKey(8, 1),
			},
			snapKeys:   []kv.Key{tablecodec.TablePrefix(), encodeTableKey(1)},
			nilSession: true,
			result:     nil,
		},
	}
	for i, c := range cases {
		inter := interceptor
		if c.nilSession {
			inter = emptyRetrieverInterceptor
		}
		snapKeys, result, err := inter.batchGetTemporaryTableKeys(ctx, c.keys)
		require.NoError(t, err, i)
		if c.snapKeys == nil {
			require.Nil(t, snapKeys, i)
		} else {
			require.Equal(t, c.snapKeys, snapKeys, i)
		}

		if c.result == nil {
			require.Nil(t, result, i)
		} else {
			require.Equal(t, c.result, result, i)
		}

		if c.nilSession {
			require.Equal(t, 0, len(retriever.GetInvokes()))
		}

		for j, invoke := range retriever.GetInvokes() {
			require.Equal(t, "Get", invoke.Method, "%d, %d", i, j)
			require.Equal(t, ctx, invoke.Args[0], "%d, %d", i, j)
		}
		retriever.ResetInvokes()
	}

	// test for error occurs
	injectedErr := errors.New("err")
	retriever.InjectMethodError("Get", injectedErr)
	snapKeys, result, err := interceptor.batchGetTemporaryTableKeys(ctx, []kv.Key{
		tablecodec.TablePrefix(),
		encodeTableKey(5),
		encodeTableKey(1),
		encodeTableKey(5, 'n'),
		encodeTableKey(8, 1),
	})
	require.Nil(t, snapKeys)
	require.Nil(t, result)
	require.Equal(t, injectedErr, err)
	retriever.ResetInvokes()
}

func TestInterceptorOnBatchGet(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)
	noTempTableData := []*kv.Entry{
		// normal table data
		{Key: encodeTableKey(1), Value: []byte("v1")},
		{Key: encodeTableKey(1, 1), Value: []byte("v11")},
		// no exist table data
		{Key: encodeTableKey(2), Value: []byte("v2")},
		{Key: encodeTableKey(2, 1), Value: []byte("v21")},
		// other data
		{Key: kv.Key("s"), Value: []byte("vs")},
		{Key: kv.Key("s0"), Value: []byte("vs0")},
		{Key: kv.Key("u"), Value: []byte("vu")},
		{Key: kv.Key("u0"), Value: []byte("vu0")},
		{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
		{Key: encodeTableKey(0), Value: []byte("v0")},
		{Key: encodeTableKey(0, 1), Value: []byte("v01")},
		{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
		{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
	}

	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}

	snap := newMockedSnapshot(newMockedRetriever(t).SetAllowedMethod("BatchGet").SetData(noTempTableData))
	retriever := newMockedRetriever(t).SetAllowedMethod("Get").SetData(localTempTableData)
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	cases := []struct {
		keys       []kv.Key
		snapKeys   []kv.Key
		nilSession bool
		result     map[string][]byte
	}{
		{
			keys:     nil,
			snapKeys: nil,
			result:   make(map[string][]byte),
		},
		{
			keys: []kv.Key{
				encodeTableKey(3),
				encodeTableKey(5, 'n'),
			},
			snapKeys: nil,
			result:   make(map[string][]byte),
		},
		{
			keys: []kv.Key{
				encodeTableKey(7),
				encodeTableKey(1, 'n'),
				[]byte("o"),
			},
			snapKeys: []kv.Key{
				encodeTableKey(7),
				encodeTableKey(1, 'n'),
				[]byte("o"),
			},
			result: make(map[string][]byte),
		},
		{
			keys: []kv.Key{
				encodeTableKey(3),
				encodeTableKey(5),
				encodeTableKey(5, 1),
				encodeTableKey(5, 2),
			},
			snapKeys: nil,
			result: map[string][]byte{
				string(encodeTableKey(5)):    []byte("v5"),
				string(encodeTableKey(5, 1)): []byte("v51"),
			},
		},
		{
			keys: []kv.Key{
				encodeTableKey(0),
				encodeTableKey(1),
				encodeTableKey(2),
				encodeTableKey(math.MaxInt64),
				tablecodec.TablePrefix(),
				kv.Key("s"),
				kv.Key("u"),
				encodeTableKey(9),
			},
			snapKeys: []kv.Key{
				encodeTableKey(0),
				encodeTableKey(1),
				encodeTableKey(2),
				encodeTableKey(math.MaxInt64),
				tablecodec.TablePrefix(),
				kv.Key("s"),
				kv.Key("u"),
				encodeTableKey(9),
			},
			result: map[string][]byte{
				string(encodeTableKey(0)):             []byte("v0"),
				string(encodeTableKey(1)):             []byte("v1"),
				string(encodeTableKey(2)):             []byte("v2"),
				string(encodeTableKey(math.MaxInt64)): []byte("vm"),
				string(tablecodec.TablePrefix()):      []byte("vt"),
				"s":                                   []byte("vs"),
				"u":                                   []byte("vu"),
			},
		},
		{
			keys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(5),
				encodeTableKey(1),
				encodeTableKey(5, 2),
				encodeTableKey(5, 'n'),
				encodeTableKey(1, 'n'),
			},
			snapKeys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(1),
				encodeTableKey(1, 'n'),
			},
			result: map[string][]byte{
				string(tablecodec.TablePrefix()): []byte("vt"),
				string(encodeTableKey(5)):        []byte("v5"),
				string(encodeTableKey(1)):        []byte("v1"),
			},
		},
		{
			keys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(5),
				encodeTableKey(1),
				encodeTableKey(5, 2),
				encodeTableKey(5, 'n'),
				encodeTableKey(1, 'n'),
			},
			nilSession: true,
			snapKeys: []kv.Key{
				tablecodec.TablePrefix(),
				encodeTableKey(1),
				encodeTableKey(1, 'n'),
			},
			result: map[string][]byte{
				string(tablecodec.TablePrefix()): []byte("vt"),
				string(encodeTableKey(1)):        []byte("v1"),
			},
		},
	}

	for i, c := range cases {
		inter := interceptor
		if c.nilSession {
			inter = emptyRetrieverInterceptor
		}
		result, err := inter.OnBatchGet(ctx, snap, c.keys)
		require.Nil(t, err, i)
		require.NotNil(t, result, i)
		require.Equal(t, c.result, result, i)
		if c.nilSession {
			require.Equal(t, 0, len(retriever.GetInvokes()))
		}
		for j, invoke := range retriever.GetInvokes() {
			require.Equal(t, "Get", invoke.Method, "%d, %d", i, j)
			require.Equal(t, ctx, invoke.Args[0], "%d, %d", i, j)
		}
		if len(c.snapKeys) > 0 {
			require.Equal(t, 1, len(snap.GetInvokes()), i)
			require.Equal(t, "BatchGet", snap.GetInvokes()[0].Method, i)
			require.Equal(t, ctx, snap.GetInvokes()[0].Args[0], i)
			require.Equal(t, c.snapKeys, snap.GetInvokes()[0].Args[1], i)
		} else {
			require.Equal(t, 0, len(snap.GetInvokes()), i)
		}

		retriever.ResetInvokes()
		snap.ResetInvokes()
	}

	// test session error occurs
	sessionErr := errors.New("errSession")
	retriever.InjectMethodError("Get", sessionErr)
	result, err := interceptor.OnBatchGet(ctx, snap, []kv.Key{encodeTableKey(5)})
	require.Nil(t, result)
	require.Equal(t, sessionErr, err)

	snapErr := errors.New("errSnap")
	snap.InjectMethodError("BatchGet", snapErr)
	result, err = interceptor.OnBatchGet(ctx, snap, []kv.Key{encodeTableKey(1)})
	require.Nil(t, result)
	require.Equal(t, snapErr, err)
}

func TestCreateUnionIter(t *testing.T) {
	t.Parallel()
	retriever := newMockedRetriever(t).SetData([]*kv.Entry{
		{Key: kv.Key("k1"), Value: []byte("v1")},
		{Key: kv.Key("k10"), Value: []byte("")},
		{Key: kv.Key("k11"), Value: []byte("v11")},
		{Key: kv.Key("k5"), Value: []byte("v5")},
	})

	snap := newMockedSnapshot(newMockedRetriever(t).SetData([]*kv.Entry{
		{Key: kv.Key("k2"), Value: []byte("v2")},
		{Key: kv.Key("k20"), Value: []byte("v20")},
		{Key: kv.Key("k21"), Value: []byte("v21")},
	}))

	cases := []struct {
		args    []kv.Key
		reverse bool
		nilSess bool
		nilSnap bool
		result  []kv.Entry
	}{
		{
			args: []kv.Key{kv.Key("k1"), kv.Key("k21")},
			result: []kv.Entry{
				{Key: kv.Key("k1"), Value: []byte("v1")},
				{Key: kv.Key("k11"), Value: []byte("v11")},
				{Key: kv.Key("k2"), Value: []byte("v2")},
				{Key: kv.Key("k20"), Value: []byte("v20")},
			},
		},
		{
			args:    []kv.Key{kv.Key("k21")},
			reverse: true,
			result: []kv.Entry{
				{Key: kv.Key("k20"), Value: []byte("v20")},
				{Key: kv.Key("k2"), Value: []byte("v2")},
				{Key: kv.Key("k11"), Value: []byte("v11")},
				{Key: kv.Key("k1"), Value: []byte("v1")},
			},
		},
		{
			args:    []kv.Key{kv.Key("k1"), kv.Key("k21")},
			nilSnap: true,
			result: []kv.Entry{
				{Key: kv.Key("k1"), Value: []byte("v1")},
				{Key: kv.Key("k11"), Value: []byte("v11")},
			},
		},
		{
			args:    []kv.Key{kv.Key("k21")},
			nilSnap: true,
			reverse: true,
			result: []kv.Entry{
				{Key: kv.Key("k11"), Value: []byte("v11")},
				{Key: kv.Key("k1"), Value: []byte("v1")},
			},
		},
		{
			args:    []kv.Key{kv.Key("k1"), kv.Key("k21")},
			nilSess: true,
			result: []kv.Entry{
				{Key: kv.Key("k2"), Value: []byte("v2")},
				{Key: kv.Key("k20"), Value: []byte("v20")},
			},
		},
		{
			args:    []kv.Key{kv.Key("k21")},
			nilSess: true,
			reverse: true,
			result: []kv.Entry{
				{Key: kv.Key("k20"), Value: []byte("v20")},
				{Key: kv.Key("k2"), Value: []byte("v2")},
			},
		},
	}

	retriever.SetAllowedMethod("Iter", "IterReverse")
	snap.SetAllowedMethod("Iter", "IterReverse")
	for i, c := range cases {
		var iter kv.Iterator
		var err error
		var method string
		var sessArg kv.Retriever
		if !c.nilSess {
			sessArg = retriever
		}

		var snapArg kv.Snapshot
		if !c.nilSnap {
			snapArg = snap
		}

		if c.reverse {
			method = "IterReverse"
			iter, err = createUnionIter(sessArg, snapArg, nil, c.args[0], c.reverse)
		} else {
			method = "Iter"
			iter, err = createUnionIter(sessArg, snapArg, c.args[0], c.args[1], c.reverse)
		}

		require.NoError(t, err, i)

		if c.nilSess && c.nilSnap {
			require.IsType(t, &kv.EmptyIterator{}, iter, i)
		} else if !c.nilSess {
			require.IsType(t, &txn.UnionIter{}, iter, i)
		} else {
			require.IsType(t, &mock.MockedIter{}, iter, i)
		}

		if !c.nilSess {
			require.Equal(t, 1, len(retriever.GetInvokes()), i)
			require.Equal(t, method, retriever.GetInvokes()[0].Method, i)
			require.Equal(t, c.args[0], retriever.GetInvokes()[0].Args[0].(kv.Key), i)
			if !c.reverse {
				require.Equal(t, c.args[1], retriever.GetInvokes()[0].Args[1].(kv.Key), i)
			}
		}

		if !c.nilSnap {
			require.Equal(t, 1, len(snap.GetInvokes()), i)
			require.Equal(t, method, snap.GetInvokes()[0].Method, i)
			require.Equal(t, c.args[0], snap.GetInvokes()[0].Args[0].(kv.Key), i)
			if !c.reverse {
				require.Equal(t, c.args[1], snap.GetInvokes()[0].Args[1].(kv.Key), i)
			}
		}

		result := make([]kv.Entry, 0)
		for iter.Valid() {
			result = append(result, kv.Entry{Key: iter.Key(), Value: iter.Value()})
			require.NoError(t, iter.Next(), i)
		}
		require.Equal(t, c.result, result, i)

		retriever.ResetInvokes()
		snap.ResetInvokes()
	}
}

func TestErrorCreateUnionIter(t *testing.T) {
	t.Parallel()
	retriever := newMockedRetriever(t).SetAllowedMethod("Iter", "IterReverse").SetData([]*kv.Entry{
		{Key: kv.Key("k1"), Value: []byte("")},
	})
	snap := newMockedSnapshot(newMockedRetriever(t).SetAllowedMethod("Iter", "IterReverse").SetData([]*kv.Entry{
		{Key: kv.Key("k1"), Value: []byte("v1")},
	}))

	// test for not allow iter with lowerBound
	iter, err := createUnionIter(retriever, snap, kv.Key("k1"), kv.Key("k2"), true)
	require.Error(t, err, "k should be nil for iter reverse")
	require.Nil(t, iter)
	require.Equal(t, 0, len(retriever.GetInvokes()))
	require.Equal(t, 0, len(snap.GetInvokes()))

	// test for iter next error
	iterNextErr := errors.New("iterNextErr")
	retriever.InjectMethodError("IterNext", iterNextErr)
	iter, err = createUnionIter(retriever, snap, kv.Key("k1"), kv.Key("k2"), false)
	require.Equal(t, iterNextErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, false)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	// test for iter reverse next error
	iterReverseNextErr := errors.New("iterReverseNextErr")
	retriever.InjectMethodError("IterReverseNext", iterReverseNextErr)
	iter, err = createUnionIter(retriever, snap, nil, kv.Key("k2"), true)
	require.Equal(t, iterReverseNextErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, true)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	// test for creating session iter error occurs
	sessionIterErr := errors.New("sessionIterErr")
	retriever.InjectMethodError("Iter", sessionIterErr)

	iter, err = createUnionIter(retriever, snap, kv.Key("k1"), kv.Key("k2"), false)
	require.Equal(t, sessionIterErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, false)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	iter, err = createUnionIter(retriever, nil, kv.Key("k1"), kv.Key("k2"), false)
	require.Equal(t, sessionIterErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, false)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	// test for creating session reverse iter occurs
	sessionIterReverseErr := errors.New("sessionIterReverseErr")
	retriever.InjectMethodError("IterReverse", sessionIterReverseErr)

	iter, err = createUnionIter(retriever, snap, nil, kv.Key("k2"), true)
	require.Equal(t, sessionIterReverseErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, true)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	iter, err = createUnionIter(retriever, snap, nil, kv.Key("k2"), true)
	require.Equal(t, sessionIterReverseErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, true)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	// test for creating snap iter error occurs
	snapIterErr := errors.New("snapIterError")
	snap.InjectMethodError("Iter", snapIterErr)

	iter, err = createUnionIter(nil, snap, kv.Key("k1"), kv.Key("k2"), false)
	require.Equal(t, snapIterErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, false)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	iter, err = createUnionIter(nil, snap, kv.Key("k1"), kv.Key("k2"), false)
	require.Equal(t, snapIterErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, false)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	// test for creating snap reverse iter error occurs
	snapIterReverseErr := errors.New("snapIterError")
	snap.InjectMethodError("IterReverse", snapIterReverseErr)

	iter, err = createUnionIter(nil, snap, nil, kv.Key("k2"), true)
	require.Equal(t, snapIterReverseErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, true)
	retriever.ResetInvokes()
	snap.ResetInvokes()

	iter, err = createUnionIter(nil, snap, nil, kv.Key("k2"), true)
	require.Equal(t, snapIterReverseErr, err)
	require.Nil(t, iter)
	checkCreatedIterClosed(t, retriever, snap, true)
	retriever.ResetInvokes()
	snap.ResetInvokes()
}

func checkCreatedIterClosed(t *testing.T, retriever *mockedRetriever, snap *mockedSnapshot, reverse bool) {
	method := "Iter"
	if reverse {
		method = "IterReverse"
	}

	require.True(t, len(retriever.GetInvokes()) <= 1)
	for _, invoke := range retriever.GetInvokes() {
		require.Equal(t, method, invoke.Method)
		err := invoke.Ret[1]
		if err == nil {
			require.NotNil(t, invoke.Ret[0])
			iter := invoke.Ret[0].(*mock.MockedIter)
			require.True(t, iter.Closed())
		}
	}

	require.True(t, len(snap.GetInvokes()) <= 1)
	for _, invoke := range snap.GetInvokes() {
		require.Equal(t, method, invoke.Method)
		err := invoke.Ret[1]
		if err == nil {
			require.NotNil(t, invoke.Ret[0])
			iter := invoke.Ret[0].(*mock.MockedIter)
			require.True(t, iter.Closed())
		}
	}
}

func TestIterTable(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)

	noTempTableData := []*kv.Entry{
		// normal table data
		{Key: encodeTableKey(1), Value: []byte("v1")},
		{Key: encodeTableKey(1, 1), Value: []byte("v11")},
		// no exist table data
		{Key: encodeTableKey(2), Value: []byte("v2")},
		{Key: encodeTableKey(2, 1), Value: []byte("v21")},
	}

	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}
	retriever := newMockedRetriever(t).SetData(localTempTableData).SetAllowedMethod("Iter")
	snap := newMockedSnapshot(newMockedRetriever(t).SetData(noTempTableData).SetAllowedMethod("Iter"))
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)

	cases := []struct {
		tblID      int64
		nilSession bool
		args       []kv.Key
		result     []kv.Entry
	}{
		{
			tblID: 1,
			args:  []kv.Key{encodeTableKey(1), encodeTableKey(2)},
			result: []kv.Entry{
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
			},
		},
		{
			tblID: 1,
			args:  []kv.Key{encodeTableKey(1), encodeTableKey(1, 1)},
			result: []kv.Entry{
				{Key: encodeTableKey(1), Value: []byte("v1")},
			},
		},
		{
			tblID: 2,
			args:  []kv.Key{encodeTableKey(2), encodeTableKey(3)},
			result: []kv.Entry{
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
			},
		},
		{
			tblID:  3,
			args:   []kv.Key{encodeTableKey(3), encodeTableKey(4)},
			result: []kv.Entry{},
		},
		{
			tblID:  4,
			args:   []kv.Key{encodeTableKey(4), encodeTableKey(5)},
			result: []kv.Entry{},
		},
		{
			tblID: 5,
			args:  []kv.Key{encodeTableKey(5), encodeTableKey(6)},
			result: []kv.Entry{
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
			},
		},
		{
			tblID: 5,
			args:  []kv.Key{encodeTableKey(5), encodeTableKey(5, 1)},
			result: []kv.Entry{
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
			},
		},
		{
			tblID:      5,
			nilSession: true,
			args:       []kv.Key{encodeTableKey(5), encodeTableKey(5, 1)},
			result:     []kv.Entry{},
		},
	}

	for i, c := range cases {
		inter := interceptor
		if c.nilSession {
			inter = emptyRetrieverInterceptor
		}
		iter, err := inter.iterTable(c.tblID, snap, c.args[0], c.args[1])
		require.NoError(t, err)
		result := make([]kv.Entry, 0, i)
		for iter.Valid() {
			result = append(result, kv.Entry{Key: iter.Key(), Value: iter.Value()})
			require.NoError(t, iter.Next(), i)
		}
		require.Equal(t, c.result, result, i)

		tbl, ok := is.TableByID(c.tblID)
		if !ok || tbl.Meta().TempTableType == model.TempTableNone {
			require.Equal(t, 0, len(retriever.GetInvokes()), i)
			require.Equal(t, 1, len(snap.GetInvokes()), i)
			require.Equal(t, "Iter", snap.GetInvokes()[0].Method)
			require.Equal(t, []interface{}{c.args[0], c.args[1]}, snap.GetInvokes()[0].Args, i)
		}

		if ok && tbl.Meta().TempTableType == model.TempTableGlobal {
			require.Equal(t, 0, len(retriever.GetInvokes()), i)
			require.Equal(t, 0, len(snap.GetInvokes()), i)
		}

		if ok && tbl.Meta().TempTableType == model.TempTableLocal {
			require.Equal(t, 0, len(snap.GetInvokes()), i)
			if c.nilSession {
				require.Equal(t, 0, len(retriever.GetInvokes()), i)
			} else {
				require.Equal(t, 1, len(retriever.GetInvokes()), i)
				require.Equal(t, "Iter", retriever.GetInvokes()[0].Method)
				require.Equal(t, []interface{}{c.args[0], c.args[1]}, retriever.GetInvokes()[0].Args, i)
			}
		}

		snap.ResetInvokes()
		retriever.ResetInvokes()
	}

	// test error for snap
	snapErr := errors.New("snapErr")
	snap.InjectMethodError("Iter", snapErr)
	iter, err := interceptor.iterTable(1, snap, encodeTableKey(1), encodeTableKey(2))
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)
	snap.InjectMethodError("Iter", nil)

	// test error for retriever
	retrieverErr := errors.New("retrieverErr")
	retriever.InjectMethodError("Iter", retrieverErr)
	iter, err = interceptor.iterTable(5, snap, encodeTableKey(5), encodeTableKey(6))
	require.Nil(t, iter)
	require.Equal(t, retrieverErr, err)
}

func TestOnIter(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)

	noTempTableData := []*kv.Entry{
		// normal table data
		{Key: encodeTableKey(1), Value: []byte("v1")},
		{Key: encodeTableKey(1, 1), Value: []byte("v11")},
		// no exist table data
		{Key: encodeTableKey(2), Value: []byte("v2")},
		{Key: encodeTableKey(2, 1), Value: []byte("v21")},
		// other data
		{Key: kv.Key("s"), Value: []byte("vs")},
		{Key: kv.Key("s0"), Value: []byte("vs0")},
		{Key: kv.Key("u"), Value: []byte("vu")},
		{Key: kv.Key("u0"), Value: []byte("vu0")},
		{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
		{Key: encodeTableKey(0), Value: []byte("v0")},
		{Key: encodeTableKey(0, 1), Value: []byte("v01")},
		{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
		{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
	}

	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}
	retriever := newMockedRetriever(t).SetData(localTempTableData).SetAllowedMethod("Iter")
	snap := newMockedSnapshot(newMockedRetriever(t).SetData(noTempTableData).SetAllowedMethod("Iter"))
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)

	cases := []struct {
		nilSession bool
		args       []kv.Key
		result     []kv.Entry
	}{
		{
			args: []kv.Key{nil, nil},
			result: []kv.Entry{
				{Key: kv.Key("s"), Value: []byte("vs")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
				{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
				{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
				{Key: kv.Key("u"), Value: []byte("vu")},
				{Key: kv.Key("u0"), Value: []byte("vu0")},
			},
		},
		{
			args: []kv.Key{nil, kv.Key("s0")},
			result: []kv.Entry{
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
		{
			args: []kv.Key{kv.Key("u"), nil},
			result: []kv.Entry{
				{Key: kv.Key("u"), Value: []byte("vu")},
				{Key: kv.Key("u0"), Value: []byte("vu0")},
			},
		},
		{
			args: []kv.Key{encodeTableKey(1), nil},
			result: []kv.Entry{
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
				{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
				{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
				{Key: kv.Key("u"), Value: []byte("vu")},
				{Key: kv.Key("u0"), Value: []byte("vu0")},
			},
		},
		{
			args: []kv.Key{nil, encodeTableKey(1, 1)},
			result: []kv.Entry{
				{Key: kv.Key("s"), Value: []byte("vs")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
			},
		},
		{
			args: []kv.Key{encodeTableKey(1), encodeTableKey(2)},
			result: []kv.Entry{
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
			},
		},
		{
			args: []kv.Key{encodeTableKey(1), encodeTableKey(3)},
			result: []kv.Entry{
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
			},
		},
		{
			args:   []kv.Key{encodeTableKey(3), encodeTableKey(4)},
			result: []kv.Entry{},
		},
		{
			args: []kv.Key{encodeTableKey(5), encodeTableKey(5, 3)},
			result: []kv.Entry{
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
			},
		},
		{
			args:       []kv.Key{nil, nil},
			nilSession: true,
			result: []kv.Entry{
				{Key: kv.Key("s"), Value: []byte("vs")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
				{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
				{Key: kv.Key("u"), Value: []byte("vu")},
				{Key: kv.Key("u0"), Value: []byte("vu0")},
			},
		},
		{
			args:       []kv.Key{encodeTableKey(5), encodeTableKey(5, 3)},
			nilSession: true,
			result:     []kv.Entry{},
		},
	}

	for i, c := range cases {
		inter := interceptor
		if c.nilSession {
			inter = emptyRetrieverInterceptor
		}

		iter, err := inter.OnIter(snap, c.args[0], c.args[1])
		require.NoError(t, err, i)
		require.NotNil(t, iter, i)
		result := make([]kv.Entry, 0, i)
		for iter.Valid() {
			result = append(result, kv.Entry{Key: iter.Key(), Value: iter.Value()})
			require.NoError(t, iter.Next(), i)
		}
		require.Equal(t, c.result, result, i)
	}

	// test error for snap
	snapErr := errors.New("snapErr")
	snap.InjectMethodError("Iter", snapErr)

	iter, err := interceptor.OnIter(snap, nil, nil)
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)

	iter, err = interceptor.OnIter(snap, nil, kv.Key("o"))
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)

	iter, err = interceptor.OnIter(snap, encodeTableKey(4), encodeTableKey(5))
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)

	snap.InjectMethodError("Iter", nil)

	// test error for retriever
	retrieverErr := errors.New("retrieverErr")
	retriever.InjectMethodError("Iter", retrieverErr)

	iter, err = interceptor.OnIter(snap, nil, nil)
	require.Nil(t, iter)
	require.Equal(t, retrieverErr, err)

	iter, err = interceptor.OnIter(snap, encodeTableKey(5), encodeTableKey(6))
	require.Nil(t, iter)
	require.Equal(t, retrieverErr, err)
}

func TestOnIterReverse(t *testing.T) {
	t.Parallel()
	is := newMockedInfoSchema(t).
		AddTable(model.TempTableNone, 1).
		AddTable(model.TempTableGlobal, 3).
		AddTable(model.TempTableLocal, 5)

	noTempTableData := []*kv.Entry{
		// normal table data
		{Key: encodeTableKey(1), Value: []byte("v1")},
		{Key: encodeTableKey(1, 1), Value: []byte("v11")},
		// no exist table data
		{Key: encodeTableKey(2), Value: []byte("v2")},
		{Key: encodeTableKey(2, 1), Value: []byte("v21")},
		// other data
		{Key: kv.Key("s"), Value: []byte("vs")},
		{Key: kv.Key("s0"), Value: []byte("vs0")},
		{Key: kv.Key("u"), Value: []byte("vu")},
		{Key: kv.Key("u0"), Value: []byte("vu0")},
		{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
		{Key: encodeTableKey(0), Value: []byte("v0")},
		{Key: encodeTableKey(0, 1), Value: []byte("v01")},
		{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
		{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
	}

	localTempTableData := []*kv.Entry{
		{Key: encodeTableKey(5), Value: []byte("v5")},
		{Key: encodeTableKey(5, 0), Value: []byte("v50")},
		{Key: encodeTableKey(5, 1), Value: []byte("v51")},
		{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
		{Key: encodeTableKey(5, 2), Value: []byte("")},
		{Key: encodeTableKey(5, 3), Value: nil},
	}
	retriever := newMockedRetriever(t).SetData(localTempTableData).SetAllowedMethod("IterReverse")
	snap := newMockedSnapshot(newMockedRetriever(t).SetData(noTempTableData).SetAllowedMethod("IterReverse"))
	interceptor := NewTemporaryTableSnapshotInterceptor(is, retriever)
	emptyRetrieverInterceptor := NewTemporaryTableSnapshotInterceptor(is, nil)

	cases := []struct {
		nilSession bool
		args       []kv.Key
		result     []kv.Entry
	}{
		{
			args: []kv.Key{nil},
			result: []kv.Entry{
				{Key: kv.Key("u0"), Value: []byte("vu0")},
				{Key: kv.Key("u"), Value: []byte("vu")},
				{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
				{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
		{
			args: []kv.Key{kv.Key("u")},
			result: []kv.Entry{
				{Key: encodeTableKey(math.MaxInt64, 1), Value: []byte("vm1")},
				{Key: encodeTableKey(math.MaxInt64), Value: []byte("vm")},
				{Key: encodeTableKey(5, 1), Value: []byte("v51")},
				{Key: encodeTableKey(5, 0, 1), Value: []byte("v501")},
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
		{
			args: []kv.Key{encodeTableKey(5, 0, 1)},
			result: []kv.Entry{
				{Key: encodeTableKey(5, 0), Value: []byte("v50")},
				{Key: encodeTableKey(5), Value: []byte("v5")},
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
		{
			args: []kv.Key{kv.Key("s0")},
			result: []kv.Entry{
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
		{
			args:       []kv.Key{encodeTableKey(5, 0, 1)},
			nilSession: true,
			result: []kv.Entry{
				{Key: encodeTableKey(2, 1), Value: []byte("v21")},
				{Key: encodeTableKey(2), Value: []byte("v2")},
				{Key: encodeTableKey(1, 1), Value: []byte("v11")},
				{Key: encodeTableKey(1), Value: []byte("v1")},
				{Key: encodeTableKey(0, 1), Value: []byte("v01")},
				{Key: encodeTableKey(0), Value: []byte("v0")},
				{Key: tablecodec.TablePrefix(), Value: []byte("vt")},
				{Key: kv.Key("s0"), Value: []byte("vs0")},
				{Key: kv.Key("s"), Value: []byte("vs")},
			},
		},
	}

	for i, c := range cases {
		inter := interceptor
		if c.nilSession {
			inter = emptyRetrieverInterceptor
		}

		iter, err := inter.OnIterReverse(snap, c.args[0])
		require.NoError(t, err, i)
		require.NotNil(t, iter, i)
		result := make([]kv.Entry, 0, i)
		for iter.Valid() {
			result = append(result, kv.Entry{Key: iter.Key(), Value: iter.Value()})
			require.NoError(t, iter.Next(), i)
		}
		require.Equal(t, c.result, result, i)
	}

	// test error for snap
	snapErr := errors.New("snapErr")
	snap.InjectMethodError("IterReverse", snapErr)

	iter, err := interceptor.OnIterReverse(snap, nil)
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)

	iter, err = interceptor.OnIterReverse(snap, kv.Key("o"))
	require.Nil(t, iter)
	require.Equal(t, snapErr, err)

	snap.InjectMethodError("IterReverse", nil)

	// test error for retriever
	retrieverErr := errors.New("retrieverErr")
	retriever.InjectMethodError("IterReverse", retrieverErr)

	iter, err = interceptor.OnIterReverse(snap, nil)
	require.Nil(t, iter)
	require.Equal(t, retrieverErr, err)
}


text:
/*
 * Copyright (c) 2008-2014 MongoDB, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.mongodb;

/**
 * Utility for constructing Query operation command with query, orderby, hint, explain, snapshot.
 */
class QueryOpBuilder {

    static final String READ_PREFERENCE_META_OPERATOR = "$readPreference";

    private DBObject query;
	private DBObject orderBy;
	private DBObject hintObj;
	private String hintStr;
	private boolean explain;
	private boolean snapshot;
    private long maxTimeMS;
	private ReadPreference readPref;

	private DBObject specialFields;

	public QueryOpBuilder(){
	}

	/**
	 * Adds the query clause to the operation
	 * @param query
	 * @return
	 */
	public QueryOpBuilder addQuery(DBObject query){
		this.query = query;
		return this;
	}

	/**
	 * Adds the orderby clause to the operation
	 * @param orderBy
	 * @return
	 */
	public QueryOpBuilder addOrderBy(DBObject orderBy){
		this.orderBy = orderBy;
		return this;
	}

	/**
	 * Adds the hint clause to the operation
	 * @param hint
	 * @return
	 */
	public QueryOpBuilder addHint(String hint){
		this.hintStr = hint;
		return this;
	}

	/**
	 * Adds hint clause to the operation
	 * @param hint
	 * @return
	 */
	public QueryOpBuilder addHint(DBObject hint){
		this.hintObj = hint;
		return this;
	}

	/**
	 * Adds special fields to the operation
	 * @param specialFields
	 * @return
	 */
	public QueryOpBuilder addSpecialFields(DBObject specialFields){
		this.specialFields = specialFields;
		return this;
	}

	/**
	 * Adds the explain flag to the operation
	 * @param explain
	 * @return
	 */
	public QueryOpBuilder addExplain(boolean explain){
		this.explain = explain;
		return this;
	}

	/**
	 * Adds the snapshot flag to the operation
	 * @param snapshot
	 * @return
	 */
	public QueryOpBuilder addSnapshot(boolean snapshot){
		this.snapshot = snapshot;
		return this;
	}

    /**
     * Adds a read preference to the query operation
     *
     * @param readPref the read preference
     * @return this
     */
    public QueryOpBuilder addReadPreference(ReadPreference readPref) {
        this.readPref = readPref;
        return this;
    }

    public QueryOpBuilder addMaxTimeMS(final long maxTimeMS) {
        this.maxTimeMS = maxTimeMS;
        return this;
    }

    /**
	 * Constructs the query operation DBObject
	 * @return DBObject representing the query command to be sent to server
	 */
        public DBObject get() {
            DBObject lclQuery = query;

            //must always have a query
            if (lclQuery == null) {
                lclQuery = new BasicDBObject();
            }

            if (hasSpecialQueryFields()) {
                DBObject queryop = (specialFields == null ? new BasicDBObject() : specialFields);

                addToQueryObject(queryop, "$query", lclQuery, true);
                addToQueryObject(queryop, "$orderby", orderBy, false);
                if (hintStr != null)
                    addToQueryObject(queryop, "$hint", hintStr);
                if (hintObj != null)
                    addToQueryObject(queryop, "$hint", hintObj);

                if (explain)
                    queryop.put("$explain", true);
                if (snapshot)
                    queryop.put("$snapshot", true);
                if (readPref != null && readPref != ReadPreference.primary())
                    queryop.put(READ_PREFERENCE_META_OPERATOR, readPref.toDBObject());
                if (maxTimeMS > 0) {
                    queryop.put("$maxTimeMS", maxTimeMS);
                }

                return queryop;
            }

            return lclQuery;
        }

    private boolean hasSpecialQueryFields(){

        if ( readPref != null )
            return true;

        if ( specialFields != null )
            return true;

        if ( orderBy != null && orderBy.keySet().size() > 0 )
            return true;

        if ( hintStr != null || hintObj != null || snapshot || explain)
            return true;

        if (maxTimeMS > 0) {
            return true;
        }

        return false;
    }

	/**
	 * Adds DBObject to the operation
	 * @param dbobj DBObject to add field to
	 * @param field name of the field
	 * @param obj object to add to the operation.  Ignore if null.
	 * @param sendEmpty if true adds obj even if it's empty.  Ignore if false and obj is empty.
	 * @return
	 */
	private void addToQueryObject(DBObject dbobj, String field, DBObject obj, boolean sendEmpty) {
		if (obj == null)
			return;

		if (!sendEmpty && obj.keySet().size() == 0)
			return;

		addToQueryObject(dbobj, field, obj);
	}

	/**
	 * Adds an Object to the operation
	 * @param dbobj DBObject to add field to
	 * @param field name of the field
	 * @param obj Object to be added.  Ignore if null
	 * @return
	 */
	private void addToQueryObject(DBObject dbobj, String field, Object obj) {

		if (obj == null)
			return;

		dbobj.put(field, obj);
	}

}

redpajama_set_name: RedPajamaGithub


text:
#ifndef _DEFS_H
#define _DEFS_H

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <math.h>
#include <sys/time.h>
#ifdef _OPENMP
#include <omp.h>
#endif
#include "sprng.h"

/* Uncomment this line, or use the flag -DDIAGNOSTIC for
   iverbose benchmark output */
/* #define DIAGNOSTIC */

/* Uncomment this line, or use the flag -DVERIFYK4 to
   generate a 2D torus */
/* #define VERIFYK4 */

#define INT_T int
#define DOUBLE_T double

#if defined(MASSIVE_GRAPH)
#define VERT_T long
#define LONG_T long
#elif defined(LARGE_GRAPH)
#define VERT_T int
#define LONG_T long
#else
#define VERT_T int
#define LONG_T int
#endif

#define WEIGHT_T VERT_T

#ifdef _OPENMP
    extern int NUM_THREADS;
#endif

/* Data structure for representing tuples
 * in the Scalable Data Generator */
typedef struct
{
    /* Edge lists */
    VERT_T* startVertex;
    VERT_T* endVertex;
    WEIGHT_T* weight;

    /* No. of edges */
    LONG_T m;

    /* No. of vertices */
    LONG_T n;

} graphSDG;

/* The graph data structure*/
typedef struct
{
    LONG_T n;
    LONG_T m;

    /* Directed edges out of vertex vi (say, k edges -- v1, v2, ... vk)
     * are stored in the contiguous block endV[numEdges[i] .. numEdges[i+1]]
     * So, numEdges[i+1] - numEdges[i] = k in this case */
    VERT_T* endV;
    LONG_T* numEdges;
    WEIGHT_T* weight;

} graph;

/* Edge data structure for Kernel 2 */
typedef struct
{
    VERT_T startVertex;
    VERT_T endVertex;
    WEIGHT_T w;
    LONG_T e;
} edge;

/* Predecessor list data structure for Kernel 4 */
typedef struct
{
    VERT_T* list;
    VERT_T count;
    LONG_T degree;
} plist;

/* Global variables */
extern INT_T SCALE;
extern LONG_T N;
extern LONG_T M;

/* R-MAT (graph generator) parameters */
extern DOUBLE_T A;
extern DOUBLE_T B;
extern DOUBLE_T C;
extern DOUBLE_T D;

extern WEIGHT_T MaxIntWeight;
extern INT_T SubGraphPathLength;
extern INT_T K4approx;

/* Function declarations */

void init(int);
double genScalData(graphSDG*);
double gen2DTorus(graphSDG*);

/* The four kernels */
double computeGraph(graph*, graphSDG*);
double getStartLists(graph*, edge**, INT_T*);
double findSubGraphs(graph*, edge*, INT_T);
double betweennessCentrality(graph*, DOUBLE_T *);

/* other useful routines */
void prefix_sums(LONG_T*, LONG_T*, LONG_T*, LONG_T);
DOUBLE_T get_seconds(void);
#endif


text:
/*
 * ProjectEuler/src/c/Problem003.c
 *
 * Largest prime factor
 * ====================
 * Published on Friday, 2nd November 2001, 06:00 pm
 *
 * The prime factors of 13195 are 5, 7, 13 and 29. What is the largest prime
 * factor of the number 600851475143 ?
 */

#include <stdio.h>
#include "ProjectEuler/ProjectEuler.h"
#include "ProjectEuler/Problem003.h"

int main(int argc, char** argv) {

    return 0;
}


text:
#include "config.h"
#include "ocinternal.h"
#include "occompile.h"
#include "ocdebug.h"

static const unsigned int MAX_UINT = 0xffffffff;

static OCerror mergedas1(OCnode* dds, OCnode* das);
static OCerror mergedods1(OCnode* dds, OCnode* das);
static OCerror mergeother1(OCnode* root, OCnode* das);
static char* pathtostring(OClist* path, char* separator);
static void computefullname(OCnode* node);

/* Process ocnodes to fix various semantic issues*/
void
occomputesemantics(OClist* ocnodes)
{
    unsigned int i,j;
    OCASSERT((ocnodes != NULL));
    for(i=0;i<oclistlength(ocnodes);i++) {
	OCnode* node = (OCnode*)oclistget(ocnodes,i);
	/* set the container for dims*/
	if(node->octype == OC_Dimension && node->dim.array != NULL) {
	    node->container = node->dim.array->container;
	}
    }
    /* Fill in array.sizes */
    for(i=0;i<oclistlength(ocnodes);i++) {
	OCnode* node = (OCnode*)oclistget(ocnodes,i);
	if(node->array.rank > 0) {
	    node->array.sizes = (size_t*)malloc(node->array.rank*sizeof(size_t));
	    for(j=0;j<node->array.rank;j++) {
		OCnode* dim = (OCnode*)oclistget(node->array.dimensions,j);
		node->array.sizes[j] = dim->dim.declsize;
	    }
	}
    }
}

void
occomputefullnames(OCnode* root)
{
    unsigned int i;
    if(root->name != NULL) computefullname(root);
    if(root->subnodes != NULL) { /* recurse*/
        for(i=0;i<oclistlength(root->subnodes);i++) {
	    OCnode* node = (OCnode*)oclistget(root->subnodes,i);
	    occomputefullnames(node);
	}
    }
}

static void
computefullname(OCnode* node)
{
    char* tmp;
    char* fullname;
    OClist* path;

    OCASSERT((node->name != NULL));
    if(node->fullname != NULL)
	return;
    path = oclistnew();
    occollectpathtonode(node,path);
    tmp = pathtostring(path,PATHSEPARATOR);
    if(tmp == NULL) {
        fullname = nulldup(node->name);
    } else {
        fullname = tmp;
    }
    node->fullname = fullname;
    oclistfree(path);
}

/* Convert path to a string */
static char*
pathtostring(OClist* path, char* separator)
{
    int slen,i,len;
    char* pathname;
    if(path == NULL) return NULL;
    len = oclistlength(path);
    if(len == 0) return NULL;
    for(i=0,slen=0;i<len;i++) {
	OCnode* node = (OCnode*)oclistget(path,(size_t)i);
	if(node->container == NULL || node->name == NULL) continue;
	slen += strlen(node->name);
    }
    slen += ((len-1)*strlen(separator));
    slen += 1;   /* for null terminator*/
    pathname = (char*)ocmalloc((size_t)slen);
    MEMCHECK(pathname,NULL);
    pathname[0] = '\0';
    for(i=0;i<len;i++) {
	OCnode* node = (OCnode*)oclistget(path,(size_t)i);
	if(node->container == NULL || node->name == NULL) continue;
	if(strlen(pathname) > 0) strcat(pathname,separator);
        strcat(pathname,node->name);
    }
    return pathname;
}

/* Collect the set of nodes ending in "node"*/
void
occollectpathtonode(OCnode* node, OClist* path)
{
    if(node == NULL) return;
    occollectpathtonode(node->container,path);
    oclistpush(path,(void*)node);
}

OCnode*
ocnode_new(char* name, OCtype ptype, OCnode* root)
{
    OCnode* cdf = (OCnode*)ocmalloc(sizeof(OCnode));
    MEMCHECK(cdf,(OCnode*)NULL);
    memset((void*)cdf,0,sizeof(OCnode));
    cdf->header.magic = OCMAGIC;
    cdf->header.occlass = OC_Node;
    cdf->name = (name?nulldup(name):NULL);
    cdf->octype = ptype;
    cdf->array.dimensions = NULL;
    cdf->root = root;
    return cdf;
}

static OCattribute*
makeattribute(char* name, OCtype ptype, OClist* values)
{
    OCattribute* att = (OCattribute*)ocmalloc(sizeof(OCattribute)); /* ocmalloc zeros*/
    MEMCHECK(att,(OCattribute*)NULL);
    att->name = nulldup(name);
    att->etype = ptype;
    att->nvalues = oclistlength(values);
    att->values = NULL;
    if(att->nvalues > 0) {
	int i;
        att->values = (char**)ocmalloc(sizeof(char*)*att->nvalues);
        for(i=0;i<att->nvalues;i++)
	    att->values[i] = nulldup((char*)oclistget(values,(size_t)i));
    }
    return att;
}

void
ocroot_free(OCnode* root)
{
    OCtree* tree;
    OCstate* state;
    int i;

    if(root == NULL || root->tree == NULL) return;

    tree = root->tree;
    state = tree->state;

    /* Free up the OCDATA instance, if any */
    if(tree->data.data != NULL)
	ocdata_free(state,tree->data.data);

    for(i=0;i<oclistlength(state->trees);i++) {
	OCnode* node = (OCnode*)oclistget(state->trees,(size_t)i);
	if(root == node)
	    oclistremove(state->trees,(size_t)i);
    }
    /* Note: it is ok if state->trees does not contain this root */
    octree_free(tree);
}

void
octree_free(OCtree* tree)
{
    if(tree == NULL) return;
    ocnodes_free(tree->nodes);
    ocfree(tree->constraint);
    ocfree(tree->text);
    if(tree->data.xdrs != NULL) {
        xxdr_free(tree->data.xdrs);
    }
    ocfree(tree->data.filename); /* may be null */
    if(tree->data.file != NULL) fclose(tree->data.file);
    ocfree(tree->data.memory);
    ocfree(tree);
}

void
ocnodes_free(OClist* nodes)
{
    unsigned int i,j;
    for(i=0;i<oclistlength(nodes);i++) {
	OCnode* node = (OCnode*)oclistget(nodes,i);
        ocfree(node->name);
        ocfree(node->fullname);
        while(oclistlength(node->att.values) > 0) {
	    char* value = (char*)oclistpop(node->att.values);
	    ocfree(value);
        }
        while(oclistlength(node->attributes) > 0) {
            OCattribute* attr = (OCattribute*)oclistpop(node->attributes);
	    ocfree(attr->name);
#if 0
	    /* If the attribute type is string, then we need to free them*/
all values are strings now
	    if(attr->etype == OC_String || attr->etype == OC_URL)
#endif
	    {
		char** strings = (char**)attr->values;
		for(j=0;j<attr->nvalues;j++) {ocfree(*strings); strings++;}
	    }
	    ocfree(attr->values);
	    ocfree(attr);
        }
        if(node->array.dimensions != NULL) oclistfree(node->array.dimensions);
        if(node->subnodes != NULL) oclistfree(node->subnodes);
        if(node->att.values != NULL) oclistfree(node->att.values);
        if(node->attributes != NULL) oclistfree(node->attributes);
	if(node->array.sizes != NULL) free(node->array.sizes);
        ocfree(node);
    }
    oclistfree(nodes);
}

/*
In order to be as compatible as possible with libdap,
we try to use the same algorithm for DAS->DDS matching.
As described there, the algorithm is as follows.
    If the [attribute] name contains one or
    more field separators then look for a [DDS]variable whose
    name matches exactly. If the name contains no field separators then
    the look first in the top level [of the DDS] and then in all subsequent
    levels and return the first occurrence found. In general, this
    searches constructor types in the order in which they appear
    in the DDS, but there is no requirement that it do so.

    Note: If a dataset contains two constructor types which have field names
    that are the same (say point.x and pair.x) one should use fully qualified
    names to get each of those variables.

*/

OCerror
ocddsdasmerge(OCstate* state, OCnode* dasroot, OCnode* ddsroot)
{
    OCerror stat = OC_NOERR;
    OClist* dasglobals = oclistnew();
    OClist* dodsglobals = oclistnew(); /* top-level DODS_XXX {...} */
    OClist* dasnodes = oclistnew();
    OClist* varnodes = oclistnew();
    OClist* ddsnodes;
    unsigned int i,j;

    if(dasroot->tree == NULL || dasroot->tree->dxdclass != OCDAS)
	{stat = OCTHROW(OC_EINVAL); goto done;}
    if(ddsroot->tree == NULL || (ddsroot->tree->dxdclass != OCDDS
        && ddsroot->tree->dxdclass != OCDATADDS))
	{stat = OCTHROW(OC_EINVAL); goto done;}

    ddsnodes = ddsroot->tree->nodes;

    /* 1. collect all the relevant DAS nodes;
          namely those that contain at least one
          attribute value, not including the leaf attributes.
          Simultaneously look for potential ambiguities
          if found; complain but continue: result are indeterminate.
          also collect globals separately*/
    for(i=0;i<oclistlength(dasroot->tree->nodes);i++) {
	OCnode* das = (OCnode*)oclistget(dasroot->tree->nodes,i);
	int hasattributes = 0;
	if(das->octype == OC_Attribute) continue; /* ignore these for now*/
	if(das->name == NULL || das->att.isglobal) {
	    oclistpush(dasglobals,(void*)das);
	    continue;
	}
	if(das->att.isdods) {
	    oclistpush(dodsglobals,(void*)das);
	    continue;
	}
	for(j=0;j<oclistlength(das->subnodes);j++) {
	    OCnode* subnode = (OCnode*)oclistget(das->subnodes,j);
	    if(subnode->octype == OC_Attribute) {hasattributes = 1; break;}
	}
	if(hasattributes) {
	    /* Look for previously collected nodes with same name*/
            for(j=0;j<oclistlength(dasnodes);j++) {
	        OCnode* das2 = (OCnode*)oclistget(dasnodes,j);
		if(das->name == NULL || das2->name == NULL) continue;
		if(strcmp(das->name,das2->name)==0) {
		    oclog(OCLOGWARN,"oc_mergedas: potentially ambiguous DAS name: %s",das->name);
		}
	    }
	    oclistpush(dasnodes,(void*)das);
	}
    }

    /* 2. collect all the leaf DDS nodes (of type OC_Atomic)*/
    for(i=0;i<oclistlength(ddsnodes);i++) {
	OCnode* dds = (OCnode*)oclistget(ddsnodes,i);
	if(dds->octype == OC_Atomic) oclistpush(varnodes,(void*)dds);
    }

    /* 3. For each das node, locate matching DDS node(s) and attach
          attributes to the DDS node(s).
          Match means:
          1. DAS->fullname :: DDS->fullname
          2. DAS->name :: DDS->fullname (support DAS names with embedded '.')
          3. DAS->name :: DDS->name
    */
    for(i=0;i<oclistlength(dasnodes);i++) {
	OCnode* das = (OCnode*)oclistget(dasnodes,i);
        for(j=0;j<oclistlength(varnodes);j++) {
	    OCnode* dds = (OCnode*)oclistget(varnodes,j);
	    if(strcmp(das->fullname,dds->fullname)==0
	       || strcmp(das->name,dds->fullname)==0
	       || strcmp(das->name,dds->name)==0) {
		mergedas1(dds,das);
		/* remove from dasnodes list*/
		oclistset(dasnodes,i,(void*)NULL);
	    }
	}
    }

    /* 4. Assign globals*/
    for(i=0;i<oclistlength(dasglobals);i++) {
	OCnode* das = (OCnode*)oclistget(dasglobals,i);
	if(das == NULL) continue;
	mergedas1(ddsroot,das);
    }
    /* 5. Assign DODS_*/
    for(i=0;i<oclistlength(dodsglobals);i++) {
	OCnode* das = (OCnode*)oclistget(dodsglobals,i);
	if(das == NULL) continue;
	mergedods1(ddsroot,das);
    }
    /* 6. Assign other orphan attributes, which means
	  construct their full name and assign as a global attribute. */
    for(i=0;i<oclistlength(dasnodes);i++) {
	OCnode* das = (OCnode*)oclistget(dasnodes,i);
	if(das == NULL) continue;
	mergeother1(ddsroot, das);
    }

done:
    /* cleanup*/
    oclistfree(dasglobals);
    oclistfree(dodsglobals);
    oclistfree(dasnodes);
    oclistfree(varnodes);
    return OCTHROW(stat);
}

static OCerror
mergedas1(OCnode* dds, OCnode* das)
{
    unsigned int i;
    OCerror stat = OC_NOERR;
    if(das == NULL) return OC_NOERR; /* nothing to do */
    if(dds->attributes == NULL) dds->attributes = oclistnew();
    /* assign the simple attributes in the das set to this dds node*/
    for(i=0;i<oclistlength(das->subnodes);i++) {
	OCnode* attnode = (OCnode*)oclistget(das->subnodes,i);
	if(attnode->octype == OC_Attribute) {
	    OCattribute* att = makeattribute(attnode->name,
						attnode->etype,
						attnode->att.values);
            oclistpush(dds->attributes,(void*)att);
	}
    }
    return OCTHROW(stat);
}

static OCerror
mergedods1(OCnode* dds, OCnode* dods)
{
    unsigned int i;
    OCerror stat = OC_NOERR;
    if(dods == NULL) return OC_NOERR; /* nothing to do */
    OCASSERT(dods->octype == OC_Attributeset);
    if(dds->attributes == NULL) dds->attributes = oclistnew();
    /* assign the simple attributes in the das set to this dds node
       with renaming to tag as DODS_
    */
    for(i=0;i<oclistlength(dods->subnodes);i++) {
	OCnode* attnode = (OCnode*)oclistget(dods->subnodes,i);
	if(attnode->octype == OC_Attribute) {
	    OCattribute* att;
	    /* prefix the attribute name with the name of the attribute
               set plus "."
            */
	    size_t len =   strlen(attnode->name)
                         + strlen(dods->name)
			 + strlen(".")
			 + 1; /*null*/
	    char* newname = (char*)malloc(len);
	    if(newname == NULL) return OC_ENOMEM;
	    strcpy(newname,dods->name);
	    strcat(newname,".");
	    strcat(newname,attnode->name);
	    att = makeattribute(newname,attnode->etype,attnode->att.values);
	    free(newname);
            oclistpush(dds->attributes,(void*)att);
	}
    }
    return OCTHROW(stat);
}

static OCerror
mergeother1(OCnode* root, OCnode* das)
{
    OCerror stat = OC_NOERR;
    OCattribute* att = NULL;

    OCASSERT(root != NULL);
    if(root->attributes == NULL) root->attributes = oclistnew();

    if(das->octype == OC_Attribute) {
        /* compute the full name of this attribute */
        computefullname(das);
        /* create attribute */
        att = makeattribute(das->fullname,das->etype,das->att.values);
        oclistpush(root->attributes,(void*)att);
    } else if(das->octype == OC_Attributeset) {
	int i;
	/* Recurse */
        for(i=0;i<oclistlength(das->subnodes);i++) {
	    OCnode* sub = (OCnode*)oclistget(das->subnodes,i);
	    if(sub == NULL) continue;
	    mergeother1(root,sub);
	}
    } else
	stat = OC_EDAS;
    return OCTHROW(stat);
}

static void
ocuncorrelate(OCnode* root)
{
    OCtree* tree = root->tree;
    unsigned int i;
    if(tree == NULL) return;
    for(i=0;i<oclistlength(tree->nodes);i++) {
	OCnode* node = (OCnode*)oclistget(tree->nodes,i);
	node->datadds = NULL;
    }
}

static OCerror
occorrelater(OCnode* dds, OCnode* dxd)
{
    int i,j;
    OCerror ocstat = OC_NOERR;

    if(dds->octype != dxd->octype) {
	OCTHROWCHK((ocstat = OC_EINVAL)); goto fail;
    }
    if(dxd->name != NULL && dxd->name != NULL
       && strcmp(dxd->name,dds->name) != 0) {
	OCTHROWCHK((ocstat = OC_EINVAL)); goto fail;
    } else if(dxd->name != dxd->name) { /* test NULL==NULL */
	OCTHROWCHK((ocstat = OC_EINVAL)); goto fail;
    }

    if(dxd->array.rank != dds->array.rank) {
	OCTHROWCHK((ocstat = OC_EINVAL)); goto fail;
    }

    dds->datadds = dxd;

    switch (dds->octype) {
    case OC_Dataset:
    case OC_Structure:
    case OC_Grid:
    case OC_Sequence:
	/* Remember: there may be fewer datadds fields than dds fields */
	for(i=0;i<oclistlength(dxd->subnodes);i++) {
	    OCnode* dxd1 = (OCnode*)oclistget(dxd->subnodes,(size_t)i);
	    for(j=0;j<oclistlength(dds->subnodes);j++) {
		OCnode* dds1 = (OCnode*)oclistget(dds->subnodes,(size_t)j);
		if(strcmp(dxd1->name,dds1->name) == 0) {
		    ocstat = occorrelater(dds1,dxd1);
		    if(ocstat != OC_NOERR) {OCTHROWCHK(ocstat); goto fail;}
		    break;
		}
	    }
	}
	break;
    case OC_Dimension:
    case OC_Atomic:
	break;
    default: OCPANIC1("unexpected node type: %d",dds->octype);
    }
    /* Correlate the dimensions */
    if(dds->array.rank > 0) {
	for(i=0;i<oclistlength(dxd->subnodes);i++) {
	    OCnode* ddsdim = (OCnode*)oclistget(dds->array.dimensions,(size_t)i);
	    OCnode* dxddim = (OCnode*)oclistget(dxd->array.dimensions,(size_t)i);
	    ocstat = occorrelater(ddsdim,dxddim);
	    if(!ocstat) goto fail;
	}
    }

fail:
    return OCTHROW(ocstat);

}

OCerror
occorrelate(OCnode* dds, OCnode* dxd)
{
    if(dds == NULL || dxd == NULL) return OCTHROW(OC_EINVAL);
    ocuncorrelate(dds);
    return occorrelater(dds,dxd);
}

/*
Mark cacheable those atomic String/URL typed nodes
that are contained only in structures with rank > 0.
*/
void
ocmarkcacheable(OCstate* state, OCnode* ddsroot)
{
    int i,j;
#if 0
    int ok;
#endif
    OClist* treenodes = ddsroot->tree->nodes;
    OClist* path = oclistnew();
    for(i=0;i<oclistlength(treenodes);i++) {
        OCnode* node = (OCnode*)oclistget(treenodes,(size_t)i);
	if(node->octype != OC_Atomic) continue;
	if(node->etype != OC_String && node->etype != OC_URL) continue;
	/* collect node path */
        oclistclear(path);
        occollectpathtonode(node,path);
#if 0
        ok = 1;
#endif
	for(j=1;j<oclistlength(path)-1;j++) {/* skip top level dataset and node itself*/
            OCnode* pathnode = (OCnode*)oclistget(path,(size_t)j);
	    if(pathnode->octype != OC_Structure
		|| pathnode->array.rank > 0) {
#if 0
	    ok=0;
#endif
	    break;
	    }
	}
#if 0
	if(ok) {
   	    node->cache.cacheable = 1;
	    node->cache.valid = 0;
	}
#endif
    }
    oclistfree(path);
}

#if 0

OCerror
ocddsdasmerge(OCstate* state, OCnode* ddsroot, OCnode* dasroot)
{
    int i,j;
    OCerror stat = OC_NOERR;
    OClist* globals = oclistnew();
    if(dasroot == NULL) return OCTHROW(stat);
    /* Start by looking for global attributes*/
    for(i=0;i<oclistlength(dasroot->subnodes);i++) {
	OCnode* node = (OCnode*)oclistget(dasroot->subnodes,i);
	if(node->att.isglobal) {
	    for(j=0;j<oclistlength(node->subnodes);j++) {
		OCnode* attnode = (OCnode*)oclistget(node->subnodes,j);
		Attribute* att = makeattribute(attnode->name,
						attnode->etype,
						attnode->att.values);
		oclistpush(globals,(void*)att);
	    }
	}
    }
    ddsroot->attributes = globals;
    /* Now try to match subnode names with attribute set names*/
    for(i=0;i<oclistlength(dasroot->subnodes);i++) {
	OCnode* das = (OCnode*)oclistget(dasroot->subnodes,i);
	int match = 0;
        if(das->att.isglobal) continue;
        if(das->octype == OC_Attributeset) {
            for(j=0;j<oclistlength(ddsroot->subnodes) && !match;j++) {
	        OCnode* dds = (OCnode*)oclistget(ddsroot->subnodes,j);
	        if(strcmp(das->name,dds->name) == 0) {
		    match = 1;
	            stat = mergedas1(dds,das);
	            if(stat != OC_NOERR) break;
		}
	    }
	}
        if(!match) {marklostattribute(das);}
    }
    if(stat == OC_NOERR) ddsroot->attributed = 1;
    return OCTHROW(stat);
}

/* Merge das attributes into the dds node*/

static int
mergedas1(OCnode* dds, OCnode* das)
{
    int i,j;
    int stat = OC_NOERR;
    if(dds->attributes == NULL) dds->attributes = oclistnew();
    /* assign the simple attributes in the das set to this dds node*/
    for(i=0;i<oclistlength(das->subnodes);i++) {
	OCnode* attnode = (OCnode*)oclistget(das->subnodes,i);
	if(attnode->octype == OC_Attribute) {
	    Attribute* att = makeattribute(attnode->name,
						attnode->etype,
						attnode->att.values);
            oclistpush(dds->attributes,(void*)att);
	}
    }
    /* Try to merge any enclosed sets with subnodes of dds*/
    for(i=0;i<oclistlength(das->subnodes);i++) {
	OCnode* dasnode = (OCnode*)oclistget(das->subnodes,i);
	int match = 0;
        if(dasnode->octype == OC_Attribute) continue; /* already dealt with above*/
        for(j=0;j<oclistlength(dds->subnodes) && !match;j++) {
	    OCnode* ddsnode = (OCnode*)oclistget(dds->subnodes,j);
	    if(strcmp(dasnode->name,ddsnode->name) == 0) {
	        match = 1;
	        stat = mergedas1(ddsnode,dasnode);
	        if(stat != OC_NOERR) break;
	    }
	}
        if(!match) {marklostattribute(dasnode);}
    }
    return OCTHROW(stat);
}

void*
oclinearize(OCtype etype, unsigned int nstrings, char** strings)
{
    int i;
    size_t typesize;
    char* memp;
    char* memory;

    if(nstrings == 0) return NULL;
    typesize = octypesize(etype);
    memory = (char*)ocmalloc(nstrings*typesize);
    MEMCHECK(memory,NULL);
    memp = memory;
    for(i=0;i<nstrings;i++) {
	char* value = strings[i];
        converttype(etype,value,memp);
	memp += typesize;
    }
    return memory;
}

static int
converttype(OCtype etype, char* value, char* memory)
{
    long iv;
    unsigned long uiv;
    double dv;
    char c[1];
    int outofrange = 0;
#ifdef HAVE_LONG_LONG_INT
    long long llv;
    unsigned long long ullv;
#endif

    switch (etype) {
    case OC_Char:
	if(sscanf(value,"%c",c) != 1) goto fail;
	*((char*)memory) = c[0];
	break;
    case OC_Byte:
	if(sscanf(value,"%ld",&iv) != 1) goto fail;
        else if(iv > OC_BYTE_MAX || iv < OC_BYTE_MIN) {iv = OC_BYTE_MAX; outofrange = 1;}
	*((signed char*)memory) = (signed char)iv;
	break;
    case OC_UByte:
	if(sscanf(value,"%lu",&uiv) != 1) goto fail;
        else if(uiv > OC_UBYTE_MAX) {uiv = OC_UBYTE_MAX; outofrange = 1;}
	*((unsigned char*)memory) = (unsigned char)uiv;
	break;
    case OC_Int16:
	if(sscanf(value,"%ld",&iv) != 1) goto fail;
        else if(iv > OC_INT16_MAX || iv < OC_INT16_MIN) {iv = OC_INT16_MAX; outofrange = 1;}
	*((signed short*)memory) = (signed short)iv;
	break;
    case OC_UInt16:
	if(sscanf(value,"%lu",&uiv) != 1) goto fail;
        else if(uiv > OC_UINT16_MAX) {uiv = OC_UINT16_MAX; outofrange = 1;}
	*((unsigned short*)memory) = (unsigned short)uiv;
	break;
    case OC_Int32:
	if(sscanf(value,"%ld",&iv) != 1) goto fail;
        else if(iv > OC_INT32_MAX || iv < OC_INT32_MIN) {iv = OC_INT32_MAX; outofrange = 1;}
	*((signed int*)memory) = (signed int)iv;
	break;
    case OC_UInt32:
	if(sscanf(value,"%lu",&uiv) != 1) goto fail;
        else if(uiv > OC_UINT32_MAX) {uiv = OC_UINT32_MAX; outofrange = 1;}
	*((unsigned char*)memory) = (unsigned int)uiv;
	break;
#ifdef HAVE_LONG_LONG_INT
    case OC_Int64:
	if(sscanf(value,"%lld",&llv) != 1) goto fail;
        /*else if(iv > OC_INT64_MAX || iv < OC_INT64_MIN) goto fail;*/
	*((signed long long*)memory) = (signed long long)llv;
	break;
    case OC_UInt64:
	if(sscanf(value,"%llu",&ullv) != 1) goto fail;
	*((unsigned long long*)memory) = (unsigned long long)ullv;
	break;
#endif
    case OC_Float32:
	if(sscanf(value,"%lf",&dv) != 1) goto fail;
	*((float*)memory) = (float)dv;
	break;
    case OC_Float64:
	if(sscanf(value,"%lf",&dv) != 1) goto fail;
	*((double*)memory) = (double)dv;
	break;
    case OC_String: case OC_URL:
	*((char**)memory) = nulldup(value);
	break;
    default:
	goto fail;
    }
    if(outofrange)
        oc_log(LOGWARN,"converttype range failure: %d: %s",etype,value);
    return 1;
fail:
    oc_log(LOGERR,"converttype bad value: %d: %s",etype,value);
    return 0;
}
#endif /*0*/


text:
set -o errexit

# Pull in shared configuration specific to this repo
base=$(dirname "$0")
source "${base}/config.bash"

# Verify ENV is set up correctly
# We validate all that need to be set in case, in an absolute emergency,
# we need to run this by hand. Otherwise the GitHub actions environment should
# provide all of these if properly configured
if [[ -z "${RELEASE_TOKEN}" ]]; then
  echo -e "\e[31mA personal access token needs to be set in RELEASE_TOKEN."
  echo -e "\e[31mIt should not be secrets.GITHUB_TOKEN. It has to be a"
  echo -e "\e[31mpersonal access token otherwise next steps in the release"
  echo -e "\e[31mprocess WILL NOT trigger."
  echo -e "\e[31mPersonal access tokens are in the form:"
  echo -e "\e[31m     TOKEN"
  echo -e "\e[31mfor example:"
  echo -e "\e[31m     1234567890"
  echo -e "\e[31mExiting.\e[0m"
  exit 1
fi

if [[ -z "${GITHUB_REF}" ]]; then
  echo -e "\e[31mThe release tag needs to be set in GITHUB_REF."
  echo -e "\e[31mThe tag should be in the following GitHub specific form:"
  echo -e "\e[31m    /refs/tags/release-X.Y.Z"
  echo -e "\e[31mwhere X.Y.Z is the version we are releasing"
  echo -e "\e[31mExiting.\e[0m"
  exit 1
fi

if [[ -z "${GITHUB_REPOSITORY}" ]]; then
  echo -e "\e[31mName of this repository needs to be set in GITHUB_REPOSITORY."
  echo -e "\e[31mShould be in the form OWNER/REPO, for example:"
  echo -e "\e[31m     ponylang/ponyup"
  echo -e "\e[31mExiting.\e[0m"
  exit 1
fi

# no unset variables allowed from here on out
# allow above so we can display nice error messages for expected unset variables
set -o nounset

# Set up GitHub credentials
git config --global user.name 'Ponylang Main Bot'
git config --global user.email 'ponylang.main@gmail.com'
git config --global push.default simple

PUSH_TO="https://${RELEASE_TOKEN}@github.com/${GITHUB_REPOSITORY}.git"

# Extract version from tag reference
# Tag ref version: "refs/tags/release-1.0.0"
# Version: "1.0.0"
VERSION="${GITHUB_REF/refs\/tags\/release-/}"

### this doesn't account for master changing commit, assumes we are HEAD
# or can otherwise push without issue. that shouldl error out without issue.
# leaving us to restart from a different HEAD commit
git checkout master
git pull

# version the changelog
echo -e "\e[34mUpdating CHANGELOG.md for release\e[0m"
changelog-tool release "${VERSION}" -e

# commit CHANGELOG update
echo -e "\e[34mCommiting CHANGELOG.md changes\e[0m"
git add CHANGELOG.md
git commit -m "${VERSION} release"

# tag release
echo -e "\e[34mTagging for release to kick off building artifacts\e[0m"
git tag "${VERSION}"

# push to release to remote
echo -e "\e[34mPushing commited changes back to master\e[0m"
git push ${PUSH_TO} master
echo -e "\e[34mPushing ${VERSION} tag\e[0m"
git push ${PUSH_TO} "${VERSION}"

# pull again, just in case, odds of this being needed are really slim
git pull

# update CHANGELOG for new entries
echo -e "\e[34mAdding new 'unreleased' section to CHANGELOG.md\e[0m"
changelog-tool unreleased -e

# commit changelog and push to master
echo -e "\e[34mCommiting CHANGELOG.md change\e[0m"
git add CHANGELOG.md
git commit -m "Add unreleased section to CHANGELOG post ${VERSION} release [skip ci]"

echo -e "\e[34mPushing CHANGELOG.md\e[0m"
git push ${PUSH_TO} master

# delete release-VERSION tag
echo -e "\e[34mDeleting no longer needed remote tag release-${VERSION}\e[0m"
git push --delete ${PUSH_TO} "release-${VERSION}"