Function TestRecursiveSplits

posting/list_test.go:1746–1793

(t *testing.T)

}

func TestRecursiveSplits(t *testing.T) {
	// For testing, set the max list size to a lower threshold.
	defer setMaxListSize(maxListSize)
	maxListSize = mb / 2

	// Create a list that should be split recursively.
	size := int(1e5)
	key := x.DataKey(x.AttrInRootNamespace(uuid.New().String()), 1331)
	ol, err := readPostingListFromDisk(key, ps, math.MaxUint64)
	require.NoError(t, err)
	commits := 0
	for i := 1; i <= size; i++ {
		commits++
		edge := &pb.DirectedEdge{
			ValueId: uint64(i),
		}
		edge.Facets = []*api.Facet{{Key: strconv.Itoa(i)}}

		txn := Txn{StartTs: uint64(i)}
		addMutationHelper(t, ol, edge, Set, &txn)
		require.NoError(t, ol.commitMutation(uint64(i), uint64(i)+1))

		// Do not roll-up the list here to ensure the final list should
		// be split more than once.
	}

	// Rollup the list. The final output should have more than two parts.
	kvs, err := ol.Rollup(nil, math.MaxUint64)
	require.NoError(t, err)
	require.NoError(t, writePostingListToDisk(kvs))
	ol, err = readPostingListFromDisk(key, ps, math.MaxUint64)
	require.NoError(t, err)
	require.True(t, len(ol.plist.Splits) > 2)

	// Read back the list and verify the data is correct.
	var facets []string
	err = ol.Iterate(uint64(size)+1, 0, func(p *pb.Posting) error {
		if len(p.Facets) > 0 {
			facets = append(facets, p.Facets[0].Key)
		}
		return nil
	})
	require.NoError(t, err)
	require.Equal(t, commits, len(facets))
	for i, facet := range facets {
		require.Equal(t, facet, strconv.Itoa(i+1))
	}
}

var ps *badger.DB