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'");
215 /* tell PostgreSQL we are using standard conforming strings
216 and avoid warnings such as:
217 WARNING: nonstandard use of \\ in a string literal
219 sql_query(mdb, "set standard_conforming_strings=on");
226 db_close_database(JCR *jcr, B_DB *mdb)
231 db_end_transaction(jcr, mdb);
233 sql_free_result(mdb);
235 if (mdb->ref_count == 0) {
237 if (mdb->connected && mdb->db) {
240 rwl_destroy(&mdb->lock);
241 free_pool_memory(mdb->errmsg);
242 free_pool_memory(mdb->cmd);
243 free_pool_memory(mdb->cached_path);
244 free_pool_memory(mdb->fname);
245 free_pool_memory(mdb->path);
246 free_pool_memory(mdb->esc_name);
247 free_pool_memory(mdb->esc_path);
254 if (mdb->db_password) {
255 free(mdb->db_password);
257 if (mdb->db_address) {
258 free(mdb->db_address);
260 if (mdb->db_socket) {
261 free(mdb->db_socket);
268 void db_thread_cleanup()
272 * Return the next unique index (auto-increment) for
273 * the given table. Return NULL on error.
275 * For PostgreSQL, NULL causes the auto-increment value
278 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
280 strcpy(index, "NULL");
286 * Escape strings so that PostgreSQL is happy
288 * NOTE! len is the length of the old string. Your new
289 * string must be long enough (max 2*old+1) to hold
290 * the escaped output.
293 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
297 PQescapeStringConn(mdb->db, snew, old, len, &error);
299 Jmsg(jcr, M_FATAL, 0, _("PQescapeStringConn returned non-zero.\n"));
300 /* error on encoding, probably invalid multibyte encoding in the source string
301 see PQescapeStringConn documentation for details. */
302 Dmsg0(500, "PQescapeStringConn failed\n");
307 * Submit a general SQL command (cmd), and for each row returned,
308 * the sqlite_handler is called with the ctx.
310 int db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
314 Dmsg0(500, "db_sql_query started\n");
317 if (sql_query(mdb, query) != 0) {
318 Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
320 Dmsg0(500, "db_sql_query failed\n");
323 Dmsg0(500, "db_sql_query succeeded. checking handler\n");
325 if (result_handler != NULL) {
326 Dmsg0(500, "db_sql_query invoking handler\n");
327 if ((mdb->result = sql_store_result(mdb)) != NULL) {
328 int num_fields = sql_num_fields(mdb);
330 Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
331 while ((row = sql_fetch_row(mdb)) != NULL) {
333 Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
334 if (result_handler(ctx, num_fields, row))
338 sql_free_result(mdb);
343 Dmsg0(500, "db_sql_query finished\n");
350 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
353 POSTGRESQL_ROW row = NULL; // by default, return NULL
355 Dmsg0(500, "my_postgresql_fetch_row start\n");
357 if (!mdb->row || mdb->row_size < mdb->num_fields) {
358 int num_fields = mdb->num_fields;
359 Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
362 Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
365 num_fields += 20; /* add a bit extra */
366 mdb->row = (POSTGRESQL_ROW)malloc(sizeof(char *) * num_fields);
367 mdb->row_size = num_fields;
369 // now reset the row_number now that we have the space allocated
373 // if still within the result set
374 if (mdb->row_number < mdb->num_rows) {
375 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
376 // get each value from this row
377 for (j = 0; j < mdb->num_fields; j++) {
378 mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
379 Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
381 // increment the row number for the next call
386 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
389 Dmsg1(500, "my_postgresql_fetch_row finishes returning %x\n", row);
394 int my_postgresql_max_length(B_DB *mdb, int field_num) {
396 // for a given column, find the max length
403 for (i = 0; i < mdb->num_rows; i++) {
404 if (PQgetisnull(mdb->result, i, field_num)) {
405 this_length = 4; // "NULL"
407 this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
410 if (max_length < this_length) {
411 max_length = this_length;
418 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
422 Dmsg0(500, "my_postgresql_fetch_field starts\n");
424 if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
428 Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
429 mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
430 mdb->fields_size = mdb->num_fields;
432 for (i = 0; i < mdb->num_fields; i++) {
433 Dmsg1(500, "filling field %d\n", i);
434 mdb->fields[i].name = PQfname(mdb->result, i);
435 mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
436 mdb->fields[i].type = PQftype(mdb->result, i);
437 mdb->fields[i].flags = 0;
439 Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
440 mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
441 mdb->fields[i].flags);
445 // increment field number for the next time around
447 Dmsg0(500, "my_postgresql_fetch_field finishes\n");
448 return &mdb->fields[mdb->field_number++];
451 void my_postgresql_data_seek(B_DB *mdb, int row)
453 // set the row number to be returned on the next call
454 // to my_postgresql_fetch_row
455 mdb->row_number = row;
458 void my_postgresql_field_seek(B_DB *mdb, int field)
460 mdb->field_number = field;
464 * Note, if this routine returns 1 (failure), Bacula expects
465 * that no result has been stored.
466 * This is where QUERY_DB comes with Postgresql.
468 * Returns: 0 on success
472 int my_postgresql_query(B_DB *mdb, const char *query)
474 Dmsg0(500, "my_postgresql_query started\n");
475 // We are starting a new query. reset everything.
477 mdb->row_number = -1;
478 mdb->field_number = -1;
481 PQclear(mdb->result); /* hmm, someone forgot to free?? */
485 Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
487 for (int i=0; i < 10; i++) {
488 mdb->result = PQexec(mdb->db, query);
495 Dmsg1(50, "Query failed: %s\n", query);
499 mdb->status = PQresultStatus(mdb->result);
500 if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
501 Dmsg1(500, "we have a result\n", query);
503 // how many fields in the set?
504 mdb->num_fields = (int)PQnfields(mdb->result);
505 Dmsg1(500, "we have %d fields\n", mdb->num_fields);
507 mdb->num_rows = PQntuples(mdb->result);
508 Dmsg1(500, "we have %d rows\n", mdb->num_rows);
510 mdb->status = 0; /* succeed */
512 Dmsg1(50, "Result status failed: %s\n", query);
516 Dmsg0(500, "my_postgresql_query finishing\n");
520 Dmsg1(500, "we failed\n", query);
521 PQclear(mdb->result);
523 mdb->status = 1; /* failed */
527 void my_postgresql_free_result(B_DB *mdb)
532 PQclear(mdb->result);
548 int my_postgresql_currval(B_DB *mdb, char *table_name)
550 // Obtain the current value of the sequence that
551 // provides the serial value for primary key of the table.
553 // currval is local to our session. It is not affected by
554 // other transactions.
556 // Determine the name of the sequence.
557 // PostgreSQL automatically creates a sequence using
558 // <table>_<column>_seq.
559 // At the time of writing, all tables used this format for
560 // for their primary key: <table>id
561 // Except for basefiles which has a primary key on baseid.
562 // Therefore, we need to special case that one table.
564 // everything else can use the PostgreSQL formula.
566 char sequence[NAMEDATALEN-1];
567 char query [NAMEDATALEN+50];
571 if (strcasecmp(table_name, "basefiles") == 0) {
572 bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
574 bstrncpy(sequence, table_name, sizeof(sequence));
575 bstrncat(sequence, "_", sizeof(sequence));
576 bstrncat(sequence, table_name, sizeof(sequence));
577 bstrncat(sequence, "id", sizeof(sequence));
580 bstrncat(sequence, "_seq", sizeof(sequence));
581 bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
583 Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
584 for (int i=0; i < 10; i++) {
585 result = PQexec(mdb->db, query);
592 Dmsg1(50, "Query failed: %s\n", query);
596 Dmsg0(500, "exec done");
598 if (PQresultStatus(result) == PGRES_TUPLES_OK) {
599 Dmsg0(500, "getting value");
600 id = atoi(PQgetvalue(result, 0, 0));
601 Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
603 Dmsg1(50, "Result status failed: %s\n", query);
604 Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
613 #ifdef HAVE_BATCH_FILE_INSERT
615 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
617 char *query = "COPY batch FROM STDIN";
619 Dmsg0(500, "my_postgresql_batch_start started\n");
621 if (my_postgresql_query(mdb,
622 " CREATE TEMPORARY TABLE batch "
628 " md5 varchar)") == 1)
630 Dmsg0(500, "my_postgresql_batch_start failed\n");
634 // We are starting a new query. reset everything.
636 mdb->row_number = -1;
637 mdb->field_number = -1;
639 my_postgresql_free_result(mdb);
641 for (int i=0; i < 10; i++) {
642 mdb->result = PQexec(mdb->db, query);
649 Dmsg1(50, "Query failed: %s\n", query);
653 mdb->status = PQresultStatus(mdb->result);
654 if (mdb->status == PGRES_COPY_IN) {
655 // how many fields in the set?
656 mdb->num_fields = (int) PQnfields(mdb->result);
660 Dmsg1(50, "Result status failed: %s\n", query);
664 Dmsg0(500, "my_postgresql_batch_start finishing\n");
670 PQclear(mdb->result);
675 /* set error to something to abort operation */
676 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
680 Dmsg0(500, "my_postgresql_batch_end started\n");
682 if (!mdb) { /* no files ? */
687 res = PQputCopyEnd(mdb->db, error);
688 } while (res == 0 && --count > 0);
696 Dmsg0(500, "we failed\n");
698 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
701 Dmsg0(500, "my_postgresql_batch_end finishing\n");
706 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
714 mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
715 my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
717 mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
718 my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
720 if (ar->Digest == NULL || ar->Digest[0] == 0) {
726 len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
727 ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
728 mdb->esc_name, ar->attr, digest);
731 res = PQputCopyData(mdb->db,
734 } while (res == 0 && --count > 0);
743 Dmsg0(500, "we failed\n");
745 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
748 Dmsg0(500, "my_postgresql_batch_insert finishing\n");
753 #endif /* HAVE_BATCH_FILE_INSERT */
756 * Escape strings so that PostgreSQL is happy on COPY
758 * NOTE! len is the length of the old string. Your new
759 * string must be long enough (max 2*old+1) to hold
760 * the escaped output.
762 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
764 /* we have to escape \t, \n, \r, \ */
767 while (len > 0 && *src) {
802 #ifdef HAVE_BATCH_FILE_INSERT
803 char *my_pg_batch_lock_path_query = "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
806 char *my_pg_batch_lock_filename_query = "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
808 char *my_pg_batch_unlock_tables_query = "COMMIT";
810 char *my_pg_batch_fill_path_query = "INSERT INTO Path (Path) "
811 " SELECT a.Path FROM "
812 " (SELECT DISTINCT Path FROM batch) AS a "
813 " WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
816 char *my_pg_batch_fill_filename_query = "INSERT INTO Filename (Name) "
817 " SELECT a.Name FROM "
818 " (SELECT DISTINCT Name FROM batch) as a "
820 " (SELECT Name FROM Filename WHERE Name = a.Name)";
821 #endif /* HAVE_BATCH_FILE_INSERT */
823 #endif /* HAVE_POSTGRESQL */