2 Bacula® - The Network Backup Solution
4 Copyright (C) 2003-2010 Free Software Foundation Europe e.V.
6 The main author of Bacula is Kern Sibbald, with contributions from
7 many others, a complete list can be found in the file AUTHORS.
8 This program is Free Software; you can redistribute it and/or
9 modify it under the terms of version two of the GNU General Public
10 License as published by the Free Software Foundation and included
13 This program is distributed in the hope that it will be useful, but
14 WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; if not, write to the Free Software
20 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
23 Bacula® is a registered trademark of Kern Sibbald.
24 The licensor of Bacula is the Free Software Foundation Europe
25 (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
26 Switzerland, email:ftf@fsfeurope.org.
29 * Bacula Catalog Database routines specific to Ingres
30 * These are Ingres specific routines
32 * Stefan Reddig, June 2009 with help of Marco van Wieringen April 2010
33 * based uopn work done
34 * by Dan Langille, December 2003 and
35 * by Kern Sibbald, March 2000
40 /* The following is necessary so that we do not include
41 * the dummy external definition of DB.
43 #define __SQL_C /* indicate that this is sql.c */
52 /* -----------------------------------------------------------------------
54 * Ingres dependent defines and subroutines
56 * -----------------------------------------------------------------------
59 /* List of open databases */
60 static BQUEUE db_list = {&db_list, &db_list};
62 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
65 * Retrieve database type
74 * Initialize database data structure. In principal this should
75 * never have errors, or it is really fatal.
78 db_init_database(JCR *jcr, const char *db_name, const char *db_user, const char *db_password,
79 const char *db_address, int db_port, const char *db_socket,
80 int mult_db_connections)
83 int next_session_id = 0;
86 Jmsg(jcr, M_FATAL, 0, _("A user name for Ingres must be supplied.\n"));
89 P(mutex); /* lock DB queue */
90 if (!mult_db_connections) {
91 /* Look to see if DB already open */
92 for (mdb=NULL; (mdb=(B_DB *)qnext(&db_list, &mdb->bq)); ) {
93 if (bstrcmp(mdb->db_name, db_name) &&
94 bstrcmp(mdb->db_address, db_address) &&
95 mdb->db_port == db_port) {
96 Dmsg2(100, "DB REopen %d %s\n", mdb->ref_count, db_name);
99 return mdb; /* already open */
102 if (mdb->session_id > next_session_id) {
103 next_session_id = mdb->session_id;
108 * See what the next available session_id is.
109 * We first see what the highest session_id is used now.
111 foreach_dlist(mdb, db_list) {
112 if (mdb->session_id > next_session_id) {
113 next_session_id = mdb->session_id;
117 Dmsg0(100, "db_open first time\n");
118 mdb = (B_DB *)malloc(sizeof(B_DB));
119 memset(mdb, 0, sizeof(B_DB));
120 mdb->db_name = bstrdup(db_name);
121 mdb->db_user = bstrdup(db_user);
123 mdb->db_password = bstrdup(db_password);
126 mdb->db_address = bstrdup(db_address);
129 mdb->db_socket = bstrdup(db_socket);
131 mdb->db_port = db_port;
132 mdb->session_id = ++next_session_id;
133 mdb->errmsg = get_pool_memory(PM_EMSG); /* get error message buffer */
135 mdb->cmd = get_pool_memory(PM_EMSG); /* get command buffer */
136 mdb->cached_path = get_pool_memory(PM_FNAME);
137 mdb->cached_path_id = 0;
139 mdb->fname = get_pool_memory(PM_FNAME);
140 mdb->path = get_pool_memory(PM_FNAME);
141 mdb->esc_name = get_pool_memory(PM_FNAME);
142 mdb->esc_path = get_pool_memory(PM_FNAME);
143 mdb->allow_transactions = mult_db_connections;
144 mdb->limit_filter = new_bregexp("/LIMIT ([0-9]+)/FETCH FIRST $1 ROW ONLY/g");
145 qinsert(&db_list, &mdb->bq); /* put db in list */
150 /* Check that the database correspond to the encoding we want */
151 static bool check_database_encoding(JCR *jcr, B_DB *mdb)
153 /* SRE: TODO! Needed?
157 if (!db_sql_query(mdb, "SELECT getdatabaseencoding()", NULL, NULL)) {
158 Jmsg(jcr, M_ERROR, 0, "%s", mdb->errmsg);
162 if ((row = sql_fetch_row(mdb)) == NULL) {
163 Mmsg1(mdb->errmsg, _("error fetching row: %s\n"), sql_strerror(mdb));
164 Jmsg(jcr, M_ERROR, 0, "Can't check database encoding %s", mdb->errmsg);
166 ret = bstrcmp(row[0], "SQL_ASCII");
169 _("Encoding error for database \"%s\". Wanted SQL_ASCII, got %s\n"),
170 mdb->db_name, row[0]);
171 Jmsg(jcr, M_WARNING, 0, "%s", mdb->errmsg);
172 Dmsg1(50, "%s", mdb->errmsg);
181 * Now actually open the database. This can generate errors,
182 * which are returned in the errmsg
184 * DO NOT close the database or free(mdb) here !!!!
187 db_open_database(JCR *jcr, B_DB *mdb)
193 if (mdb->connected) {
197 mdb->connected = false;
199 if ((errstat=rwl_init(&mdb->lock)) != 0) {
201 Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
202 be.bstrerror(errstat));
208 bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
214 mdb->db = INGconnectDB(mdb->db_name, mdb->db_user, mdb->db_password, mdb->session_id);
216 Dmsg0(50, "Ingres real CONNECT done\n");
217 Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
218 mdb->db_password==NULL?"(NULL)":mdb->db_password);
221 Mmsg2(&mdb->errmsg, _("Unable to connect to Ingres server.\n"
222 "Database=%s User=%s\n"
223 "It is probably not running or your password is incorrect.\n"),
224 mdb->db_name, mdb->db_user);
229 mdb->connected = true;
231 if (!check_tables_version(jcr, mdb)) {
236 //sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
238 /* check that encoding is SQL_ASCII */
239 check_database_encoding(jcr, mdb);
246 db_close_database(JCR *jcr, B_DB *mdb)
251 db_end_transaction(jcr, mdb);
253 sql_free_result(mdb);
255 if (mdb->ref_count == 0) {
257 if (mdb->connected && mdb->db) {
260 rwl_destroy(&mdb->lock);
261 free_pool_memory(mdb->errmsg);
262 free_pool_memory(mdb->cmd);
263 free_pool_memory(mdb->cached_path);
264 free_pool_memory(mdb->fname);
265 free_pool_memory(mdb->path);
266 free_pool_memory(mdb->esc_name);
267 free_pool_memory(mdb->esc_path);
268 free_bregexp(mdb->limit_filter);
275 if (mdb->db_password) {
276 free(mdb->db_password);
278 if (mdb->db_address) {
279 free(mdb->db_address);
281 if (mdb->db_socket) {
282 free(mdb->db_socket);
289 void db_check_backend_thread_safe()
294 void db_thread_cleanup()
298 * Return the next unique index (auto-increment) for
299 * the given table. Return NULL on error.
301 * For Ingres, NULL causes the auto-increment value SRE: true?
304 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
306 strcpy(index, "NULL");
312 * Escape strings so that Ingres is happy
314 * NOTE! len is the length of the old string. Your new
315 * string must be long enough (max 2*old+1) to hold
316 * the escaped output.
319 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
346 * Submit a general SQL command (cmd), and for each row returned,
347 * the sqlite_handler is called with the ctx.
349 bool db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
353 Dmsg0(500, "db_sql_query started\n");
356 if (sql_query(mdb, query) != 0) {
357 Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
359 Dmsg0(500, "db_sql_query failed\n");
362 Dmsg0(500, "db_sql_query succeeded. checking handler\n");
364 if (result_handler != NULL) {
365 Dmsg0(500, "db_sql_query invoking handler\n");
366 if (mdb->result != NULL) {
367 int num_fields = sql_num_fields(mdb);
369 Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
370 while ((row = sql_fetch_row(mdb)) != NULL) {
372 Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
373 if (result_handler(ctx, num_fields, row))
377 sql_free_result(mdb);
382 Dmsg0(500, "db_sql_query finished\n");
388 * Close database connection
390 void my_ingres_close(B_DB *mdb)
392 Dmsg0(500, "my_ingres_close closing database connection\n");
393 INGdisconnectDB(mdb->db);
394 //SRE: error handling?
397 INGRES_ROW my_ingres_fetch_row(B_DB *mdb)
400 INGRES_ROW row = NULL; // by default, return NULL
405 if (mdb->result->num_rows <= 0) {
409 Dmsg0(500, "my_ingres_fetch_row start\n");
411 if (!mdb->row || mdb->row_size < mdb->num_fields) {
412 int num_fields = mdb->num_fields;
413 Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
416 Dmsg0(500, "my_ingres_fetch_row freeing space\n");
419 num_fields += 20; /* add a bit extra */
420 mdb->row = (INGRES_ROW)malloc(sizeof(char *) * num_fields);
421 mdb->row_size = num_fields;
423 // now reset the row_number now that we have the space allocated
424 //mdb->row_number = 1;
428 // if still within the result set
429 //if (mdb->row_number <= mdb->num_rows) {
430 if (mdb->row_number < mdb->num_rows) {
431 Dmsg2(500, "my_ingres_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
432 // get each value from this row
433 for (j = 0; j < mdb->num_fields; j++) {
434 mdb->row[j] = INGgetvalue(mdb->result, mdb->row_number, j);
435 Dmsg2(500, "my_ingres_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
437 // increment the row number for the next call
442 Dmsg2(500, "my_ingres_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
445 Dmsg1(500, "my_ingres_fetch_row finishes returning %p\n", row);
451 int my_ingres_max_length(B_DB *mdb, int field_num) {
453 // for a given column, find the max length
460 for (i = 0; i < mdb->num_rows; i++) {
461 if (INGgetisnull(mdb->result, i, field_num)) {
462 this_length = 4; // "NULL"
464 this_length = cstrlen(INGgetvalue(mdb->result, i, field_num));
467 if (max_length < this_length) {
468 max_length = this_length;
475 INGRES_FIELD * my_ingres_fetch_field(B_DB *mdb)
479 Dmsg0(500, "my_ingres_fetch_field starts\n");
481 if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
485 Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
486 mdb->fields = (INGRES_FIELD *)malloc(sizeof(INGRES_FIELD) * mdb->num_fields);
487 mdb->fields_size = mdb->num_fields;
489 for (i = 0; i < mdb->num_fields; i++) {
490 Dmsg1(500, "filling field %d\n", i);
491 strcpy(mdb->fields[i].name,INGfname(mdb->result, i));
492 mdb->fields[i].max_length = my_ingres_max_length(mdb, i);
493 mdb->fields[i].type = INGftype(mdb->result, i);
494 mdb->fields[i].flags = 0;
496 Dmsg4(500, "my_ingres_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
497 mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
498 mdb->fields[i].flags);
502 // increment field number for the next time around
504 Dmsg0(500, "my_ingres_fetch_field finishes\n");
505 return &mdb->fields[mdb->field_number++];
508 void my_ingres_data_seek(B_DB *mdb, int row)
510 // set the row number to be returned on the next call
511 // to my_ingres_fetch_row
512 mdb->row_number = row;
515 void my_ingres_field_seek(B_DB *mdb, int field)
517 mdb->field_number = field;
521 * Note, if this routine returns 1 (failure), Bacula expects
522 * that no result has been stored.
524 * Returns: 0 on success
528 int my_ingres_query(B_DB *mdb, const char *query)
530 const char *new_query;
532 if (strstr(query, "LIMIT") != NULL) {
533 new_query = mdb->limit_filter->replace(query);
538 Dmsg0(500, "my_ingres_query started\n");
539 // We are starting a new query. reset everything.
541 mdb->row_number = -1;
542 mdb->field_number = -1;
547 INGclear(mdb->result); /* hmm, someone forgot to free?? */
551 Dmsg1(500, "my_ingres_query starts with '%s'\n", new_query);
554 * See if we are getting a transaction start or end.
556 if (!strcasecmp(new_query, "BEGIN")) {
558 * Start of a transaction.
560 Dmsg0(500,"my_ingres_query: Start of transaction\n");
561 mdb->transaction = true;
563 } else if (!strcasecmp(new_query, "COMMIT")) {
565 * End of a transaction.
567 Dmsg0(500,"my_ingres_query: End of transaction, commiting work\n");
568 mdb->transaction = false;
573 /* TODO: differentiate between SELECTs and other queries */
575 if ((cols = INGgetCols(mdb->db, new_query, mdb->transaction)) <= 0) {
577 Dmsg0(500,"my_ingres_query: neg.columns: no DML stmt!\n");
579 Dmsg0(500,"my_ingres_query (non SELECT) starting...\n");
581 mdb->num_rows = INGexec(mdb->db, new_query, mdb->transaction);
583 Dmsg0(500,"my_ingres_query (non SELECT) went wrong\n");
586 Dmsg0(500,"my_ingres_query (non SELECT) seems ok\n");
591 Dmsg0(500,"my_ingres_query (SELECT) starting...\n");
592 mdb->result = INGquery(mdb->db, new_query, mdb->transaction);
593 if (mdb->result != NULL) {
594 Dmsg1(500, "we have a result\n", new_query);
596 // how many fields in the set?
597 mdb->num_fields = (int)INGnfields(mdb->result);
598 Dmsg1(500, "we have %d fields\n", mdb->num_fields);
600 mdb->num_rows = INGntuples(mdb->result);
601 Dmsg1(500, "we have %d rows\n", mdb->num_rows);
603 mdb->status = 0; /* succeed */
605 Dmsg0(500, "No resultset...\n");
606 mdb->status = 1; /* failed */
610 Dmsg0(500, "my_ingres_query finishing\n");
614 void my_ingres_free_result(B_DB *mdb)
619 INGclear(mdb->result);
635 static int my_ingres_currval(B_DB *mdb, const char *table_name)
638 * Obtain the current value of the sequence that
639 * provides the serial value for primary key of the table.
641 * currval is local to our session. It is not affected by
642 * other transactions.
644 * Determine the name of the sequence.
645 * As we name all sequences as <table>_seq this is easy.
653 bstrncpy(sequence, table_name, sizeof(sequence));
654 bstrncat(sequence, "_seq", sizeof(sequence));
656 bsnprintf(query, sizeof(query), "SELECT %s.currval FROM %s", sequence, table_name);
658 Dmsg1(500, "my_ingres_currval invoked with '%s'\n", query);
660 result = INGquery(mdb->db, query, mdb->transaction);
663 Dmsg1(50, "Query failed: %s\n", query);
667 Dmsg0(500, "exec done");
669 id = atoi(INGgetvalue(result, 0, 0));
677 int my_ingres_insert_id(B_DB *mdb, const char *query, const char *table_name)
680 * First execute the insert query and then retrieve the currval.
681 * By setting transaction to true we make it an atomic transaction
682 * and as such we can get the currval after which we commit if
683 * mdb->transaction is false. This way its an atomic operation for
684 * Ingres and things work.
686 mdb->num_rows = INGexec(mdb->db, query, true);
693 return my_ingres_currval(mdb, table_name);
696 #ifdef HAVE_BATCH_FILE_INSERT
697 int my_ingres_batch_start(JCR *jcr, B_DB *mdb)
702 ok = db_sql_query(mdb,
703 "DECLARE GLOBAL TEMPORARY TABLE batch ("
708 "LStat varchar(256),"
710 " ON COMMIT PRESERVE ROWS WITH NORECOVERY",NULL, NULL);
715 int my_ingres_batch_end(JCR *jcr, B_DB *mdb, const char *error)
723 int my_ingres_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
729 mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
730 db_escape_string(jcr, mdb, mdb->esc_name, mdb->fname, mdb->fnl);
732 mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
733 db_escape_string(jcr, mdb, mdb->esc_path, mdb->path, mdb->pnl);
735 if (ar->Digest == NULL || ar->Digest[0] == 0) {
741 len = Mmsg(mdb->cmd, "INSERT INTO batch VALUES (%u,%s,'%s','%s','%s','%s')",
742 ar->FileIndex, edit_int64(ar->JobId,ed1), mdb->esc_path,
743 mdb->esc_name, ar->attr, digest);
745 return INSERT_DB(jcr, mdb, mdb->cmd);
747 #endif /* HAVE_BATCH_FILE_INSERT */
750 * Escape strings so that Ingres is happy on COPY
752 * NOTE! len is the length of the old string. Your new
753 * string must be long enough (max 2*old+1) to hold
754 * the escaped output.
756 char *my_ingres_copy_escape(char *dest, char *src, size_t len)
758 /* we have to escape \t, \n, \r, \ */
761 while (len > 0 && *src) {
796 #ifdef HAVE_BATCH_FILE_INSERT
797 const char *my_ingres_batch_lock_path_query = "BEGIN";
799 const char *my_ingres_batch_lock_filename_query = "BEGIN";
801 const char *my_ingres_batch_unlock_tables_query = "COMMIT";
803 const char *my_ingres_batch_fill_path_query =
804 "INSERT INTO Path (Path) "
805 "SELECT a.Path FROM "
806 "(SELECT DISTINCT Path FROM batch) AS a "
807 "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
810 const char *my_ingres_batch_fill_filename_query =
811 "INSERT INTO Filename (Name) "
812 "SELECT a.Name FROM "
813 "(SELECT DISTINCT Name FROM batch) as a "
815 "(SELECT Name FROM Filename WHERE Name = a.Name)";
816 #endif /* HAVE_BATCH_FILE_INSERT */
818 #endif /* HAVE_INGRES */