VirtCsvFileBulkLoader Virtuoso CSV File Bulk Loader Virtuoso CSV File Bulk Loader The Virtuoso CSV File Bulk loader enables the bulk loading of CSV files into Virtuoso, to be stored as tables. CSV Bulk Load Functions The functions listed here are used for performing the CSV file bulk load operations. The full source of these functions are included below, for users whose Virtuoso instances do not have them pre-loaded. csv_register (path, mask) Takes a single directory path containing the gzipped, zipped or uncompressed CSV files to be loaded. All files matching the mask pattern will be registered for loading. csv_register_all (path, mask) Same as above but recursive over the directory specified in path. csv_loader_run (max_files, log_enable := 2) Run the CSV bulk loader to load the data into the database. The maximum number of files to be loaded can be specified as max_file, and log_enable=2 can be set to minimize locks during load. Configuration and Usage All directories (or their parents) containing CSV files to be loaded must be included in the DirsAllowed parameter of the active Virtuoso INI file. Each CSV file or compressed archive of CSV files may have its own configuration options. If no options are given, the structure will be guessed from the content of the CSV file. If a CSV file contains no headers to indicate the column names, then a table corresponding to this file must be created manually before importing. To load a CSV file into an existing table, create a file corresponding to the name of the CSV file with a .tb filename extension. Place a single entry in the new file with the fully qualified name of the desired target table in the database. By default, CSV files are loaded in the csv.DBA. schema of the database, with a table name matching the CSV filename. If a .tb file of the same name is found, the schema location and table name contained there will be used instead. The csv_register() function creates or appends to the table DB.DBA.csv_load_list which maintains a list of CSV files and the tables into which they are loaded. Note: Before reloading a CSV file, its entry must be removed from the DB.DBA.csv_load_list table and its table must be dropped from the csv.DBA. schema of the database (or wherever it has been and will be loaded). If a CSV file structure differs from the default CSV file settings, the loader will look for a configuration file of the same name as the CSV file with a .cfg filename extension. This file should contain parameters similar to those below, indicating the CSV file's structure: [csv] csv-delimiter=<delimiter char> csv-quote=<quote char> header=<zero based header offset> offset=<zero based data offset> Invisible "tab" and "space" delimiters should be specified by those names, without the quotation marks. Other delimiter characters (comma, period, etc.) should simply be typed in. "Smart" quotation marks which differ at start and end (including but not limited to « », ‹ ›, “ ”, and ‘ ’) are not currently supported. Example Consider loading a gzipped CSV file, csv-example.csv.gz, with the non-default CSV structure below: 'Southern North Island wood availability forecast for the period 2008-2040' 'Table 14: Wood availability and average clearfell age for other species in Eastern Southern North Island' 'Year ending' 'Recoverable volume' 'Average age' 'December' '(000 m3 i.b.)' '(years)' 2006 0 0 2007 0 0 2008 48 49 2009 45 46 ... In this example the header is on the third line, #2 with a zero-base the data starts from the fifth line, #4 with a zero-base the delimiter is tab the quote char is the single-quote, or apostrophe Loading this file requires the creation of a configuration file, csv-example.cfg, containing the entries: [csv] csv-delimiter=tab csv-quote=' header=2 offset=4 Assuming all files are located in a directory called CSV, found within the Virtuoso server's working (i.e., database) directory, and that this directory is included in the DirsAllowed Virtuoso configuration parameter, the commands to register and load the CSV data files are: csv_register ('./CSV', '*.gz'); csv_loader_run (); CSV Bulk Loader scripts The following scripts need to be run through Virtuoso's iSQL tool (either command-line or HTTP-based), to create the procedure for performing CSV file bulk load operations. (This may be pre-loaded in more recent Virtuoso instances.) -- DROP TABLE csv_load_list; CREATE TABLE csv_load_list ( cl_file VARCHAR, cl_file_in_zip VARCHAR, cl_state INT DEFAULT 0, cl_error LONG VARCHAR, cl_table VARCHAR, cl_options ANY, cl_started DATETIME, cl_done DATETIME, PRIMARY KEY (cl_file, cl_file_in_zip)) CREATE INDEX cl_state ON csv_load_list (cl_state) ; CREATE PROCEDURE csv_cols_cb ( INOUT r ANY, IN inx INT, INOUT cbd ANY) { IF (cbd IS NULL) cbd := VECTOR (); cbd := vector_concat (cbd, VECTOR (r)); } ; CREATE PROCEDURE csv_get_cols_array ( INOUT ss ANY, IN hr INT, IN offs INT, IN opts ANY) { DECLARE h, res ANY; DECLARE inx, j, ncols, no_head INT; h := NULL; no_head := 0; IF (hr < 0) { no_head := 1; hr := 0; } IF (offs < 0) offs := 0; res := VECTOR (); csv_parse (ss, 'DB.DBA.csv_cols_cb', h, 0, offs + 10, opts); IF (h IS NOT NULL AND LENGTH (h) > offs) { DECLARE _row ANY; _row := h[hr]; FOR (j := 0; j < LENGTH (_row); j := j + 1) { res := vector_concat (res, VECTOR (VECTOR (SYS_ALFANUM_NAME (CAST (_row[j] AS VARCHAR)), NULL))); } FOR (inx := offs; inx < LENGTH (h); inx := inx + 1) { _row := h[inx]; FOR (j := 0; j < LENGTH (_row); j := j + 1) { IF (res[j][1] IS NULL AND NOT (ISSTRING (_row[j]) AND _row[j] = '') AND _row[j] IS NOT NULL) res[j][1] := __tag (_row[j]); ELSE IF (__tag (_row[j]) <> res[j][1] AND 189 = res[j][1] AND (ISDOUBLE (_row[j]) OR isfloat (_row[j]))) res[j][1] := __tag (_row[j]); ELSE IF (__tag (_row[j]) <> res[j][1] AND ISINTEGER (_row[j]) AND (res[j][1] = 219 OR 190 = res[j][1])) ; ELSE IF (__tag (_row[j]) <> res[j][1]) res[j][1] := -1; } } } FOR (inx := 0; inx < LENGTH (res); inx := inx + 1) { IF (NOT ISSTRING (res[inx][0]) AND NOT ISNULL (res[inx][0])) no_head := 1; ELSE IF (trim (res[inx][0]) = '' OR ISNULL (res[inx][0])) res[inx][0] := sprintf ('COL%d', inx); } FOR (inx := 0; inx < LENGTH (res); inx := inx + 1) { IF (res[inx][1] = -1 OR res[inx][1] IS NULL) res[inx][1] := 'VARCHAR'; ELSE res[inx][1] := dv_type_title (res[inx][1]); } IF (no_head) { FOR (inx := 0; inx < LENGTH (res); inx := inx + 1) { res[inx][0] := sprintf ('COL%d', inx); } } RETURN res; } ; CREATE PROCEDURE csv_get_table_def (IN fn VARCHAR, IN f VARCHAR, IN opts ANY) { DECLARE arr ANY; DECLARE s, r, ss ANY; DECLARE i, offs, st INT; IF (__tag (f) = 185) s := f; ELSE IF (f LIKE '%.gz') s := gz_file_open (f); ELSE s := file_open (f); st := 0; offs := 1; IF (ISVECTOR (opts) AND MOD (LENGTH (opts), 2) = 0) { st := atoi (get_keyword ('header', opts, '0')); offs := atoi (get_keyword ('offset', opts, '1')); } arr := csv_get_cols_array (s, st, offs, opts); ss := string_output (); http (sprintf ('CREATE TABLE %s ( \n', fn), ss); FOR (i := 0; i < LENGTH (arr); i := i + 1) { http (sprintf ('\t"%I" %s', arr[i][0], arr[i][1]), ss); IF (i < LENGTH (arr) - 1) http (', \n', ss); } http (')', ss); RETURN string_output_string (ss); } ; CREATE PROCEDURE csv_register (IN path VARCHAR, IN mask VARCHAR) { DECLARE ls ANY; DECLARE inx INT; ls := sys_dirlist (path, 1); FOR (inx := 0; inx < LENGTH (ls); inx := inx + 1) { IF (ls[inx] LIKE mask) { IF (NOT (EXISTS (SELECT 1 FROM DB.DBA.CSV_LOAD_LIST WHERE CL_FILE = path || '/' || ls[inx] FOR UPDATE))) { DECLARE tbfile, ofile, tb, f, tbname VARCHAR; DECLARE opts ANY; tb := NULL; f := ls[inx]; tbfile := path || '/' || regexp_replace (f, '(\\.csv(\\.gz)?)|(\\.zip)', '') || '.tb'; ofile := path || '/' || regexp_replace (f, '(\\.csv(\\.gz)?)|(\\.zip)', '') || '.cfg'; opts := NULL; IF (file_stat (ofile) <> 0) { DECLARE delim, quot, header, offs, enc VARCHAR; delim := cfg_item_value (ofile, 'csv', 'csv-delimiter'); quot := cfg_item_value (ofile, 'csv', 'csv-quote'); enc := cfg_item_value (ofile, 'csv', 'encoding'); header := cfg_item_value (ofile, 'csv', 'header'); offs := cfg_item_value (ofile, 'csv', 'offset'); IF (delim IS NOT NULL) { delim := REPLACE (delim, 'tab', '\t'); delim := REPLACE (delim, 'space', ' '); opts := vector_concat (opts, VECTOR ('csv-delimiter', delim)); } IF (quot IS NOT NULL) opts := vector_concat (opts, VECTOR ('csv-quote', quot)); IF (enc IS NOT NULL) opts := vector_concat (opts, VECTOR ('encoding', enc)); IF (header IS NOT NULL) opts := vector_concat (opts, VECTOR ('header', header)); IF (offs IS NOT NULL) opts := vector_concat (opts, VECTOR ('offset', offs)); } IF (file_stat (tbfile) <> 0) tbname := TRIM (file_to_string (tbfile), ' \r\n'); ELSE tbname := complete_table_name ('CSV.DBA.'||SYS_ALFANUM_NAME (f), 1); IF (EXISTS (SELECT 1 FROM SYS_KEYS WHERE KEY_TABLE = tbname)) { tb := tbname; } ELSE { IF (f LIKE '%.csv' OR f LIKE '%.csv.gz') { DECLARE stat, msg ANY; stat := '00000'; DECLARE CONTINUE HANDLER FOR SQLSTATE '*' { log_message (sprintf ('Cannot guess table name FOR file %s', f)); }; { EXEC (csv_get_table_def (tbname, path||'/'||f, opts), stat, msg); IF (stat = '00000') tb := tbname; ELSE log_message (sprintf ('Cannot guess table name FOR file %s', f)); } } ELSE IF (f LIKE '%.zip') { DECLARE ff, ss ANY; ff := unzip_list (path || '/' || f); FOREACH (ANY zf IN ff) DO { IF (zf[1] > 0 AND zf[0] LIKE '%.csv') { ss := unzip_file (path || '/' || f, zf[0]); tbname := complete_table_name ('CSV.DBA.'||SYS_ALFANUM_NAME (zf[0]), 1); DECLARE stat, msg ANY; tb := NULL; stat := '00000'; DECLARE CONTINUE HANDLER FOR SQLSTATE '*' { log_message (sprintf ('Cannot guess table name FOR zipped file %s', zf[0])); }; { EXEC (csv_get_table_def (tbname, ss, opts), stat, msg); IF (stat = '00000') tb := tbname; ELSE log_message (sprintf ('Cannot guess table name FOR zipped file %s', zf[0])); IF (tb IS NOT NULL) INSERT INTO DB.DBA.CSV_LOAD_LIST (cl_file, cl_file_in_zip, cl_table, cl_options) VALUES (path || '/' || f, zf[0], tb, opts); } } } tb := NULL; } ELSE log_message (sprintf ('Cannot guess table name FOR file %s', f)); } IF (tb IS NOT NULL) { INSERT INTO DB.DBA.CSV_LOAD_LIST (cl_file, cl_file_in_zip, cl_table, cl_options) VALUES (path || '/' || f, '', tb, opts); } } COMMIT WORK; } } } ; CREATE PROCEDURE csv_register_all (IN path VARCHAR, IN mask VARCHAR) { DECLARE ls ANY; DECLARE inx INT; ls := sys_dirlist (path, 0); csv_register (path, mask); FOR (inx := 0; inx < LENGTH (ls); inx := inx + 1) { IF (ls[inx] <> '.' AND ls[inx] <> '..') { csv_register_all (path||'/'||ls[inx], mask); } } } ; CREATE PROCEDURE csv_ld_file (IN f VARCHAR, IN zf VARCHAR, IN tb VARCHAR, IN ld_mode INT, IN opts ANY) { DECLARE ss ANY; DECLARE offs, st INT; st := 0; offs := 1; DECLARE EXIT HANDLER FOR SQLSTATE '*' { ROLLBACK WORK; UPDATE DB.DBA.CSV_LOAD_LIST SET CL_STATE = 2, CL_DONE = NOW (), CL_ERROR = __sql_state || ' ' || __sql_message WHERE CL_FILE = f AND CL_FILE_IN_ZIP = zf; COMMIT WORK; log_message (sprintf (' File %s error %s %s', f, __sql_state, __sql_message)); RETURN; }; IF (ISVECTOR (opts) AND MOD (LENGTH (opts), 2) = 0) { st := atoi (get_keyword ('header', opts, '0')); offs := atoi (get_keyword ('offset', opts, '1')); } IF (f LIKE '%.zip' AND LENGTH (zf) = 0) { DECLARE ff ANY; ff := unzip_list (f); FOREACH (ANY zzf IN ff) DO { IF (zzf[1] > 0 AND zzf[0] LIKE '%.csv') { ss := unzip_file (f, zzf[0]); csv_load (ss, offs, NULL, tb, ld_mode, opts); } } } ELSE IF (f LIKE '%.zip' AND LENGTH (zf) > 0) { ss := unzip_file (f, zf); csv_load (ss, offs, NULL, tb, ld_mode, opts); } ELSE IF (f LIKE '%.gz') { ss := gz_file_open (f); csv_load (ss, offs, NULL, tb, ld_mode, opts); } ELSE csv_load_file (f, offs, NULL, tb, ld_mode, opts); } ; CREATE PROCEDURE csv_ld_array () { DECLARE first, last, zfirst, zlast, arr, len, local, opt, zf ANY; DECLARE cr CURSOR FOR SELECT TOP 100 CL_FILE, CL_TABLE, CL_OPTIONS, CL_FILE_IN_ZIP FROM DB.DBA.CSV_LOAD_LIST TABLE OPTION (INDEX cl_state) WHERE CL_STATE = 0 FOR UPDATE; DECLARE fill INT; DECLARE f, g VARCHAR; DECLARE r ANY; WHENEVER NOT FOUND GOTO done; first := 0; last := 0; arr := make_array (100, 'any'); fill := 0; OPEN cr; len := 0; FOR (;;) { FETCH cr INTO f, g, opt, zf; IF (0 = first) { first := f; zfirst := zf; } last := f; zlast := zf; arr[fill] := VECTOR (f, g, opt, zf); len := len + CAST (file_stat (f, 1) as INT); fill := fill + 1; IF (len > 2000000) GOTO done; } done: IF (0 = first) RETURN 0; UPDATE CSV_LOAD_LIST SET cl_state = 1, cl_started = NOW () WHERE cl_file >= first AND cl_file <= last AND CL_FILE_IN_ZIP >= zfirst AND CL_FILE_IN_ZIP <= zlast; RETURN arr; } ; CREATE PROCEDURE csv_loader_run (IN max_files integer := NULL, IN log_enable INT := 2) { DECLARE sec_delay float; DECLARE _f, _graph VARCHAR; DECLARE arr ANY; DECLARE xx, inx, tx_mode, ld_mode INT; ld_mode := log_enable; WHILE (1) { SET ISOLATION = 'repeatable'; DECLARE EXIT HANDLER FOR SQLSTATE '40001' { ROLLBACK WORK; sec_delay := RND(1000)*0.001; log_message(sprintf('deadlock in loader, waiting %d milliseconds', CAST (sec_delay * 1000 AS INTEGER))); DELAY(sec_delay); GOTO again; }; again:; IF (EXISTS (SELECT 1 FROM DB.DBA.CSV_LOAD_LIST WHERE CL_FILE = '##stop')) { log_message ('File load stopped by rdf_load_stop.'); RETURN; } log_enable (tx_mode, 1); IF (max_files IS NOT NULL AND max_files <= 0) { COMMIT WORK; log_message ('Max_files reached. Finishing.'); RETURN; } WHENEVER NOT FOUND GOTO looks_empty; -- log_message ('Getting next file.'); SET ISOLATION = 'serializable'; arr := csv_ld_array (); COMMIT WORK; IF (0 = arr) GOTO looks_empty; log_enable (ld_mode, 1); FOR (inx := 0; inx < 100; inx := inx + 1) { IF (0 = arr[inx]) GOTO arr_done; csv_ld_file (arr[inx][0], arr[inx][3], arr[inx][1], ld_mode, arr[inx][2]); UPDATE DB.DBA.CSV_LOAD_LIST SET CL_STATE = 2, CL_DONE = CURDATETIME () WHERE CL_FILE = arr[inx][0] AND CL_FILE_IN_ZIP = arr[inx][3]; } arr_done: log_enable (tx_mode, 1); IF (max_files IS NOT NULL) max_files := max_files - 100; COMMIT WORK; } looks_empty: COMMIT WORK; log_message ('No more files to load. Loader has finished,'); RETURN; } ;