View difference between Paste ID: 4hyTgNUU and zUnVc2Z2
SHOW: | | - or go back to the newest paste.
1
// Rationale: our inputs are huge (in the 30+gb range, possibly higher), and
2
// the original technique of reading the entire file into a string would not
3
// work on most computers (an aciss compute node has 72gb of ram), and we would
4
// need alternative methods. Current candidates are:
5
// 1. Memory mapping - platform dependent and no experience atm - done ~103k inserts per sec
6
// 2. Read in big ram chunks - performance might be affected
7
// 3. Capability dependent - the function checks for system memory size then
8-
// 		determines which method to use; e.g., input > memory, chunk. input < mem,
8+
// determines which method to use; e.g., input > memory, chunk. input < mem,
9-
//		read entire file into string. (appears to be the most flexible)
9+
// read entire file into string. (appears to be the most flexible)
10
// Boost memory mapped files seem to be the easiest and performant way
11
// to insert large files into the db.
12
int mm_read_and_insert(const string& filename, sqlite3* db)
13
{
14
	// create a file mapping from filename in readonly mode
15
	bi::file_mapping m_file(filename.c_str(),bi::read_only);
16
	// create the mapped region of the entire file in read only mode
17
	bi::mapped_region region(m_file,bi::read_only);
18
	region.advise(bi::mapped_region::advice_sequential); // inform the OS of our access pattern
19
	auto* addr = (char*)region.get_address(); // starting pointer
20
	auto size = region.get_size();
21
	pip::fastq_parser<const char*> g;
22
	
23
	if (size > 0) {
24-
	  auto startClock = clock();
24+
		auto startClock = clock();
25-
	  sqlite3_stmt *stmt;
25+
		sqlite3_stmt *stmt;
26
		char *sql_error_msg = 0;
27-
	// const char* insert_rawreads = "INSERT INTO rawreads VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13);";
27+
		// const char* insert_rawreads = "INSERT INTO rawreads VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13);";
28-
	  if (sqlite3_prepare_v2(db,sqlite::insert_rawreads,-1,&stmt, NULL) != SQLITE_OK) {
28+
	 	if (sqlite3_prepare_v2(db,sqlite::insert_rawreads,-1,&stmt, NULL) != SQLITE_OK) {
29-
	  	cout << sqlite3_errmsg(db) << endl;
29+
	  		cout << sqlite3_errmsg(db) << endl;
30
			return 0;
31-
	  } 
31+
	  	} 
32-
	  sqlite3_exec(db, "PRAGMA cache_size=400000;PRAGMA journal_mode=MEMORY;PRAGMA locking_mode=EXCLUSIVE;PRAGMA count_changes=OFF;PRAGMA auto_vacuum=NONE;PRAGMA temp_store = MEMORY;PRAGMA synchronous=OFF;BEGIN EXCLUSIVE TRANSACTION", NULL, NULL, &sql_error_msg);
32+
	 	sqlite3_exec(db, "PRAGMA cache_size=400000;PRAGMA journal_mode=MEMORY;PRAGMA locking_mode=EXCLUSIVE;PRAGMA count_changes=OFF;PRAGMA auto_vacuum=NONE;PRAGMA temp_store = MEMORY;PRAGMA synchronous=OFF;BEGIN EXCLUSIVE TRANSACTION", NULL, NULL, &sql_error_msg);
33
		unsigned long long n = 0; // we will be dealing with huge numbers here
34
		pip::fastq fq;
35
		char const* f(addr); // first iterator/pointer
36
		char const* l(f + size); // last iterator/pointer
37
		while (parse(f,l,g,fq)) {
38
			pack::Pack packed(fq.sequence,fq.quality,1);
39
			// Bind parameters to sequence data
40
			sqlite3_bind_text(stmt,1,fq.instrument.c_str(),-1,SQLITE_TRANSIENT);
41
			sqlite3_bind_int(stmt,2,fq.run);
42
			sqlite3_bind_text(stmt,3,fq.flowcell.c_str(),-1,SQLITE_TRANSIENT);
43
			sqlite3_bind_int(stmt,4,fq.lane);
44
			sqlite3_bind_int(stmt,5,fq.tile);
45
			sqlite3_bind_int(stmt,6,fq.x);
46
			sqlite3_bind_int(stmt,7,fq.y);
47
			sqlite3_bind_int(stmt,8,fq.pair);
48
			sqlite3_bind_int(stmt,9,fq.filter == 'Y' ? 1:0);
49
			sqlite3_bind_int(stmt,10,fq.control);
50
			sqlite3_bind_text(stmt,11,fq.index.c_str(),-1,SQLITE_TRANSIENT);
51
			sqlite3_bind_int(stmt,12,packed.qualityFormat());
52-
	    sqlite3_bind_blob(stmt,13,packed.rawData(),fq.sequence.length(),SQLITE_TRANSIENT);
52+
	    		sqlite3_bind_blob(stmt,13,packed.rawData(),fq.sequence.length(),SQLITE_TRANSIENT);
53
			sqlite3_step(stmt);
54-
	    sqlite3_reset(stmt);
54+
	   		sqlite3_reset(stmt);
55
			++n;
56
			fq = {};
57
			// Show an update every 100k inserts
58
			if (n % 100000 == 0)
59
				cerr << "Inserted " << n << " sequences...\r";
60
		}
61-
	  sqlite3_exec(db, "COMMIT TRANSACTION", NULL, NULL, &sql_error_msg);
61+
	 	sqlite3_exec(db, "COMMIT TRANSACTION", NULL, NULL, &sql_error_msg);
62
		sqlite3_finalize(stmt);
63
		auto endClock = clock() - startClock;
64
		printf("Pip: %llu sequences imported in %4.2f seconds\n",n,endClock/(double)CLOCKS_PER_SEC);
65
		return n; // return number of rows inserted
66
	}
67
	return 0;
68
}