Line data Source code
1 : /*
2 : * SPDX-License-Identifier: MPL-2.0
3 : *
4 : * This Source Code Form is subject to the terms of the Mozilla Public
5 : * License, v. 2.0. If a copy of the MPL was not distributed with this
6 : * file, You can obtain one at http://mozilla.org/MPL/2.0/.
7 : *
8 : * Copyright 2024 MonetDB Foundation;
9 : * Copyright August 2008 - 2023 MonetDB B.V.;
10 : * Copyright 1997 - July 2008 CWI.
11 : */
12 :
13 : #include "monetdb_config.h"
14 : #include "rel_file_loader.h"
15 : #include "rel_exp.h"
16 :
17 : #include "mal_instruction.h"
18 : #include "mal_interpreter.h"
19 : #include "mal_parser.h"
20 : #include "mal_builder.h"
21 : #include "mal_namespace.h"
22 : #include "mal_exception.h"
23 : #include "mal_linker.h"
24 : #include "mal_backend.h"
25 : #include "sql_types.h"
26 : #include "rel_bin.h"
27 : #include "json.h"
28 : #include "mutils.h"
29 :
30 : #ifdef HAVE_FCNTL_H
31 : #include <fcntl.h>
32 : #endif
33 :
34 : #include <unistd.h>
35 : // #include <glob.h> // not available on Windows
36 :
37 :
38 : typedef struct JSONFileHandle {
39 : allocator *sa;
40 : char *filename;
41 : int fd;
42 : size_t size;
43 : } JSONFileHandle;
44 :
45 :
46 : static JSONFileHandle *
47 8 : json_open(const char *fname, allocator *sa)
48 : {
49 8 : if (!sa)
50 : return NULL;
51 8 : int fd = MT_open(fname, O_RDONLY);
52 8 : if (fd < 0){
53 : // TODO add relevant trace component
54 0 : TRC_ERROR(SQL_EXECUTION, "Error opening file %s: %s", fname, strerror(errno));
55 0 : return NULL;
56 : }
57 8 : struct stat stb;
58 8 : if (MT_stat(fname, &stb) != 0) {
59 0 : TRC_ERROR(SQL_EXECUTION, "Error stat file %s: %s", fname, strerror(errno));
60 0 : close(fd);
61 0 : return NULL;
62 : }
63 8 : if (stb.st_size > 2000000000) {
64 0 : close(fd);
65 0 : TRC_ERROR(SQL_EXECUTION, "Error file %s: Too large", fname);
66 0 : return NULL;
67 : }
68 :
69 8 : JSONFileHandle *res = sa_alloc(sa, sizeof(JSONFileHandle));
70 8 : res->sa = sa;
71 8 : res->filename = sa_strdup(sa, fname);
72 8 : res->fd = fd;
73 8 : res->size = (size_t)stb.st_size;
74 8 : return res;
75 : }
76 :
77 :
78 : static void
79 8 : json_close(JSONFileHandle *jfh)
80 : {
81 8 : if (jfh && jfh->fd)
82 8 : close(jfh->fd);
83 8 : }
84 :
85 :
86 : static char *
87 8 : read_json_file(JSONFileHandle *jfh)
88 : {
89 8 : char *content = NULL;
90 8 : if (jfh) {
91 8 : unsigned int length = (unsigned int)jfh->size;
92 8 : content = sa_zalloc(jfh->sa, length + 1);
93 8 : if (content) {
94 8 : ssize_t nbytes = read(jfh->fd, content, length);
95 8 : if (nbytes < 0)
96 : return NULL;
97 8 : content[length + 1] = '\0';
98 : }
99 : }
100 : return content;
101 : }
102 :
103 :
104 : static str
105 6 : append_terms(allocator *sa, JSON *jt, BAT *b)
106 : {
107 6 : str error = MAL_SUCCEED;
108 6 : size_t offset = 0;
109 6 : JSONterm *root = jt->elm, *t;
110 6 : char *v = NULL;
111 6 : int depth = 0;
112 :
113 312 : while(offset < (size_t) jt->free && !error) {
114 306 : t = jt->elm + offset;
115 306 : JSONterm *prev = offset > 0 ? jt->elm + (offset - 1) : NULL;
116 306 : switch(t->kind) {
117 84 : case JSON_ARRAY:
118 : case JSON_OBJECT:
119 84 : if ((root->kind == JSON_ARRAY && depth == 1) || root->kind == JSON_OBJECT) {
120 12 : v = sa_strndup(sa, t->value, t->valuelen);
121 12 : if (v) {
122 12 : if (BUNappend(b, v, false) != GDK_SUCCEED) {
123 0 : error = createException(SQL, "json.append_terms", "BUNappend failed!");
124 : }
125 : }
126 : }
127 84 : if ((prev && (prev->kind == JSON_ARRAY || prev->kind == JSON_VALUE)) || (prev==NULL && root->kind ==
128 : JSON_ARRAY))
129 42 : depth ++;
130 : break;
131 : case JSON_ELEMENT:
132 : case JSON_STRING:
133 : case JSON_NUMBER:
134 : case JSON_NULL:
135 : break;
136 36 : case JSON_VALUE:
137 36 : depth --;
138 36 : break;
139 0 : default:
140 0 : error = createException(SQL, "json.append_terms", "unknown json term");
141 0 : break;
142 : }
143 306 : offset +=1;
144 : }
145 6 : return error;
146 : }
147 :
148 :
149 : static str
150 6 : json_relation(mvc *sql, sql_subfunc *f, char *filename, list *res_exps, char *tname)
151 : {
152 6 : (void) filename;
153 6 : char *res = MAL_SUCCEED;
154 6 : list *types = sa_list(sql->sa);
155 6 : list *names = sa_list(sql->sa);
156 : // use file name as columnn name ?
157 6 : char *cname = sa_strdup(sql->sa, "json");
158 6 : list_append(names, cname);
159 6 : sql_schema *jsons = mvc_bind_schema(sql, "sys");
160 6 : if (!jsons)
161 : return NULL;
162 6 : sql_subtype *st = SA_ZNEW(sql->sa, sql_subtype);
163 6 : st->digits = st->scale = 0;
164 6 : st->multiset = 0;
165 6 : st->type = schema_bind_type(sql, jsons, "json");
166 6 : list_append(types, st);
167 6 : sql_exp *ne = exp_column(sql->sa, a_create(sql->sa, tname), cname, st, CARD_MULTI, 1, 0, 0);
168 6 : set_basecol(ne);
169 6 : ne->alias.label = -(sql->nid++);
170 6 : list_append(res_exps, ne);
171 6 : f->tname = tname;
172 6 : f->res = types;
173 6 : f->coltypes = types;
174 6 : f->colnames = names;
175 6 : return res;
176 : }
177 :
178 :
179 : static void *
180 6 : json_load(void *BE, sql_subfunc *f, char *filename, sql_exp *topn)
181 : {
182 6 : (void) topn; // TODO include topn
183 6 : backend *be = BE;
184 6 : allocator *sa = be->mvc->sa;
185 6 : sql_subtype *tpe = f->res->h->data;
186 6 : const char *tname = f->tname;
187 6 : const char *cname = f->colnames->h->data;
188 :
189 6 : stmt *s = stmt_none(be);
190 6 : InstrPtr q = newStmt(be->mb, "json", "read_json");
191 6 : q = pushStr(be->mb, q, filename);
192 6 : pushInstruction(be->mb, q);
193 6 : s->nr = getDestVar(q);
194 6 : s->q = q;
195 6 : s->nrcols = 1;
196 6 : s->subtype = *tpe;
197 : // is alias essential here?
198 6 : s = stmt_alias(be, s, 1, a_create(sa, tname), cname);
199 6 : return s;
200 : }
201 :
202 :
203 : static int TYPE_json;
204 :
205 :
206 : static str
207 346 : JSONprelude(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
208 : {
209 346 : (void)cntxt; (void)mb; (void)stk; (void)pci;
210 346 : TYPE_json = ATOMindex("json");
211 :
212 346 : fl_register("json", &json_relation, &json_load);
213 346 : return MAL_SUCCEED;
214 : }
215 :
216 :
217 : static str
218 345 : JSONepilogue(void *ret)
219 : {
220 345 : fl_unregister("json");
221 345 : (void)ret;
222 345 : return MAL_SUCCEED;
223 : }
224 :
225 :
226 : static str
227 6 : JSONread_json(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
228 : {
229 6 : (void) cntxt; (void) mb;
230 6 : char *msg = MAL_SUCCEED;
231 6 : char *fname = *(str*)getArgReference(stk, pci, pci->retc);
232 6 : allocator *sa = sa_create(NULL);
233 6 : JSONFileHandle *jfh = json_open(fname, sa);
234 6 : const char* json_str = NULL;
235 6 : JSON *jt = NULL;
236 6 : BAT *b = NULL;
237 6 : if (!jfh) {
238 0 : sa_destroy(sa);
239 0 : msg = createException(SQL, "json.read_json", "Failed to open file %s", fname);
240 0 : return msg;
241 : }
242 6 : json_str = read_json_file(jfh);
243 6 : json_close(jfh);
244 6 : if (json_str)
245 6 : jt = JSONparse(json_str);
246 6 : if (jt) {
247 6 : if (jt->error == NULL) {
248 6 : b = COLnew(0, TYPE_json, 0, TRANSIENT);
249 6 : if ((msg = append_terms(sa, jt, b)) == MAL_SUCCEED) {
250 6 : bat *res = getArgReference_bat(stk, pci, 0);
251 6 : *res = b->batCacheid;
252 6 : BBPkeepref(b);
253 : } else
254 0 : BBPreclaim(b);
255 : } else {
256 : msg = jt->error;
257 : }
258 6 : JSONfree(jt);
259 : } else {
260 0 : msg = createException(SQL, "json.read_json", "JSONparse error");
261 : }
262 6 : sa_destroy(sa);
263 6 : return msg;
264 : }
265 :
266 :
267 : static str
268 2 : JSONread_nd_json(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
269 : {
270 2 : (void) cntxt; (void) mb;
271 2 : char *msg = MAL_SUCCEED;
272 2 : char *fname = *(str*)getArgReference(stk, pci, pci->retc);
273 2 : allocator *sa = sa_create(NULL);
274 2 : JSONFileHandle *jfh = json_open(fname, sa);
275 2 : if (!jfh) {
276 0 : sa_destroy(sa);
277 0 : msg = createException(SQL, "json.read_nd_json", "Failed to open file %s", fname);
278 0 : return msg;
279 : }
280 2 : char *content = read_json_file(jfh);
281 2 : json_close(jfh);
282 2 : BAT *b = COLnew(0, TYPE_json, 0, TRANSIENT);
283 2 : if (content) {
284 2 : if (b) {
285 : size_t cnt = 0;
286 : char *head = content;
287 : char *tail = content;
288 11180 : while (cnt < (jfh->size + 1)) {
289 11178 : if (head[0] == '\n' || (head[0] == '\r' && head[1] == '\n')) {
290 20 : int skip = 1;
291 20 : if (head[0] == '\r' && head[1] == '\n')
292 20 : skip = 2;
293 20 : head[0] = '\0';
294 20 : JSON *jt = JSONparse(tail);
295 20 : if (jt) {
296 : // must be valid json obj str
297 20 : if (BUNappend(b, tail, false) != GDK_SUCCEED) {
298 0 : msg = createException(SQL, "json.read_nd_json", "BUNappend failed!");
299 0 : break;
300 : }
301 : } else {
302 0 : msg = createException(SQL, "json.read_nd_json", "Invalid json object, JSONparse failed!");
303 0 : break;
304 : }
305 20 : tail = head + skip;
306 20 : while (tail[0] == '\n') // multiple newlines e.g. \n\n
307 0 : tail ++;
308 : head = tail;
309 : }
310 11178 : head ++;
311 11178 : cnt ++;
312 : }
313 : } else {
314 0 : msg = createException(SQL, "json.read_nd_json", "Failed to allocate bat");
315 : }
316 : } else {
317 0 : msg = createException(SQL, "json.read_nd_json", "Failed to read file %s", fname);
318 : }
319 0 : if (msg == MAL_SUCCEED) {
320 2 : bat *res = getArgReference_bat(stk, pci, 0);
321 2 : *res = b->batCacheid;
322 2 : BBPkeepref(b);
323 : } else {
324 0 : BBPreclaim(b);
325 : }
326 :
327 2 : sa_destroy(sa);
328 2 : return msg;
329 : }
330 :
331 :
332 : #include "mel.h"
333 :
334 : unsigned char _json_sql[106] = {
335 : "create function sys.read_nd_json(fname string)\n"
336 : "returns table(json JSON)\n"
337 : "external name json.read_nd_json;\n"
338 : };
339 : #include "monetdb_config.h"
340 : #include "sql_import.h"
341 :
342 : static mel_func json_init_funcs[] = {
343 : pattern("json", "prelude", JSONprelude, false, "", noargs),
344 : command("json", "epilogue", JSONepilogue, false, "", noargs),
345 : pattern("json", "read_json", JSONread_json, false, "Reads json file", args(1,2, batarg("", json), arg("filename", str))),
346 : pattern("json", "read_nd_json", JSONread_nd_json, false, "Reads new line delimited json objects", args(1,2, batarg("", json), arg("filename", str))),
347 : { .imp=NULL }
348 : };
349 :
350 : #ifdef _MSC_VER
351 : #undef read
352 : #pragma section(".CRT$XCU",read)
353 : #endif
354 346 : LIB_STARTUP_FUNC(init_json_mal)
355 346 : { mal_module("json", NULL, json_init_funcs);
356 346 : sql_register("json", _json_sql); }
357 :
|
