2 Bacula® - The Network Backup Solution
4 Copyright (C) 2003-2007 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 John Walker.
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 PostgreSQL
30 * These are PostgreSQL specific routines
32 * Dan Langille, December 2003
33 * based upon work done by Kern Sibbald, March 2000
39 /* The following is necessary so that we do not include
40 * the dummy external definition of DB.
42 #define __SQL_C /* indicate that this is sql.c */
47 #ifdef HAVE_POSTGRESQL
49 #include "postgres_ext.h" /* needed for NAMEDATALEN */
51 /* -----------------------------------------------------------------------
53 * PostgreSQL dependent defines and subroutines
55 * -----------------------------------------------------------------------
58 /* List of open databases */
59 static BQUEUE db_list = {&db_list, &db_list};
61 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
64 * 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)
85 Jmsg(jcr, M_FATAL, 0, _("A user name for PostgreSQL must be supplied.\n"));
88 P(mutex); /* lock DB queue */
89 if (!mult_db_connections) {
90 /* Look to see if DB already open */
91 for (mdb=NULL; (mdb=(B_DB *)qnext(&db_list, &mdb->bq)); ) {
92 if (bstrcmp(mdb->db_name, db_name) &&
93 bstrcmp(mdb->db_address, db_address) &&
94 mdb->db_port == db_port) {
95 Dmsg2(100, "DB REopen %d %s\n", mdb->ref_count, db_name);
98 return mdb; /* already open */
102 Dmsg0(100, "db_open first time\n");
103 mdb = (B_DB *)malloc(sizeof(B_DB));
104 memset(mdb, 0, sizeof(B_DB));
105 mdb->db_name = bstrdup(db_name);
106 mdb->db_user = bstrdup(db_user);
108 mdb->db_password = bstrdup(db_password);
111 mdb->db_address = bstrdup(db_address);
114 mdb->db_socket = bstrdup(db_socket);
116 mdb->db_port = db_port;
117 mdb->have_insert_id = TRUE;
118 mdb->errmsg = get_pool_memory(PM_EMSG); /* get error message buffer */
120 mdb->cmd = get_pool_memory(PM_EMSG); /* get command buffer */
121 mdb->cached_path = get_pool_memory(PM_FNAME);
122 mdb->cached_path_id = 0;
124 mdb->fname = get_pool_memory(PM_FNAME);
125 mdb->path = get_pool_memory(PM_FNAME);
126 mdb->esc_name = get_pool_memory(PM_FNAME);
127 mdb->esc_path = get_pool_memory(PM_FNAME);
128 mdb->allow_transactions = mult_db_connections;
129 qinsert(&db_list, &mdb->bq); /* put db in list */
135 * Now actually open the database. This can generate errors,
136 * which are returned in the errmsg
138 * DO NOT close the database or free(mdb) here !!!!
141 db_open_database(JCR *jcr, B_DB *mdb)
147 if (!PQisthreadsafe()) {
148 Jmsg(jcr, M_ABORT, 0, _("PostgreSQL configuration problem. "
149 "PostgreSQL library is not thread safe. Connot continue.\n"));
153 if (mdb->connected) {
157 mdb->connected = false;
159 if ((errstat=rwl_init(&mdb->lock)) != 0) {
161 Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
162 be.bstrerror(errstat));
168 bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
174 /* If connection fails, try at 5 sec intervals for 30 seconds. */
175 for (int retry=0; retry < 6; retry++) {
176 /* connect to the database */
177 mdb->db = PQsetdbLogin(
178 mdb->db_address, /* default = localhost */
179 port, /* default port */
180 NULL, /* pg options */
181 NULL, /* tty, ignored */
182 mdb->db_name, /* database name */
183 mdb->db_user, /* login name */
184 mdb->db_password); /* password */
186 /* If no connect, try once more in case it is a timing problem */
187 if (PQstatus(mdb->db) == CONNECTION_OK) {
193 Dmsg0(50, "pg_real_connect done\n");
194 Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
195 mdb->db_password==NULL?"(NULL)":mdb->db_password);
197 if (PQstatus(mdb->db) != CONNECTION_OK) {
198 Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server.\n"
199 "Database=%s User=%s\n"
200 "It is probably not running or your password is incorrect.\n"),
201 mdb->db_name, mdb->db_user);
206 mdb->connected = true;
208 if (!check_tables_version(jcr, mdb)) {
213 sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
220 db_close_database(JCR *jcr, B_DB *mdb)
225 db_end_transaction(jcr, mdb);
227 sql_free_result(mdb);
229 if (mdb->ref_count == 0) {
231 if (mdb->connected && mdb->db) {
234 rwl_destroy(&mdb->lock);
235 free_pool_memory(mdb->errmsg);
236 free_pool_memory(mdb->cmd);
237 free_pool_memory(mdb->cached_path);
238 free_pool_memory(mdb->fname);
239 free_pool_memory(mdb->path);
240 free_pool_memory(mdb->esc_name);
241 free_pool_memory(mdb->esc_path);
248 if (mdb->db_password) {
249 free(mdb->db_password);
251 if (mdb->db_address) {
252 free(mdb->db_address);
254 if (mdb->db_socket) {
255 free(mdb->db_socket);
262 void db_thread_cleanup()
266 * Return the next unique index (auto-increment) for
267 * the given table. Return NULL on error.
269 * For PostgreSQL, NULL causes the auto-increment value
272 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
274 strcpy(index, "NULL");
280 * Escape strings so that PostgreSQL is happy
282 * NOTE! len is the length of the old string. Your new
283 * string must be long enough (max 2*old+1) to hold
284 * the escaped output.
287 db_escape_string(char *snew, char *old, int len)
289 PQescapeString(snew, old, len);
293 * Submit a general SQL command (cmd), and for each row returned,
294 * the sqlite_handler is called with the ctx.
296 int db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
300 Dmsg0(500, "db_sql_query started\n");
303 if (sql_query(mdb, query) != 0) {
304 Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
306 Dmsg0(500, "db_sql_query failed\n");
309 Dmsg0(500, "db_sql_query succeeded. checking handler\n");
311 if (result_handler != NULL) {
312 Dmsg0(500, "db_sql_query invoking handler\n");
313 if ((mdb->result = sql_store_result(mdb)) != NULL) {
314 int num_fields = sql_num_fields(mdb);
316 Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
317 while ((row = sql_fetch_row(mdb)) != NULL) {
319 Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
320 if (result_handler(ctx, num_fields, row))
324 sql_free_result(mdb);
329 Dmsg0(500, "db_sql_query finished\n");
336 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
339 POSTGRESQL_ROW row = NULL; // by default, return NULL
341 Dmsg0(500, "my_postgresql_fetch_row start\n");
343 if (!mdb->row || mdb->row_size < mdb->num_fields) {
344 int num_fields = mdb->num_fields;
345 Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
348 Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
351 num_fields += 20; /* add a bit extra */
352 mdb->row = (POSTGRESQL_ROW)malloc(sizeof(char *) * num_fields);
353 mdb->row_size = num_fields;
355 // now reset the row_number now that we have the space allocated
359 // if still within the result set
360 if (mdb->row_number < mdb->num_rows) {
361 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
362 // get each value from this row
363 for (j = 0; j < mdb->num_fields; j++) {
364 mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
365 Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
367 // increment the row number for the next call
372 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
375 Dmsg1(500, "my_postgresql_fetch_row finishes returning %x\n", row);
380 int my_postgresql_max_length(B_DB *mdb, int field_num) {
382 // for a given column, find the max length
389 for (i = 0; i < mdb->num_rows; i++) {
390 if (PQgetisnull(mdb->result, i, field_num)) {
391 this_length = 4; // "NULL"
393 this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
396 if (max_length < this_length) {
397 max_length = this_length;
404 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
408 Dmsg0(500, "my_postgresql_fetch_field starts\n");
410 if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
414 Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
415 mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
416 mdb->fields_size = mdb->num_fields;
418 for (i = 0; i < mdb->num_fields; i++) {
419 Dmsg1(500, "filling field %d\n", i);
420 mdb->fields[i].name = PQfname(mdb->result, i);
421 mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
422 mdb->fields[i].type = PQftype(mdb->result, i);
423 mdb->fields[i].flags = 0;
425 Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
426 mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
427 mdb->fields[i].flags);
431 // increment field number for the next time around
433 Dmsg0(500, "my_postgresql_fetch_field finishes\n");
434 return &mdb->fields[mdb->field_number++];
437 void my_postgresql_data_seek(B_DB *mdb, int row)
439 // set the row number to be returned on the next call
440 // to my_postgresql_fetch_row
441 mdb->row_number = row;
444 void my_postgresql_field_seek(B_DB *mdb, int field)
446 mdb->field_number = field;
450 * Note, if this routine returns 1 (failure), Bacula expects
451 * that no result has been stored.
452 * This is where QUERY_DB comes with Postgresql.
454 * Returns: 0 on success
458 int my_postgresql_query(B_DB *mdb, const char *query)
460 Dmsg0(500, "my_postgresql_query started\n");
461 // We are starting a new query. reset everything.
463 mdb->row_number = -1;
464 mdb->field_number = -1;
467 PQclear(mdb->result); /* hmm, someone forgot to free?? */
471 Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
473 for (int i=0; i < 10; i++) {
474 mdb->result = PQexec(mdb->db, query);
481 Dmsg1(50, "Query failed: %s\n", query);
485 mdb->status = PQresultStatus(mdb->result);
486 if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
487 Dmsg1(500, "we have a result\n", query);
489 // how many fields in the set?
490 mdb->num_fields = (int)PQnfields(mdb->result);
491 Dmsg1(500, "we have %d fields\n", mdb->num_fields);
493 mdb->num_rows = PQntuples(mdb->result);
494 Dmsg1(500, "we have %d rows\n", mdb->num_rows);
496 mdb->status = 0; /* succeed */
498 Dmsg1(50, "Result status failed: %s\n", query);
502 Dmsg0(500, "my_postgresql_query finishing\n");
506 Dmsg1(500, "we failed\n", query);
507 PQclear(mdb->result);
509 mdb->status = 1; /* failed */
513 void my_postgresql_free_result(B_DB *mdb)
518 PQclear(mdb->result);
534 int my_postgresql_currval(B_DB *mdb, char *table_name)
536 // Obtain the current value of the sequence that
537 // provides the serial value for primary key of the table.
539 // currval is local to our session. It is not affected by
540 // other transactions.
542 // Determine the name of the sequence.
543 // PostgreSQL automatically creates a sequence using
544 // <table>_<column>_seq.
545 // At the time of writing, all tables used this format for
546 // for their primary key: <table>id
547 // Except for basefiles which has a primary key on baseid.
548 // Therefore, we need to special case that one table.
550 // everything else can use the PostgreSQL formula.
552 char sequence[NAMEDATALEN-1];
553 char query [NAMEDATALEN+50];
557 if (strcasecmp(table_name, "basefiles") == 0) {
558 bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
560 bstrncpy(sequence, table_name, sizeof(sequence));
561 bstrncat(sequence, "_", sizeof(sequence));
562 bstrncat(sequence, table_name, sizeof(sequence));
563 bstrncat(sequence, "id", sizeof(sequence));
566 bstrncat(sequence, "_seq", sizeof(sequence));
567 bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
569 Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
570 for (int i=0; i < 10; i++) {
571 result = PQexec(mdb->db, query);
578 Dmsg1(50, "Query failed: %s\n", query);
582 Dmsg0(500, "exec done");
584 if (PQresultStatus(result) == PGRES_TUPLES_OK) {
585 Dmsg0(500, "getting value");
586 id = atoi(PQgetvalue(result, 0, 0));
587 Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
589 Dmsg1(50, "Result status failed: %s\n", query);
590 Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
599 #ifdef HAVE_BATCH_FILE_INSERT
601 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
603 char *query = "COPY batch FROM STDIN";
605 Dmsg0(500, "my_postgresql_batch_start started\n");
607 if (my_postgresql_query(mdb,
608 " CREATE TEMPORARY TABLE batch "
614 " md5 varchar)") == 1)
616 Dmsg0(500, "my_postgresql_batch_start failed\n");
620 // We are starting a new query. reset everything.
622 mdb->row_number = -1;
623 mdb->field_number = -1;
625 my_postgresql_free_result(mdb);
627 for (int i=0; i < 10; i++) {
628 mdb->result = PQexec(mdb->db, query);
635 Dmsg1(50, "Query failed: %s\n", query);
639 mdb->status = PQresultStatus(mdb->result);
640 if (mdb->status == PGRES_COPY_IN) {
641 // how many fields in the set?
642 mdb->num_fields = (int) PQnfields(mdb->result);
646 Dmsg1(50, "Result status failed: %s\n", query);
650 Dmsg0(500, "my_postgresql_batch_start finishing\n");
656 PQclear(mdb->result);
661 /* set error to something to abort operation */
662 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
666 Dmsg0(500, "my_postgresql_batch_end started\n");
668 if (!mdb) { /* no files ? */
673 res = PQputCopyEnd(mdb->db, error);
674 } while (res == 0 && --count > 0);
682 Dmsg0(500, "we failed\n");
684 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
687 Dmsg0(500, "my_postgresql_batch_end finishing\n");
692 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
700 mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
701 my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
703 mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
704 my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
706 if (ar->Digest == NULL || ar->Digest[0] == 0) {
712 len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
713 ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
714 mdb->esc_name, ar->attr, digest);
717 res = PQputCopyData(mdb->db,
720 } while (res == 0 && --count > 0);
729 Dmsg0(500, "we failed\n");
731 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
734 Dmsg0(500, "my_postgresql_batch_insert finishing\n");
739 #endif /* HAVE_BATCH_FILE_INSERT */
742 * Escape strings so that PostgreSQL is happy on COPY
744 * NOTE! len is the length of the old string. Your new
745 * string must be long enough (max 2*old+1) to hold
746 * the escaped output.
748 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
750 /* we have to escape \t, \n, \r, \ */
753 while (len > 0 && *src) {
788 char *my_pg_batch_lock_path_query = "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
791 char *my_pg_batch_lock_filename_query = "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
793 char *my_pg_batch_unlock_tables_query = "COMMIT";
795 char *my_pg_batch_fill_path_query = "INSERT INTO Path (Path) "
796 " SELECT a.Path FROM "
797 " (SELECT DISTINCT Path FROM batch) AS a "
798 " WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
801 char *my_pg_batch_fill_filename_query = "INSERT INTO Filename (Name) "
802 " SELECT a.Name FROM "
803 " (SELECT DISTINCT Name FROM batch) as a "
805 " (SELECT Name FROM Filename WHERE Name = a.Name)";
806 #endif /* HAVE_POSTGRESQL */