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 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 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 */
50 #include "pg_config_manual.h" /* get NAMEDATALEN on version 8.3 or later */
52 /* -----------------------------------------------------------------------
54 * PostgreSQL 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
75 * Initialize database data structure. In principal this should
76 * never have errors, or it is really fatal.
79 db_init_database(JCR *jcr, const char *db_name, const char *db_user, const char *db_password,
80 const char *db_address, int db_port, const char *db_socket,
81 int mult_db_connections)
86 Jmsg(jcr, M_FATAL, 0, _("A user name for PostgreSQL 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 */
103 Dmsg0(100, "db_open first time\n");
104 mdb = (B_DB *)malloc(sizeof(B_DB));
105 memset(mdb, 0, sizeof(B_DB));
106 mdb->db_name = bstrdup(db_name);
107 mdb->db_user = bstrdup(db_user);
109 mdb->db_password = bstrdup(db_password);
112 mdb->db_address = bstrdup(db_address);
115 mdb->db_socket = bstrdup(db_socket);
117 mdb->db_port = db_port;
118 mdb->have_insert_id = TRUE;
119 mdb->errmsg = get_pool_memory(PM_EMSG); /* get error message buffer */
121 mdb->cmd = get_pool_memory(PM_EMSG); /* get command buffer */
122 mdb->cached_path = get_pool_memory(PM_FNAME);
123 mdb->cached_path_id = 0;
125 mdb->fname = get_pool_memory(PM_FNAME);
126 mdb->path = get_pool_memory(PM_FNAME);
127 mdb->esc_name = get_pool_memory(PM_FNAME);
128 mdb->esc_path = get_pool_memory(PM_FNAME);
129 mdb->allow_transactions = mult_db_connections;
130 qinsert(&db_list, &mdb->bq); /* put db in list */
135 /* Check that the database correspond to the encoding we want */
136 static bool check_database_encoding(JCR *jcr, B_DB *mdb)
141 if (!db_sql_query(mdb, "SELECT getdatabaseencoding()", NULL, NULL)) {
142 Jmsg(jcr, M_ERROR, 0, "%s", mdb->errmsg);
146 if ((row = sql_fetch_row(mdb)) == NULL) {
147 Mmsg1(mdb->errmsg, _("error fetching row: %s\n"), sql_strerror(mdb));
148 Jmsg(jcr, M_ERROR, 0, "Can't check database encoding %s", mdb->errmsg);
150 ret = bstrcmp(row[0], "SQL_ASCII");
153 /* if we are in SQL_ASCII, we can force the client_encoding to SQL_ASCII too */
154 db_sql_query(mdb, "SET client_encoding TO 'SQL_ASCII'", NULL, NULL);
156 } else { /* something is wrong with database encoding */
158 _("Encoding error for database \"%s\". Wanted SQL_ASCII, got %s\n"),
159 mdb->db_name, row[0]);
160 Jmsg(jcr, M_WARNING, 0, "%s", mdb->errmsg);
161 Dmsg1(50, "%s", mdb->errmsg);
168 * Now actually open the database. This can generate errors,
169 * which are returned in the errmsg
171 * DO NOT close the database or free(mdb) here !!!!
174 db_open_database(JCR *jcr, B_DB *mdb)
179 #ifdef xxx /* require libpq >= 8.2 */
180 if (!PQisthreadsafe()) {
181 Jmsg(jcr, M_ABORT, 0, _("PostgreSQL configuration problem. "
182 "PostgreSQL library is not thread safe. Cannot continue.\n"));
186 if (mdb->connected) {
190 mdb->connected = false;
192 if ((errstat=rwl_init(&mdb->lock)) != 0) {
194 Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
195 be.bstrerror(errstat));
201 bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
207 /* If connection fails, try at 5 sec intervals for 30 seconds. */
208 for (int retry=0; retry < 6; retry++) {
209 /* connect to the database */
210 mdb->db = PQsetdbLogin(
211 mdb->db_address, /* default = localhost */
212 port, /* default port */
213 NULL, /* pg options */
214 NULL, /* tty, ignored */
215 mdb->db_name, /* database name */
216 mdb->db_user, /* login name */
217 mdb->db_password); /* password */
219 /* If no connect, try once more in case it is a timing problem */
220 if (PQstatus(mdb->db) == CONNECTION_OK) {
226 Dmsg0(50, "pg_real_connect done\n");
227 Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
228 mdb->db_password==NULL?"(NULL)":mdb->db_password);
230 if (PQstatus(mdb->db) != CONNECTION_OK) {
231 Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server. Database=%s User=%s\n"
232 "Possible causes: SQL server not running; password incorrect; max_connections exceeded.\n"),
233 mdb->db_name, mdb->db_user);
238 mdb->connected = true;
240 if (!check_tables_version(jcr, mdb)) {
245 sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
247 /* tell PostgreSQL we are using standard conforming strings
248 and avoid warnings such as:
249 WARNING: nonstandard use of \\ in a string literal
251 sql_query(mdb, "set standard_conforming_strings=on");
253 /* check that encoding is SQL_ASCII */
254 check_database_encoding(jcr, mdb);
261 db_close_database(JCR *jcr, B_DB *mdb)
266 db_end_transaction(jcr, mdb);
268 sql_free_result(mdb);
270 if (mdb->ref_count == 0) {
272 if (mdb->connected && mdb->db) {
275 rwl_destroy(&mdb->lock);
276 free_pool_memory(mdb->errmsg);
277 free_pool_memory(mdb->cmd);
278 free_pool_memory(mdb->cached_path);
279 free_pool_memory(mdb->fname);
280 free_pool_memory(mdb->path);
281 free_pool_memory(mdb->esc_name);
282 free_pool_memory(mdb->esc_path);
289 if (mdb->db_password) {
290 free(mdb->db_password);
292 if (mdb->db_address) {
293 free(mdb->db_address);
295 if (mdb->db_socket) {
296 free(mdb->db_socket);
303 void db_thread_cleanup()
307 * Return the next unique index (auto-increment) for
308 * the given table. Return NULL on error.
310 * For PostgreSQL, NULL causes the auto-increment value
313 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
315 strcpy(index, "NULL");
321 * Escape strings so that PostgreSQL is happy
323 * NOTE! len is the length of the old string. Your new
324 * string must be long enough (max 2*old+1) to hold
325 * the escaped output.
328 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
332 PQescapeStringConn(mdb->db, snew, old, len, &error);
334 Jmsg(jcr, M_FATAL, 0, _("PQescapeStringConn returned non-zero.\n"));
335 /* error on encoding, probably invalid multibyte encoding in the source string
336 see PQescapeStringConn documentation for details. */
337 Dmsg0(500, "PQescapeStringConn failed\n");
342 * Submit a general SQL command (cmd), and for each row returned,
343 * the sqlite_handler is called with the ctx.
345 bool db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
349 Dmsg0(500, "db_sql_query started\n");
352 if (sql_query(mdb, query) != 0) {
353 Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
355 Dmsg0(500, "db_sql_query failed\n");
358 Dmsg0(500, "db_sql_query succeeded. checking handler\n");
360 if (result_handler != NULL) {
361 Dmsg0(500, "db_sql_query invoking handler\n");
362 if ((mdb->result = sql_store_result(mdb)) != NULL) {
363 int num_fields = sql_num_fields(mdb);
365 Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
366 while ((row = sql_fetch_row(mdb)) != NULL) {
368 Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
369 if (result_handler(ctx, num_fields, row))
373 sql_free_result(mdb);
378 Dmsg0(500, "db_sql_query finished\n");
385 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
388 POSTGRESQL_ROW row = NULL; // by default, return NULL
390 Dmsg0(500, "my_postgresql_fetch_row start\n");
392 if (!mdb->row || mdb->row_size < mdb->num_fields) {
393 int num_fields = mdb->num_fields;
394 Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
397 Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
400 num_fields += 20; /* add a bit extra */
401 mdb->row = (POSTGRESQL_ROW)malloc(sizeof(char *) * num_fields);
402 mdb->row_size = num_fields;
404 // now reset the row_number now that we have the space allocated
408 // if still within the result set
409 if (mdb->row_number >= 0 && mdb->row_number < mdb->num_rows) {
410 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
411 // get each value from this row
412 for (j = 0; j < mdb->num_fields; j++) {
413 mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
414 Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
416 // increment the row number for the next call
421 Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
424 Dmsg1(500, "my_postgresql_fetch_row finishes returning %p\n", row);
429 int my_postgresql_max_length(B_DB *mdb, int field_num) {
431 // for a given column, find the max length
438 for (i = 0; i < mdb->num_rows; i++) {
439 if (PQgetisnull(mdb->result, i, field_num)) {
440 this_length = 4; // "NULL"
442 this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
445 if (max_length < this_length) {
446 max_length = this_length;
453 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
457 Dmsg0(500, "my_postgresql_fetch_field starts\n");
459 if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
463 Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
464 mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
465 mdb->fields_size = mdb->num_fields;
467 for (i = 0; i < mdb->num_fields; i++) {
468 Dmsg1(500, "filling field %d\n", i);
469 mdb->fields[i].name = PQfname(mdb->result, i);
470 mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
471 mdb->fields[i].type = PQftype(mdb->result, i);
472 mdb->fields[i].flags = 0;
474 Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
475 mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
476 mdb->fields[i].flags);
480 // increment field number for the next time around
482 Dmsg0(500, "my_postgresql_fetch_field finishes\n");
483 return &mdb->fields[mdb->field_number++];
486 void my_postgresql_data_seek(B_DB *mdb, int row)
488 // set the row number to be returned on the next call
489 // to my_postgresql_fetch_row
490 mdb->row_number = row;
493 void my_postgresql_field_seek(B_DB *mdb, int field)
495 mdb->field_number = field;
499 * Note, if this routine returns 1 (failure), Bacula expects
500 * that no result has been stored.
501 * This is where QUERY_DB comes with Postgresql.
503 * Returns: 0 on success
507 int my_postgresql_query(B_DB *mdb, const char *query)
509 Dmsg0(500, "my_postgresql_query started\n");
510 // We are starting a new query. reset everything.
512 mdb->row_number = -1;
513 mdb->field_number = -1;
516 PQclear(mdb->result); /* hmm, someone forgot to free?? */
520 Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
522 for (int i=0; i < 10; i++) {
523 mdb->result = PQexec(mdb->db, query);
530 Dmsg1(50, "Query failed: %s\n", query);
534 mdb->status = PQresultStatus(mdb->result);
535 if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
536 Dmsg1(500, "we have a result\n", query);
538 // how many fields in the set?
539 mdb->num_fields = (int)PQnfields(mdb->result);
540 Dmsg1(500, "we have %d fields\n", mdb->num_fields);
542 mdb->num_rows = PQntuples(mdb->result);
543 Dmsg1(500, "we have %d rows\n", mdb->num_rows);
545 mdb->row_number = 0; /* we can start to fetch something */
546 mdb->status = 0; /* succeed */
548 Dmsg1(50, "Result status failed: %s\n", query);
552 Dmsg0(500, "my_postgresql_query finishing\n");
556 Dmsg1(500, "we failed\n", query);
557 PQclear(mdb->result);
559 mdb->status = 1; /* failed */
563 void my_postgresql_free_result(B_DB *mdb)
568 PQclear(mdb->result);
584 int my_postgresql_currval(B_DB *mdb, const char *table_name)
586 // Obtain the current value of the sequence that
587 // provides the serial value for primary key of the table.
589 // currval is local to our session. It is not affected by
590 // other transactions.
592 // Determine the name of the sequence.
593 // PostgreSQL automatically creates a sequence using
594 // <table>_<column>_seq.
595 // At the time of writing, all tables used this format for
596 // for their primary key: <table>id
597 // Except for basefiles which has a primary key on baseid.
598 // Therefore, we need to special case that one table.
600 // everything else can use the PostgreSQL formula.
602 char sequence[NAMEDATALEN-1];
603 char query [NAMEDATALEN+50];
607 if (strcasecmp(table_name, "basefiles") == 0) {
608 bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
610 bstrncpy(sequence, table_name, sizeof(sequence));
611 bstrncat(sequence, "_", sizeof(sequence));
612 bstrncat(sequence, table_name, sizeof(sequence));
613 bstrncat(sequence, "id", sizeof(sequence));
616 bstrncat(sequence, "_seq", sizeof(sequence));
617 bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
619 Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
620 for (int i=0; i < 10; i++) {
621 result = PQexec(mdb->db, query);
628 Dmsg1(50, "Query failed: %s\n", query);
632 Dmsg0(500, "exec done");
634 if (PQresultStatus(result) == PGRES_TUPLES_OK) {
635 Dmsg0(500, "getting value");
636 id = atoi(PQgetvalue(result, 0, 0));
637 Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
639 Dmsg1(50, "Result status failed: %s\n", query);
640 Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
649 #ifdef HAVE_BATCH_FILE_INSERT
651 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
653 const char *query = "COPY batch FROM STDIN";
655 Dmsg0(500, "my_postgresql_batch_start started\n");
657 if (my_postgresql_query(mdb,
658 "CREATE TEMPORARY TABLE batch ("
664 "md5 varchar)") == 1)
666 Dmsg0(500, "my_postgresql_batch_start failed\n");
670 // We are starting a new query. reset everything.
672 mdb->row_number = -1;
673 mdb->field_number = -1;
675 my_postgresql_free_result(mdb);
677 for (int i=0; i < 10; i++) {
678 mdb->result = PQexec(mdb->db, query);
685 Dmsg1(50, "Query failed: %s\n", query);
689 mdb->status = PQresultStatus(mdb->result);
690 if (mdb->status == PGRES_COPY_IN) {
691 // how many fields in the set?
692 mdb->num_fields = (int) PQnfields(mdb->result);
696 Dmsg1(50, "Result status failed: %s\n", query);
700 Dmsg0(500, "my_postgresql_batch_start finishing\n");
705 Mmsg1(&mdb->errmsg, _("error starting batch mode: %s"), PQerrorMessage(mdb->db));
707 PQclear(mdb->result);
712 /* set error to something to abort operation */
713 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
718 Dmsg0(500, "my_postgresql_batch_end started\n");
720 if (!mdb) { /* no files ? */
725 res = PQputCopyEnd(mdb->db, error);
726 } while (res == 0 && --count > 0);
734 Dmsg0(500, "we failed\n");
736 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
739 /* Check command status and return to normal libpq state */
740 result = PQgetResult(mdb->db);
741 if (PQresultStatus(result) != PGRES_COMMAND_OK) {
742 Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
747 Dmsg0(500, "my_postgresql_batch_end finishing\n");
752 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
760 mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
761 my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
763 mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
764 my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
766 if (ar->Digest == NULL || ar->Digest[0] == 0) {
772 len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
773 ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
774 mdb->esc_name, ar->attr, digest);
777 res = PQputCopyData(mdb->db,
780 } while (res == 0 && --count > 0);
789 Dmsg0(500, "we failed\n");
791 Mmsg1(&mdb->errmsg, _("error copying in batch mode: %s"), PQerrorMessage(mdb->db));
794 Dmsg0(500, "my_postgresql_batch_insert finishing\n");
799 #endif /* HAVE_BATCH_FILE_INSERT */
802 * Escape strings so that PostgreSQL is happy on COPY
804 * NOTE! len is the length of the old string. Your new
805 * string must be long enough (max 2*old+1) to hold
806 * the escaped output.
808 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
810 /* we have to escape \t, \n, \r, \ */
813 while (len > 0 && *src) {
848 #ifdef HAVE_BATCH_FILE_INSERT
849 const char *my_pg_batch_lock_path_query =
850 "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
853 const char *my_pg_batch_lock_filename_query =
854 "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
856 const char *my_pg_batch_unlock_tables_query = "COMMIT";
858 const char *my_pg_batch_fill_path_query =
859 "INSERT INTO Path (Path) "
860 "SELECT a.Path FROM "
861 "(SELECT DISTINCT Path FROM batch) AS a "
862 "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
865 const char *my_pg_batch_fill_filename_query =
866 "INSERT INTO Filename (Name) "
867 "SELECT a.Name FROM "
868 "(SELECT DISTINCT Name FROM batch) as a "
870 "(SELECT Name FROM Filename WHERE Name = a.Name)";
871 #endif /* HAVE_BATCH_FILE_INSERT */
873 #endif /* HAVE_POSTGRESQL */