Skip to content

Use one atomic variable to generate blank node ids for json objects #3795

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Aug 14, 2019
Merged

Use one atomic variable to generate blank node ids for json objects #3795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
46ece60
Use one atomic variable to generate blank node ids for json objects
Aug 13, 2019
9aed88f
Fixed typo
Aug 13, 2019
e237ac0
Fixing tests by inserting into dgraph and then verify the query response
Aug 13, 2019
d4187e8
Fixing tests and removed the collision checking
Aug 13, 2019
8bdb85b
Fixing Martin's comments
Aug 14, 2019
7d395cf
Merge branch 'master' into gitlw/atomic_json_idx
Aug 14, 2019
aea702b
Fixing 100 chars
Aug 14, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 26 additions & 8 deletions chunker/json_parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,10 @@ import (
"bytes"
"encoding/json"
"fmt"
"math/rand"
"strconv"
"strings"
"sync/atomic"
"unicode"

"github.com/dgraph-io/dgo/protos/api"
Expand Down Expand Up @@ -267,8 +269,26 @@ func (buf *NQuadBuffer) Flush() {
close(buf.nqCh)
}

// nextIdx is the index that is used to generate blank node ids for a json map object
// when the map object does not have a "uid" field.
// It should only be accessed through the atomic APIs.
var nextIdx uint64

// randomID will be used to generate blank node ids.
// We use a random number to avoid collision with user specified uids.
var randomID uint32

func init() {
randomID = rand.Uint32()
}

func getNextBlank() string {
id := atomic.AddUint64(&nextIdx, 1)
return fmt.Sprintf("_:dg.%d.%d", randomID, id)
}

// TODO - Abstract these parameters to a struct.
func (buf *NQuadBuffer) mapToNquads(m map[string]interface{}, idx *int, op int, parentPred string) (
func (buf *NQuadBuffer) mapToNquads(m map[string]interface{}, op int, parentPred string) (
mapResponse, error) {
var mr mapResponse
// Check field in map.
Expand Down Expand Up @@ -308,8 +328,7 @@ func (buf *NQuadBuffer) mapToNquads(m map[string]interface{}, idx *int, op int,
return mr, errors.Errorf("UID must be present and non-zero while deleting edges.")
}

mr.uid = fmt.Sprintf("_:blank-%d", *idx)
*idx++
mr.uid = getNextBlank()
}

for pred, v := range m {
Expand Down Expand Up @@ -378,7 +397,7 @@ func (buf *NQuadBuffer) mapToNquads(m map[string]interface{}, idx *int, op int,
continue
}

cr, err := buf.mapToNquads(v, idx, op, pred)
cr, err := buf.mapToNquads(v, op, pred)
if err != nil {
return mr, err
}
Expand Down Expand Up @@ -411,7 +430,7 @@ func (buf *NQuadBuffer) mapToNquads(m map[string]interface{}, idx *int, op int,
continue
}

cr, err := buf.mapToNquads(iv, idx, op, pred)
cr, err := buf.mapToNquads(iv, op, pred)
if err != nil {
return mr, err
}
Expand Down Expand Up @@ -462,13 +481,12 @@ func (buf *NQuadBuffer) ParseJSON(b []byte, op int) error {
return errors.Errorf("Couldn't parse json as a map or an array")
}

var idx int
if len(list) > 0 {
for _, obj := range list {
if _, ok := obj.(map[string]interface{}); !ok {
return errors.Errorf("Only array of map allowed at root.")
}
mr, err := buf.mapToNquads(obj.(map[string]interface{}), &idx, op, "")
mr, err := buf.mapToNquads(obj.(map[string]interface{}), op, "")
if err != nil {
return err
}
Expand All @@ -477,7 +495,7 @@ func (buf *NQuadBuffer) ParseJSON(b []byte, op int) error {
return nil
}

mr, err := buf.mapToNquads(ms, &idx, op, "")
mr, err := buf.mapToNquads(ms, op, "")
buf.checkForDeletion(mr, ms, op)
return err
}
Expand Down
205 changes: 147 additions & 58 deletions chunker/json_parser_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,21 @@
package chunker

import (
"context"
"encoding/binary"
"encoding/json"
"fmt"
"math"
"testing"
"time"

"github.com/dgraph-io/dgo/protos/api"
"github.com/dgraph-io/dgraph/testutil"
"github.com/dgraph-io/dgraph/tok"
"github.com/dgraph-io/dgraph/types"
"github.com/golang/glog"

"github.com/dgraph-io/dgo/protos/api"
"github.com/dgraph-io/dgraph/types"
"github.com/stretchr/testify/require"
"github.com/twpayne/go-geom"
"github.com/twpayne/go-geom/encoding/geojson"
)

func makeNquad(sub, pred string, val *api.Value) *api.NQuad {
Expand Down Expand Up @@ -75,9 +76,33 @@ func Parse(b []byte, op int) ([]*api.NQuad, error) {
return nqs.nquads, err
}

func (exp *Experiment) verify() {
// insert the data into dgraph
dg := testutil.DgraphClientWithGroot(testutil.SockAddr)
ctx := context.Background()
require.NoError(exp.t, dg.Alter(ctx, &api.Operation{DropAll: true}), "drop all failed")
require.NoError(exp.t, dg.Alter(ctx, &api.Operation{Schema: exp.schema}),
"schema change failed")

_, err := dg.NewTxn().Mutate(ctx,
&api.Mutation{Set: exp.nqs, CommitNow: true})
require.NoError(exp.t, err, "mutation failed")

response, err := dg.NewReadOnlyTxn().Query(ctx, exp.query)
require.NoError(exp.t, err, "qeury failed")
testutil.CompareJSON(exp.t, exp.expected, string(response.GetJson()))
}

type Experiment struct {
t *testing.T
nqs []*api.NQuad
schema string
query string
expected string
}

func TestNquadsFromJson1(t *testing.T) {
tn := time.Now().UTC()
geoVal := `{"Type":"Point", "Coordinates":[1.1,2.0]}`
m := true
p := Person{
Name: "Alice",
Expand All @@ -97,26 +122,24 @@ func TestNquadsFromJson1(t *testing.T) {
require.NoError(t, err)

require.Equal(t, 5, len(nq))

oval := &api.Value{Val: &api.Value_StrVal{StrVal: "Alice"}}
require.Contains(t, nq, makeNquad("_:blank-0", "name", oval))

oval = &api.Value{Val: &api.Value_IntVal{IntVal: 26}}
require.Contains(t, nq, makeNquad("_:blank-0", "age", oval))

oval = &api.Value{Val: &api.Value_BoolVal{BoolVal: true}}
require.Contains(t, nq, makeNquad("_:blank-0", "married", oval))

oval = &api.Value{Val: &api.Value_StrVal{StrVal: tn.Format(time.RFC3339Nano)}}
require.Contains(t, nq, makeNquad("_:blank-0", "now", oval))

var g geom.T
err = geojson.Unmarshal([]byte(geoVal), &g)
require.NoError(t, err)
geo, err := types.ObjectValue(types.GeoID, g)
require.NoError(t, err)

require.Contains(t, nq, makeNquad("_:blank-0", "address", geo))
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
age
married
address
}}`,
expected: `{"alice": [
{"name": "Alice",
"age": 26,
"married": true,
"address": {"coordinates": [2,1.1], "type": "Point"}}
]}
`}
exp.verify()
}

func TestNquadsFromJson2(t *testing.T) {
Expand All @@ -138,21 +161,33 @@ func TestNquadsFromJson2(t *testing.T) {

nq, err := Parse(b, SetNquads)
require.NoError(t, err)

require.Equal(t, 6, len(nq))
require.Contains(t, nq, makeNquadEdge("_:blank-0", "friend", "_:blank-1"))
require.Contains(t, nq, makeNquadEdge("_:blank-0", "friend", "1000"))

oval := &api.Value{Val: &api.Value_StrVal{StrVal: "Charlie"}}
require.Contains(t, nq, makeNquad("_:blank-1", "name", oval))

oval = &api.Value{Val: &api.Value_BoolVal{BoolVal: false}}
require.Contains(t, nq, makeNquad("_:blank-1", "married", oval))

oval = &api.Value{Val: &api.Value_StrVal{StrVal: "Bob"}}
require.Contains(t, nq, makeNquad("1000", "name", oval))
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
friend {
name
married
}}}`,
expected: `{"alice":[{
"name":"Alice",
"friend": [
{"name":"Charlie", "married":false},
{"name":"Bob"}
]
}]}`,
}
exp.verify()
}

func outputNq(nqs []*api.NQuad) {
for _, nq := range nqs {
fmt.Printf("%v\n", nq)
}
}
func TestNquadsFromJson3(t *testing.T) {
p := Person{
Name: "Alice",
Expand All @@ -163,28 +198,61 @@ func TestNquadsFromJson3(t *testing.T) {

b, err := json.Marshal(p)
require.NoError(t, err)

nq, err := Parse(b, SetNquads)
require.NoError(t, err)

require.Equal(t, 3, len(nq))
require.Contains(t, nq, makeNquadEdge("_:blank-0", "school", "_:blank-1"))

oval := &api.Value{Val: &api.Value_StrVal{StrVal: "Wellington Public School"}}
require.Contains(t, nq, makeNquad("_:blank-1", "Name", oval))
outputNq(nq)
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
school {Name}
}}`,
expected: `{"alice":[{
"name":"Alice",
"school": [{"Name":"Wellington Public School"}]
}]}`,
}
exp.verify()
}

func TestNquadsFromJson4(t *testing.T) {
json := `[{"name":"Alice","mobile":"040123456","car":"MA0123", "age": 21, "weight": 58.7}]`

nq, err := Parse([]byte(json), SetNquads)
require.NoError(t, err)
require.Equal(t, 5, len(nq))
oval := &api.Value{Val: &api.Value_StrVal{StrVal: "Alice"}}
require.Contains(t, nq, makeNquad("_:blank-0", "name", oval))
require.Contains(t, nq, makeNquad("_:blank-0", "age", &api.Value{Val: &api.Value_IntVal{IntVal: 21}}))
require.Contains(t, nq, makeNquad("_:blank-0", "weight",
&api.Value{Val: &api.Value_DoubleVal{DoubleVal: 58.7}}))
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
mobile
car
age
weight
}}`,
expected: fmt.Sprintf(`{"alice":%s}`, json),
}
exp.verify()
}

func TestNquadsFromMultipleJsonObjects(t *testing.T) {
json := `[{"name":"Alice", "age": 25}, {"name": "Bob", "age": 26}]`

nq, err := Parse([]byte(json), SetNquads)
require.NoError(t, err)
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{people(func: has(name)) {
name
age
}}`,
expected: fmt.Sprintf(`{"people":%s}`, json),
}
exp.verify()
}

func TestJsonNumberParsing(t *testing.T) {
Expand Down Expand Up @@ -227,14 +295,24 @@ func TestNquadsFromJson_NegativeUidError(t *testing.T) {
}

func TestNquadsFromJson_EmptyUid(t *testing.T) {
json := `{"uid":"","name":"Name","following":[{"name":"Bob"}],"school":[{"uid":"","name":"Crown Public School"}]}`

json := `{"uid":"","name":"Alice","following":[{"name":"Bob"}],"school":[{"uid":"",
"name":"Crown Public School"}]}`
nq, err := Parse([]byte(json), SetNquads)
require.NoError(t, err)

require.Equal(t, 5, len(nq))
oval := &api.Value{Val: &api.Value_StrVal{StrVal: "Name"}}
require.Contains(t, nq, makeNquad("_:blank-0", "name", oval))
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
following { name}
school { name}
}}`,
expected: `{"alice":[{"name":"Alice","following":[{"name":"Bob"}],"school":[{
"name":"Crown Public School"}]}]}`,
}
exp.verify()
}

func TestNquadsFromJson_BlankNodes(t *testing.T) {
Expand All @@ -243,8 +321,19 @@ func TestNquadsFromJson_BlankNodes(t *testing.T) {
nq, err := Parse([]byte(json), SetNquads)
require.NoError(t, err)

require.Equal(t, 5, len(nq))
require.Contains(t, nq, makeNquadEdge("_:alice", "school", "_:school"))
exp := &Experiment{
t: t,
nqs: nq,
schema: "name: string @index(exact) .",
query: `{alice(func: eq(name, "Alice")) {
name
following { name}
school { name}
}}`,
expected: `{"alice":[{"name":"Alice","following":[{"name":"Bob"}],"school":[{
"name":"Crown Public School"}]}]}`,
}
exp.verify()
}

func TestNquadsDeleteEdges(t *testing.T) {
Expand Down
Loading