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 plus additions
11 that are listed in the file LICENSE.
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 (mdb->connected) {
151 mdb->connected = false;
153 if ((errstat=rwl_init(&mdb->lock)) != 0) {
154 Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
161 bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
167 /* If connection fails, try at 5 sec intervals for 30 seconds. */
168 for (int retry=0; retry < 6; retry++) {
169 /* connect to the database */
170 mdb->db = PQsetdbLogin(
171 mdb->db_address, /* default = localhost */
172 port, /* default port */
173 NULL, /* pg options */
174 NULL, /* tty, ignored */
175 mdb->db_name, /* database name */
176 mdb->db_user, /* login name */
177 mdb->db_password); /* password */
179 /* If no connect, try once more in case it is a timing problem */
180 if (PQstatus(mdb->db) == CONNECTION_OK) {
186 Dmsg0(50, "pg_real_connect done\n");
187 Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
188 mdb->db_password==NULL?"(NULL)":mdb->db_password);
190 if (PQstatus(mdb->db) != CONNECTION_OK) {
191 Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server.\n"
192 "Database=%s User=%s\n"
193 "It is probably not running or your password is incorrect.\n"),
194 mdb->db_name, mdb->db_user);
199 mdb->connected = true;
201 if (!check_tables_version(jcr, mdb)) {
206 sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
213 db_close_database(JCR *jcr, B_DB *mdb)
218 db_end_transaction(jcr, mdb);
220 sql_free_result(mdb);
222 if (mdb->ref_count == 0) {
224 if (mdb->connected && mdb->db) {
227 rwl_destroy(&mdb->lock);
228 free_pool_memory(mdb->errmsg);
229 free_pool_memory(mdb->cmd);
230 free_pool_memory(mdb->cached_path);
231 free_pool_memory(mdb->fname);
232 free_pool_memory(mdb->path);
233 free_pool_memory(mdb->esc_name);
234 free_pool_memory(mdb->esc_path);
241 if (mdb->db_password) {
242 free(mdb->db_password);
244 if (mdb->db_address) {
245 free(mdb->db_address);
247 if (mdb->db_socket) {
248 free(mdb->db_socket);
250 my_postgresql_free_result(mdb);
256 void db_thread_cleanup()
260 * Return the next unique index (auto-increment) for
261 * the given table. Return NULL on error.
263 * For PostgreSQL, NULL causes the auto-increment value
266 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
268 strcpy(index, "NULL");
274 * Escape strings so that PostgreSQL is happy
276 * NOTE! len is the length of the old string. Your new
277 * string must be long enough (max 2*old+1) to hold
278 * the escaped output.
281 db_escape_string(char *snew, char *old, int len)
283 PQescapeString(snew, old, len);
287 * Submit a general SQL command (cmd), and for each row returned,
288 * the sqlite_handler is called with the ctx.
290 int db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
294 Dmsg0(500, "db_sql_query started\n");
297 if (sql_query(mdb, query) != 0) {
298 Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
300 Dmsg0(500, "db_sql_query failed\n");
303 Dmsg0(500, "db_sql_query succeeded. checking handler\n");
305 if (result_handler != NULL) {
306 Dmsg0(500, "db_sql_query invoking handler\n");
307 if ((mdb->result = sql_store_result(mdb)) != NULL) {
308 int num_fields = sql_num_fields(mdb);
310 Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
311 while ((row = sql_fetch_row(mdb)) != NULL) {
313 Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
314 if (result_handler(ctx, num_fields, row))
318 sql_free_result(mdb);
323 Dmsg0(500, "db_sql_query finished\n");
330 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
333 POSTGRESQL_ROW row = NULL; // by default, return NULL
335 Dmsg0(500, "my_postgresql_fetch_row start\n");
337 if (mdb->row_number == -1 || mdb->row == NULL) {
338 Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
340 if (mdb->row != NULL) {
341 Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
346 mdb->row = (POSTGRESQL_ROW) malloc(sizeof(char *) * mdb->num_fields);
348 // now reset the row_number now that we have the space allocated
352 // if still within the result set
353 if (mdb->row_number < mdb->num_rows) {
354 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
355 // get each value from this row
356 for (j = 0; j < mdb->num_fields; j++) {
357 mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
358 Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
360 // increment the row number for the next call
365 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
368 Dmsg1(500, "my_postgresql_fetch_row finishes returning %x\n", row);
373 int my_postgresql_max_length(B_DB *mdb, int field_num) {
375 // for a given column, find the max length
382 for (i = 0; i < mdb->num_rows; i++) {
383 if (PQgetisnull(mdb->result, i, field_num)) {
384 this_length = 4; // "NULL"
386 this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
389 if (max_length < this_length) {
390 max_length = this_length;
397 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
401 Dmsg0(500, "my_postgresql_fetch_field starts\n");
402 if (mdb->fields == NULL) {
403 Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
404 mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
406 for (i = 0; i < mdb->num_fields; i++) {
407 Dmsg1(500, "filling field %d\n", i);
408 mdb->fields[i].name = PQfname(mdb->result, i);
409 mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
410 mdb->fields[i].type = PQftype(mdb->result, i);
411 mdb->fields[i].flags = 0;
413 Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
414 mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
415 mdb->fields[i].flags);
419 // increment field number for the next time around
421 Dmsg0(500, "my_postgresql_fetch_field finishes\n");
422 return &mdb->fields[mdb->field_number++];
425 void my_postgresql_data_seek(B_DB *mdb, int row)
427 // set the row number to be returned on the next call
428 // to my_postgresql_fetch_row
429 mdb->row_number = row;
432 void my_postgresql_field_seek(B_DB *mdb, int field)
434 mdb->field_number = field;
438 * Note, if this routine returns 1 (failure), Bacula expects
439 * that no result has been stored.
440 * This is where QUERY_DB comes with Postgresql.
442 * Returns: 0 on success
446 int my_postgresql_query(B_DB *mdb, const char *query)
448 Dmsg0(500, "my_postgresql_query started\n");
449 // We are starting a new query. reset everything.
451 mdb->row_number = -1;
452 mdb->field_number = -1;
455 PQclear(mdb->result); /* hmm, someone forgot to free?? */
459 Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
461 for (int i=0; i < 10; i++) {
462 mdb->result = PQexec(mdb->db, query);
472 mdb->status = PQresultStatus(mdb->result);
473 if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
474 Dmsg1(500, "we have a result\n", query);
476 // how many fields in the set?
477 mdb->num_fields = (int)PQnfields(mdb->result);
478 Dmsg1(500, "we have %d fields\n", mdb->num_fields);
480 mdb->num_rows = PQntuples(mdb->result);
481 Dmsg1(500, "we have %d rows\n", mdb->num_rows);
483 mdb->status = 0; /* succeed */
488 Dmsg0(500, "my_postgresql_query finishing\n");
492 Dmsg1(500, "we failed\n", query);
493 PQclear(mdb->result);
495 mdb->status = 1; /* failed */
499 void my_postgresql_free_result(B_DB *mdb)
502 PQclear(mdb->result);
517 int my_postgresql_currval(B_DB *mdb, char *table_name)
519 // Obtain the current value of the sequence that
520 // provides the serial value for primary key of the table.
522 // currval is local to our session. It is not affected by
523 // other transactions.
525 // Determine the name of the sequence.
526 // PostgreSQL automatically creates a sequence using
527 // <table>_<column>_seq.
528 // At the time of writing, all tables used this format for
529 // for their primary key: <table>id
530 // Except for basefiles which has a primary key on baseid.
531 // Therefore, we need to special case that one table.
533 // everything else can use the PostgreSQL formula.
535 char sequence[NAMEDATALEN-1];
536 char query [NAMEDATALEN+50];
540 if (strcasecmp(table_name, "basefiles") == 0) {
541 bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
543 bstrncpy(sequence, table_name, sizeof(sequence));
544 bstrncat(sequence, "_", sizeof(sequence));
545 bstrncat(sequence, table_name, sizeof(sequence));
546 bstrncat(sequence, "id", sizeof(sequence));
549 bstrncat(sequence, "_seq", sizeof(sequence));
550 bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
552 // Mmsg(query, "SELECT currval('%s')", sequence);
553 Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
554 result = PQexec(mdb->db, query);
556 Dmsg0(500, "exec done");
558 if (PQresultStatus(result) == PGRES_TUPLES_OK) {
559 Dmsg0(500, "getting value");
560 id = atoi(PQgetvalue(result, 0, 0));
561 Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
563 Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
571 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
573 Dmsg0(500, "my_postgresql_batch_start started\n");
575 if (my_postgresql_query(mdb,
576 " CREATE TEMPORARY TABLE batch "
582 " md5 varchar)") == 1)
584 Dmsg0(500, "my_postgresql_batch_start failed\n");
588 // We are starting a new query. reset everything.
590 mdb->row_number = -1;
591 mdb->field_number = -1;
593 if (mdb->result != NULL) {
594 my_postgresql_free_result(mdb);
597 mdb->result = PQexec(mdb->db, "COPY batch FROM STDIN");
598 mdb->status = PQresultStatus(mdb->result);
599 if (mdb->status == PGRES_COPY_IN) {
600 // how many fields in the set?
601 mdb->num_fields = (int) PQnfields(mdb->result);
605 Dmsg0(500, "we failed\n");
609 Dmsg0(500, "my_postgresql_batch_start finishing\n");
614 /* set error to something to abort operation */
615 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
619 Dmsg0(500, "my_postgresql_batch_end started\n");
621 if (!mdb) { /* no files ? */
626 res = PQputCopyEnd(mdb->db, error);
627 } while (res == 0 && --count > 0);
635 Dmsg0(500, "we failed\n");
637 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
640 Dmsg0(500, "my_postgresql_batch_end finishing\n");
645 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
653 mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
654 my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
656 mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
657 my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
659 if (ar->Digest == NULL || ar->Digest[0] == 0) {
665 len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
666 ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
667 mdb->esc_name, ar->attr, digest);
670 res = PQputCopyData(mdb->db,
673 } 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_insert finishing\n");
693 * Escape strings so that PostgreSQL is happy on COPY
695 * NOTE! len is the length of the old string. Your new
696 * string must be long enough (max 2*old+1) to hold
697 * the escaped output.
699 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
701 /* we have to escape \t, \n, \r, \ */
704 while (len > 0 && *src) {
739 char *my_pg_batch_lock_path_query = "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
742 char *my_pg_batch_lock_filename_query = "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
744 char *my_pg_batch_unlock_tables_query = "COMMIT";
746 char *my_pg_batch_fill_path_query = "INSERT INTO Path (Path) "
747 " SELECT a.Path FROM "
748 " (SELECT DISTINCT Path FROM batch) AS a "
749 " WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
752 char *my_pg_batch_fill_filename_query = "INSERT INTO Filename (Name) "
753 " SELECT a.Name FROM "
754 " (SELECT DISTINCT Name FROM batch) as a "
756 " (SELECT Name FROM Filename WHERE Name = a.Name)";
757 #endif /* HAVE_POSTGRESQL */