Changeset: f74f4382f384 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=f74f4382f384
Modified Files:
sql/backends/monet5/bam/85_bam.sql
sql/backends/monet5/bam/Tests/benchmark2.sql
sql/backends/monet5/bam/bam.mal
sql/backends/monet5/bam/bam_loader.c
sql/backends/monet5/bam/bam_loader.h
Branch: DVframework_bam
Log Message:
* Splitted bam_loader into bam_loader_repos and bam_loader_file, which loads an
entire repository of files and just a single file respectively.
* Made minor improvements on benchmark
diffs (truncated from 506 to 300 lines):
diff --git a/sql/backends/monet5/bam/85_bam.sql
b/sql/backends/monet5/bam/85_bam.sql
--- a/sql/backends/monet5/bam/85_bam.sql
+++ b/sql/backends/monet5/bam/85_bam.sql
@@ -1,5 +1,8 @@
-CREATE PROCEDURE bam_loader(repo STRING, dbschema INT, num_threads INT)
-EXTERNAL NAME bam.bam_loader;
+CREATE PROCEDURE bam_loader_repos(bam_repos STRING, dbschema INT, num_threads
INT)
+EXTERNAL NAME bam.bam_loader_repos;
+
+CREATE PROCEDURE bam_loader_file(bam_file STRING, dbschema INT, num_threads
INT)
+EXTERNAL NAME bam.bam_loader_file;
CREATE FUNCTION bam_flag(flag SMALLINT, name STRING)
diff --git a/sql/backends/monet5/bam/Tests/benchmark2.sql
b/sql/backends/monet5/bam/Tests/benchmark2.sql
--- a/sql/backends/monet5/bam/Tests/benchmark2.sql
+++ b/sql/backends/monet5/bam/Tests/benchmark2.sql
@@ -6,9 +6,12 @@ WITH qnames AS (
SELECT qname
FROM bam.alignments
WHERE file_id = 1
+ AND seq <> '*'
+ AND qual <> '*'
AND bam_flag(flag, 'firs_segm') <> bam_flag(flag, 'last_segm')
AND (bam_flag(flag, 'seco_alig') = False
OR bam_flag(flag, 'segm_unma') = True)
+ AND mapq < 100
GROUP BY qname
HAVING COUNT(*) = 2
AND SUM(bam_flag(flag, 'firs_segm')) = 1
@@ -26,6 +29,7 @@ FROM (
AND bam_flag(flag, 'firs_segm') <> bam_flag(flag, 'last_segm')
AND (bam_flag(flag, 'seco_alig') = False
OR bam_flag(flag, 'segm_unma') = True)
+ AND mapq < 100
AND qname IN (
SELECT *
FROM qnames
@@ -42,6 +46,7 @@ FROM (
AND bam_flag(flag, 'firs_segm') <> bam_flag(flag, 'last_segm')
AND (bam_flag(flag, 'seco_alig') = False
OR bam_flag(flag, 'segm_unma') = True)
+ AND mapq < 100
AND qname IN (
SELECT *
FROM qnames
@@ -57,7 +62,7 @@ ORDER BY qname;
-- and the qual/qual-reverse for both reads of this qname. I.e., every tuple
in the result contains a read pair.
-- The outer query joins two subresults together. Both subresults only contain
primary and unmapped alignments
-- from file with file_id=1. Also, alignments that have not stored their seq
or qual value are filtered out in the
--- subresults.
+-- subresults and alignments with a mapping quality >= 100 are also filtered
out.
-- The subresults contain alignments with their 'firs_segm' and 'last_segm'
flag set respectively. Furthermore,
-- since we don't want alignments showing up in the result more than once, we
have to make sure that for every
-- qname, exactly one left and exactly one right alignment remains in the
subresults. Therefore, qnames that
@@ -68,6 +73,8 @@ ORDER BY qname;
-- 'firs_segm' and one has to be flagged as'last_segm'. This makes sure that
only qnames will result that
-- have exactly one matching alignment for both the left side and the right
side of the result.
-- The result is ordered by qname.
+-- A custom renderer has been developed that can be used in mclient to output
actual FASTQ files as a result
+-- of this query.
@@ -86,22 +93,20 @@ WITH alig AS (
AND bam_flag(flag, 'firs_segm') <> bam_flag(flag, 'last_segm')
AND (bam_flag(flag, 'seco_alig') = False
OR bam_flag(flag, 'segm_unma') = True)
-), qnames AS (
- SELECT qname
- FROM alig
- GROUP BY qname
- HAVING COUNT(*) = 2
- AND SUM(bam_flag(flag, 'firs_segm')) = 1
- AND SUM(bam_flag(flag, 'last_segm')) = 1
+ AND mapq < 100
), alig_proj AS (
SELECT qname, flag,
CASE WHEN bam_flag(flag, 'segm_reve') THEN reverse_seq(seq) ELSE seq
END AS seq,
CASE WHEN bam_flag(flag, 'segm_reve') THEN reverse_qual(qual) ELSE
qual END AS qual
FROM alig
WHERE qname IN (
- SELECT *
- FROM qnames
- )
+ SELECT qname
+ FROM alig
+ GROUP BY qname
+ HAVING COUNT(*) = 2
+ AND SUM(bam_flag(flag, 'firs_segm')) = 1
+ AND SUM(bam_flag(flag, 'last_segm')) = 1
+ )
)
SELECT a1.qname AS qname, a1.seq AS seq1, a1.qual AS qual1, a2.seq AS seq2,
a2.qual AS qual2
FROM (
diff --git a/sql/backends/monet5/bam/bam.mal b/sql/backends/monet5/bam/bam.mal
--- a/sql/backends/monet5/bam/bam.mal
+++ b/sql/backends/monet5/bam/bam.mal
@@ -1,8 +1,12 @@
module bam;
-pattern bam.bam_loader(entry:str, dbschema:int, num_threads:int):void
-address bam_loader
-comment "Read the files in the BAM repository, fill and return a
temp_container accordingly.";
+pattern bam.bam_loader_repos(bam_repos:str, dbschema:int, num_threads:int):void
+address bam_loader_repos
+comment "Read the files in the given BAM repository; fill and return a
temp_container accordingly.";
+
+pattern bam.bam_loader_file(bam_file:str, dbschema:int, num_threads:int):void
+address bam_loader_file
+comment "Read the bam file given as the first parameter; fill and return a
temp_container accordingly.";
pattern bam.register_table(ticket:lng, table_idx:int)(:any...)
address register_table
diff --git a/sql/backends/monet5/bam/bam_loader.c
b/sql/backends/monet5/bam/bam_loader.c
--- a/sql/backends/monet5/bam/bam_loader.c
+++ b/sql/backends/monet5/bam/bam_loader.c
@@ -113,11 +113,8 @@ str _coln_alignments_extra[] = {"tag"
int _colt_alignments_extra[] = {TYPE_str, TYPE_lng , TYPE_sht ,
TYPE_str, TYPE_str};
-
-
/* File format specific functions */
-static lng _get_line_nr(str filename);
-static lng _get_file_paths(str repo_path, str** ret_file_paths);
+static str _bam_loader_file(Client cntxt, MalBlkPtr mb, str file_path, sht
dbschema, sht nr_threads);
static str _next_file_id(Client cntxt, MalBlkPtr mb, sht *next_file_id);
static str _init_tc_full_load(_temp_container *ret_tc);
static str _init_tc_simple_split(_temp_container *ret_tc);
@@ -130,6 +127,8 @@ static str _store_alignment_full_load(_a
static str _store_alignment_simple_split(_alignment *a, _temp_container
*ret_tc);
/* Generic functions */
+static lng _get_line_nr(str filename);
+static lng _get_file_paths(str repo_path, str** ret_file_paths);
static str _init_temp_subcontainer(_temp_subcontainer *ret_tsc, str
*col_names, int *col_types, sht num_cols);
static str _insert_into_vault(Client cntxt, _temp_container* tc);
static int _read_string_until_delim(str *src, str *ret, char *delims, sht
num_delims);
@@ -149,10 +148,10 @@ sht nr_alignment_tables[] = {1, 2};
-/* File format specific functions */
+
str
-bam_loader(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+bam_loader_repos(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
/* TODO: This function is now called for BAM files, eventually there will
have to be one generic */
/* function that e.g. receives the file format for which a DV should be
initialized as an argument */
@@ -174,165 +173,80 @@ bam_loader(Client cntxt, MalBlkPtr mb, M
for(k=0; k<nr_files; k++)
{
- int bufsize = 255;
- str buf = GDKmalloc(bufsize * sizeof(char));
-
- int start = GDKms();
- float duration_sec;
-
- str file_path = file_paths[k];
- sht next_file_id = 0;
- str err1 = NULL, err2 = NULL;
- _temp_container *tc = (_temp_container
*)GDKmalloc(sizeof(_temp_container));
-
- if(buf == NULL || tc == NULL)
- throw(MAL, "bam_loader", MAL_MALLOC_FAIL);
-
- if((err1 = _next_file_id(cntxt, mb, &next_file_id)) != MAL_SUCCEED)
- err2 = "Error while retrieving next file_id: %s\n";
- else if((err1 = _init_tc(tc, dbschema)) != MAL_SUCCEED)
- err2 = "Error while creating _temp_container: %s\n";
- else if((err1 = _loadfile(file_path, tc, dbschema, next_file_id)) !=
MAL_SUCCEED)
- {
- _free_temp_container(tc);
- err2 = "Error while loading BAM file: %s\n";
- }
- else if((err1 = _insert_into_vault(cntxt, tc)) != MAL_SUCCEED)
- {
- _free_temp_container(tc);
- err2 = "Error inserting data into database: %s\n";
- }
-
- if(err2 != NULL)
+ str err;
+ if((err = _bam_loader_file(cntxt, mb, file_paths[k], dbschema,
nr_threads)) != MAL_SUCCEED)
{
GDKfree(file_paths);
- throw(MAL, "bam_loader", err2, err1);
+ throw(MAL, "bam_loader_repos", "Error on loading file %s: \n",
file_paths[k]);
}
- _free_temp_container(tc);
-
- duration_sec = (GDKms() - start) / 1000.0f;
- snprintf(buf, bufsize, "Processed file '%s' in %f seconds = %f minutes
= %f hours", file_path, duration_sec, duration_sec/60, duration_sec/3600);
- _append_to_log(buf);
- GDKfree(buf);
}
fclose(_logfile);
GDKfree(file_paths);
+
+ return MAL_SUCCEED;
+}
+
+str
+bam_loader_file(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+{
+ str *file_path = (str*) getArgReference(stk,pci,pci->retc); /* arg 1:
repo_path */
+ sht dbschema = *(sht*) getArgReference(stk,pci,pci->retc+1); /* arg 2:
schema to use, see bam_loader.h for possible options */
+ sht nr_threads = *(sht*) getArgReference(stk,pci,pci->retc+2); /* arg
3: 1: no threads, >1: multi-threaded */
+
+ return _bam_loader_file(cntxt, mb, *file_path, dbschema, nr_threads);
+
+}
+
+static str
+_bam_loader_file(Client cntxt, MalBlkPtr mb, str file_path, sht dbschema, sht
nr_threads)
+{
+ int bufsize = 255;
+ str buf = GDKmalloc(bufsize * sizeof(char));
+
+ int start = GDKms();
+ float duration_sec;
+
+ sht next_file_id = 0;
+ str err1 = NULL, err2 = NULL;
+ _temp_container *tc = (_temp_container
*)GDKmalloc(sizeof(_temp_container));
+
+ if(buf == NULL || tc == NULL)
+ throw(MAL, "bam_loader", MAL_MALLOC_FAIL);
+
+ if((err1 = _next_file_id(cntxt, mb, &next_file_id)) != MAL_SUCCEED)
+ err2 = "Error while retrieving next file_id: %s\n";
+ else if((err1 = _init_tc(tc, dbschema)) != MAL_SUCCEED)
+ err2 = "Error while creating _temp_container: %s\n";
+ else if((err1 = _loadfile(file_path, tc, dbschema, next_file_id)) !=
MAL_SUCCEED)
+ {
+ _free_temp_container(tc);
+ err2 = "Error while loading BAM file: %s\n";
+ }
+ else if((err1 = _insert_into_vault(cntxt, tc)) != MAL_SUCCEED)
+ {
+ _free_temp_container(tc);
+ err2 = "Error inserting data into database: %s\n";
+ }
+
+ if(err2 != NULL)
+ throw(MAL, "bam_loader", err2, err1);
+ _free_temp_container(tc);
+
+ duration_sec = (GDKms() - start) / 1000.0f;
+ snprintf(buf, bufsize, "Processed file '%s' in %f seconds = %f minutes =
%f hours", file_path, duration_sec, duration_sec/60, duration_sec/3600);
+ _append_to_log(buf);
+ GDKfree(buf);
+
(void)nr_threads;
return MAL_SUCCEED;
}
-/*
- * Note: Copied directly from registrar.c
-
- * returns number of lines in a file.
- *
- * WARNING: always counts EOF as a line. So proper return is taken if the file
does
- * not have a newline at the end.
- */
-static lng
-_get_line_nr(str filename)
-{
- FILE *f;
- char c;
- lng lines = 0;
+/* File format specific functions */
- f = fopen(filename, "r");
-
- if(f == NULL)
- return 0;
-
- while((c = fgetc(f)) != EOF)
- if(c == '\n')
- lines++;
-
- fclose(f);
-
-/* if(c != '\n') */
-/* lines++; */
-
- return lines;
-}
-
-/*
- * Note: Copied directly from registrar.c
_______________________________________________
checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
