FilesExpand file tree

 master

/

cmd_bulk_insert.c

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

121 lines (94 loc) · 3.64 KB

 master

/

cmd_bulk_insert.c

Copy path

Top

File metadata and controls

• Code
• Blame

121 lines (94 loc) · 3.64 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

/*

* Copyright (c) 2006-Present, Redis Ltd.

* All rights reserved.

*

* Licensed under your choice of the Redis Source Available License 2.0

* (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the

* GNU Affero General Public License v3 (AGPLv3).

*/

#include "cmd_bulk_insert.h"

#include "query_ctx.h"

#include "bulk_insert/bulk_insert.h"

// process "BEGIN" token, expected to be present only on first bulk-insert

// batch, make sure graph key doesn't exists, fails if "BEGIN" token is present

// and graph key 'graphname' already exists

static int _Graph_Bulk_Begin(RedisModuleCtx *ctx, RedisModuleString ***argv,

int *argc, RedisModuleString *rs_graph_name, const char *graphname,

bool *begin) {

ASSERT(argv != NULL);

ASSERT(argc != NULL);

ASSERT(begin != NULL);

ASSERT(graphname != NULL);

ASSERT(rs_graph_name != NULL);

const char *token = RedisModule_StringPtrLen(**argv, NULL);

*begin = strcmp(token, "BEGIN") == 0;

// do nothing if this is not the first BULK call

if(*begin == false) return BULK_OK;

// "BEGIN" token present, skip "BEGIN" token

(*argv) ++;

(*argc) --;

// lock GIL, verify that graph does not already exist

RedisModuleKey *key = NULL;

key = RedisModule_OpenKey(ctx, rs_graph_name, REDISMODULE_READ);

RedisModule_CloseKey(key);

if(key) {

char *err;

int rc __attribute__((unused));

rc = asprintf(&err, "Graph with name '%s' cannot be created, "\

"as key '%s' already exists.", graphname, graphname);

RedisModule_ReplyWithError(ctx, err);

free(err);

return BULK_FAIL;

}

return BULK_OK;

}

int Graph_BulkInsert(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {

if(argc < 3) return RedisModule_WrongArity(ctx);

// bulk-insert process a batch of node/edge creation generated by the

// bulk-insert utility

GraphContext *gc = NULL;

long long node_count = 0; // number of declared nodes

long long edge_count = 0; // number of declared edges

// get graph name

argv += 1; // skip "GRAPH.BULK"

RedisModuleString *rs_graph_name = *argv++;

const char *graphname = RedisModule_StringPtrLen(rs_graph_name, NULL);

argc -= 2; // skip "GRAPH.BULK [GRAPHNAME]"

bool begin = false;

if(_Graph_Bulk_Begin(ctx, &argv, &argc, rs_graph_name, graphname, &begin)

!= BULK_OK) goto cleanup;

gc = GraphContext_Retrieve(ctx, rs_graph_name, false, begin);

// failed to retrieve GraphContext; an error has been emitted

if(gc == NULL) goto cleanup;

// read the user-provided counts for nodes and edges in the current query

if(RedisModule_StringToLongLong(*argv++, &node_count) != REDISMODULE_OK) {

RedisModule_ReplyWithError(ctx, "Error parsing node count.");

goto cleanup;

}

if(RedisModule_StringToLongLong(*argv++, &edge_count) != REDISMODULE_OK) {

RedisModule_ReplyWithError(ctx, "Error parsing relation count.");

goto cleanup;

}

argc -= 2; // already read node count and edge count

int rc = BulkInsert(ctx, gc, argv, argc, node_count, edge_count);

if(rc == BULK_FAIL) {

// if insertion failed, clean up keyspace and free added entities

GraphContext_DecreaseRefCount(gc);

RedisModuleKey *key = NULL;

key = RedisModule_OpenKey(ctx, rs_graph_name, REDISMODULE_WRITE);

RedisModule_DeleteKey(key);

RedisModule_CloseKey(key);

gc = NULL;

goto cleanup;

}

// successful bulk commands should always modify slaves

RedisModule_ReplicateVerbatim(ctx);

// replay to caller

char reply[1024];

int len = snprintf(reply, 1024, "%llu nodes created, %llu edges created",

node_count, edge_count);

RedisModule_ReplyWithStringBuffer(ctx, reply, len);

cleanup:

if(gc) GraphContext_DecreaseRefCount(gc);

return REDISMODULE_OK;

}