Index: config.h.in =================================================================== RCS file: /cvsroot/pgpool/pgpool-II/config.h.in,v retrieving revision 1.12 diff -c -r1.12 config.h.in *** config.h.in 10 Nov 2010 01:53:50 -0000 1.12 --- config.h.in 29 Jun 2011 10:15:13 -0000 *************** *** 292,300 **** --- 292,308 ---- /* Use replacement snprintf() functions. */ #undef USE_REPL_SNPRINTF + /* Define to 1 if you want to use row lock against the sequence table for + insert_lock. (--enable-sequence-lock) */ + #undef USE_SEQUENCE_LOCK + /* Define to 1 to build with SSL support. (--with-openssl) */ #undef USE_SSL + /* Define to 1 if you want to use table lock against the target table for + insert_lock. (--enable-table-lock) */ + #undef USE_TABLE_LOCK + /* Version number of package */ #undef VERSION Index: configure.in =================================================================== RCS file: /cvsroot/pgpool/pgpool-II/configure.in,v retrieving revision 1.45 diff -c -r1.45 configure.in *** configure.in 22 Apr 2011 02:50:27 -0000 1.45 --- configure.in 29 Jun 2011 10:15:14 -0000 *************** *** 332,337 **** --- 332,363 ---- ) AM_CONDITIONAL([enable_rpath], test x$rpath = xyes) + # Decide whether to use row lock against the sequence table for insert_lock. + # This lock method is compatible with pgpool-II 3.0 series(until 3.0.4). + AC_MSG_CHECKING([whether to use row lock against the sequence table for insert_lock]) + PGAC_ARG_BOOL(enable, sequence-lock, no, [insert_lock compatible with pgpool-II 3.0 series (until 3.0.4)]) + if test "$enable_sequence_lock" = yes && test "$enable_table_lock" = yes ; then + AC_MSG_ERROR([--enable-table-lock cannot be enabled at the same time.]) + fi + if test "$enable_sequence_lock" = yes ; then + AC_DEFINE([USE_SEQUENCE_LOCK], 1, + [Define to 1 if you want to use row lock against the sequence table for insert_lock. (--enable-sequence-lock)]) + fi + AC_MSG_RESULT([$enable_sequence_lock]) + + # Decide whether to use table lock against the target table for insert_lock. + # This lock method is compatible with pgpool-II 2.2 and 2.3 series. + AC_MSG_CHECKING([whether to use table lock against the target table for insert_lock]) + PGAC_ARG_BOOL(enable, table-lock, no, [insert_lock compatible with pgpool-II 2.2 and 2.3 series]) + if test "$enable_table_lock" = yes && test "$enable_sequence_lock" = yes ; then + AC_MSG_ERROR([--enable-sequence-lock cannot be enabled at the same time.]) + fi + if test "$enable_table_lock" = yes ; then + AC_DEFINE([USE_TABLE_LOCK], 1, + [Define to 1 if you want to use table lock against the target table for insert_lock. (--enable-table-lock)]) + fi + AC_MSG_RESULT([$enable_table_lock]) + AM_CONFIG_HEADER(config.h) AC_OUTPUT([Makefile parser/Makefile pcp/Makefile]) Index: configure =================================================================== RCS file: /cvsroot/pgpool/pgpool-II/configure,v retrieving revision 1.48 diff -c -r1.48 configure *** configure 22 Apr 2011 02:50:27 -0000 1.48 --- configure 29 Jun 2011 10:15:23 -0000 *************** *** 917,922 **** --- 917,924 ---- with_openssl with_pam enable_rpath + enable_sequence_lock + enable_table_lock ' ac_precious_vars='build_alias host_alias *************** *** 1567,1572 **** --- 1569,1578 ---- --disable-float4-byval disable float4 passed by value --disable-float8-byval disable float8 passed by value --disable-rpath do not embed shared library search path in executables + --enable-sequence-lock insert_lock compatible with pgpool-II 3.0 series + (until 3.0.4) + --enable-table-lock insert_lock compatible with pgpool-II 2.2 and 2.3 + series Optional Packages: --with-PACKAGE[=ARG] use PACKAGE [ARG=yes] *************** *** 4423,4435 **** else lt_cv_nm_interface="BSD nm" echo "int some_variable = 0;" > conftest.$ac_ext ! (eval echo "\"\$as_me:4426: $ac_compile\"" >&5) (eval "$ac_compile" 2>conftest.err) cat conftest.err >&5 ! (eval echo "\"\$as_me:4429: $NM \\\"conftest.$ac_objext\\\"\"" >&5) (eval "$NM \"conftest.$ac_objext\"" 2>conftest.err > conftest.out) cat conftest.err >&5 ! (eval echo "\"\$as_me:4432: output\"" >&5) cat conftest.out >&5 if $GREP 'External.*some_variable' conftest.out > /dev/null; then lt_cv_nm_interface="MS dumpbin" --- 4429,4441 ---- else lt_cv_nm_interface="BSD nm" echo "int some_variable = 0;" > conftest.$ac_ext ! (eval echo "\"\$as_me:4432: $ac_compile\"" >&5) (eval "$ac_compile" 2>conftest.err) cat conftest.err >&5 ! (eval echo "\"\$as_me:4435: $NM \\\"conftest.$ac_objext\\\"\"" >&5) (eval "$NM \"conftest.$ac_objext\"" 2>conftest.err > conftest.out) cat conftest.err >&5 ! (eval echo "\"\$as_me:4438: output\"" >&5) cat conftest.out >&5 if $GREP 'External.*some_variable' conftest.out > /dev/null; then lt_cv_nm_interface="MS dumpbin" *************** *** 5635,5641 **** ;; *-*-irix6*) # Find out which ABI we are using. ! echo '#line 5638 "configure"' > conftest.$ac_ext if { (eval echo "$as_me:$LINENO: \"$ac_compile\"") >&5 (eval $ac_compile) 2>&5 ac_status=$? --- 5641,5647 ---- ;; *-*-irix6*) # Find out which ABI we are using. ! echo '#line 5644 "configure"' > conftest.$ac_ext if { (eval echo "$as_me:$LINENO: \"$ac_compile\"") >&5 (eval $ac_compile) 2>&5 ac_status=$? *************** *** 7463,7468 **** --- 7469,7478 ---- $RM -r conftest* + ## CAVEAT EMPTOR: + ## There is no encapsulation within the following macros, do not change + ## the running order or otherwise move them around unless you know exactly + ## what you are doing... if test -n "$compiler"; then lt_prog_compiler_no_builtin_flag= *************** *** 7488,7498 **** -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7491: $lt_compile\"" >&5) (eval "$lt_compile" 2>conftest.err) ac_status=$? cat conftest.err >&5 ! echo "$as_me:7495: \$? = $ac_status" >&5 if (exit $ac_status) && test -s "$ac_outfile"; then # The compiler can only warn and ignore the option if not recognized # So say no if there are warnings other than the usual output. --- 7498,7508 ---- -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7501: $lt_compile\"" >&5) (eval "$lt_compile" 2>conftest.err) ac_status=$? cat conftest.err >&5 ! echo "$as_me:7505: \$? = $ac_status" >&5 if (exit $ac_status) && test -s "$ac_outfile"; then # The compiler can only warn and ignore the option if not recognized # So say no if there are warnings other than the usual output. *************** *** 7827,7837 **** -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7830: $lt_compile\"" >&5) (eval "$lt_compile" 2>conftest.err) ac_status=$? cat conftest.err >&5 ! echo "$as_me:7834: \$? = $ac_status" >&5 if (exit $ac_status) && test -s "$ac_outfile"; then # The compiler can only warn and ignore the option if not recognized # So say no if there are warnings other than the usual output. --- 7837,7847 ---- -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7840: $lt_compile\"" >&5) (eval "$lt_compile" 2>conftest.err) ac_status=$? cat conftest.err >&5 ! echo "$as_me:7844: \$? = $ac_status" >&5 if (exit $ac_status) && test -s "$ac_outfile"; then # The compiler can only warn and ignore the option if not recognized # So say no if there are warnings other than the usual output. *************** *** 7932,7942 **** -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7935: $lt_compile\"" >&5) (eval "$lt_compile" 2>out/conftest.err) ac_status=$? cat out/conftest.err >&5 ! echo "$as_me:7939: \$? = $ac_status" >&5 if (exit $ac_status) && test -s out/conftest2.$ac_objext then # The compiler can only warn and ignore the option if not recognized --- 7942,7952 ---- -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7945: $lt_compile\"" >&5) (eval "$lt_compile" 2>out/conftest.err) ac_status=$? cat out/conftest.err >&5 ! echo "$as_me:7949: \$? = $ac_status" >&5 if (exit $ac_status) && test -s out/conftest2.$ac_objext then # The compiler can only warn and ignore the option if not recognized *************** *** 7987,7997 **** -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:7990: $lt_compile\"" >&5) (eval "$lt_compile" 2>out/conftest.err) ac_status=$? cat out/conftest.err >&5 ! echo "$as_me:7994: \$? = $ac_status" >&5 if (exit $ac_status) && test -s out/conftest2.$ac_objext then # The compiler can only warn and ignore the option if not recognized --- 7997,8007 ---- -e 's:.*FLAGS}\{0,1\} :&$lt_compiler_flag :; t' \ -e 's: [^ ]*conftest\.: $lt_compiler_flag&:; t' \ -e 's:$: $lt_compiler_flag:'` ! (eval echo "\"\$as_me:8000: $lt_compile\"" >&5) (eval "$lt_compile" 2>out/conftest.err) ac_status=$? cat out/conftest.err >&5 ! echo "$as_me:8004: \$? = $ac_status" >&5 if (exit $ac_status) && test -s out/conftest2.$ac_objext then # The compiler can only warn and ignore the option if not recognized *************** *** 9839,9851 **** # before this can be enabled. hardcode_into_libs=yes - # Add ABI-specific directories to the system library path. - sys_lib_dlsearch_path_spec="/lib64 /usr/lib64 /lib /usr/lib" - # Append ld.so.conf contents to the search path if test -f /etc/ld.so.conf; then lt_ld_extra=`awk '/^include / { system(sprintf("cd /etc; cat %s 2>/dev/null", \$2)); skip = 1; } { if (!skip) print \$0; skip = 0; }' < /etc/ld.so.conf | $SED -e 's/#.*//;/^[ ]*hwcap[ ]/d;s/[:, ]/ /g;s/=[^=]*$//;s/=[^= ]* / /g;/^$/d' | tr '\n' ' '` ! sys_lib_dlsearch_path_spec="$sys_lib_dlsearch_path_spec $lt_ld_extra" fi # We used to test for /lib/ld.so.1 and disable shared libraries on --- 9849,9858 ---- # before this can be enabled. hardcode_into_libs=yes # Append ld.so.conf contents to the search path if test -f /etc/ld.so.conf; then lt_ld_extra=`awk '/^include / { system(sprintf("cd /etc; cat %s 2>/dev/null", \$2)); skip = 1; } { if (!skip) print \$0; skip = 0; }' < /etc/ld.so.conf | $SED -e 's/#.*//;/^[ ]*hwcap[ ]/d;s/[:, ]/ /g;s/=[^=]*$//;s/=[^= ]* / /g;/^$/d' | tr '\n' ' '` ! sys_lib_dlsearch_path_spec="/lib /usr/lib $lt_ld_extra" fi # We used to test for /lib/ld.so.1 and disable shared libraries on *************** *** 10790,10796 **** lt_dlunknown=0; lt_dlno_uscore=1; lt_dlneed_uscore=2 lt_status=$lt_dlunknown cat > conftest.$ac_ext <<_LT_EOF ! #line 10793 "configure" #include "confdefs.h" #if HAVE_DLFCN_H --- 10797,10803 ---- lt_dlunknown=0; lt_dlno_uscore=1; lt_dlneed_uscore=2 lt_status=$lt_dlunknown cat > conftest.$ac_ext <<_LT_EOF ! #line 10800 "configure" #include "confdefs.h" #if HAVE_DLFCN_H *************** *** 10886,10892 **** lt_dlunknown=0; lt_dlno_uscore=1; lt_dlneed_uscore=2 lt_status=$lt_dlunknown cat > conftest.$ac_ext <<_LT_EOF ! #line 10889 "configure" #include "confdefs.h" #if HAVE_DLFCN_H --- 10893,10899 ---- lt_dlunknown=0; lt_dlno_uscore=1; lt_dlneed_uscore=2 lt_status=$lt_dlunknown cat > conftest.$ac_ext <<_LT_EOF ! #line 10896 "configure" #include "confdefs.h" #if HAVE_DLFCN_H *************** *** 17409,17414 **** --- 17416,17511 ---- fi + # Decide whether to use row lock against the sequence table for insert_lock. + # This lock method is compatible with pgpool-II 3.0 series(until 3.0.4). + { $as_echo "$as_me:$LINENO: checking whether to use row lock against the sequence table for insert_lock" >&5 + $as_echo_n "checking whether to use row lock against the sequence table for insert_lock... " >&6; } + + pgac_args="$pgac_args enable_sequence_lock" + + # Check whether --enable-sequence-lock was given. + if test "${enable_sequence_lock+set}" = set; then + enableval=$enable_sequence_lock; + case $enableval in + yes) + : + ;; + no) + : + ;; + *) + { { $as_echo "$as_me:$LINENO: error: no argument expected for --enable-sequence-lock option" >&5 + $as_echo "$as_me: error: no argument expected for --enable-sequence-lock option" >&2;} + { (exit 1); exit 1; }; } + ;; + esac + + else + enable_sequence_lock=no + + fi + + + if test "$enable_sequence_lock" = yes && test "$enable_table_lock" = yes ; then + { { $as_echo "$as_me:$LINENO: error: --enable-table-lock cannot be enabled at the same time." >&5 + $as_echo "$as_me: error: --enable-table-lock cannot be enabled at the same time." >&2;} + { (exit 1); exit 1; }; } + fi + if test "$enable_sequence_lock" = yes ; then + + cat >>confdefs.h <<\_ACEOF + #define USE_SEQUENCE_LOCK 1 + _ACEOF + + fi + { $as_echo "$as_me:$LINENO: result: $enable_sequence_lock" >&5 + $as_echo "$enable_sequence_lock" >&6; } + + # Decide whether to use table lock against the target table for insert_lock. + # This lock method is compatible with pgpool-II 2.2 and 2.3 series. + { $as_echo "$as_me:$LINENO: checking whether to use table lock against the target table for insert_lock" >&5 + $as_echo_n "checking whether to use table lock against the target table for insert_lock... " >&6; } + + pgac_args="$pgac_args enable_table_lock" + + # Check whether --enable-table-lock was given. + if test "${enable_table_lock+set}" = set; then + enableval=$enable_table_lock; + case $enableval in + yes) + : + ;; + no) + : + ;; + *) + { { $as_echo "$as_me:$LINENO: error: no argument expected for --enable-table-lock option" >&5 + $as_echo "$as_me: error: no argument expected for --enable-table-lock option" >&2;} + { (exit 1); exit 1; }; } + ;; + esac + + else + enable_table_lock=no + + fi + + + if test "$enable_table_lock" = yes && test "$enable_sequence_lock" = yes ; then + { { $as_echo "$as_me:$LINENO: error: --enable-sequence-lock cannot be enabled at the same time." >&5 + $as_echo "$as_me: error: --enable-sequence-lock cannot be enabled at the same time." >&2;} + { (exit 1); exit 1; }; } + fi + if test "$enable_table_lock" = yes ; then + + cat >>confdefs.h <<\_ACEOF + #define USE_TABLE_LOCK 1 + _ACEOF + + fi + { $as_echo "$as_me:$LINENO: result: $enable_table_lock" >&5 + $as_echo "$enable_table_lock" >&6; } + ac_config_headers="$ac_config_headers config.h" Index: pool_process_query.c =================================================================== RCS file: /cvsroot/pgpool/pgpool-II/pool_process_query.c,v retrieving revision 1.259 diff -c -r1.259 pool_process_query.c *** pool_process_query.c 13 May 2011 06:53:05 -0000 1.259 --- pool_process_query.c 29 Jun 2011 10:15:25 -0000 *************** *** 72,79 **** static bool is_panic_or_fatal_error(const char *message, int major); static int detect_error(POOL_CONNECTION *master, char *error_code, int major, char class, bool unread); static int detect_postmaster_down_error(POOL_CONNECTION *master, int major); - static bool is_internal_transaction_needed(Node *node); /* timeout sec for pool_check_fd */ static int timeoutsec; --- 72,82 ---- static bool is_panic_or_fatal_error(const char *message, int major); static int detect_error(POOL_CONNECTION *master, char *error_code, int major, char class, bool unread); static int detect_postmaster_down_error(POOL_CONNECTION *master, int major); static bool is_internal_transaction_needed(Node *node); + static bool pool_has_insert_lock(void); + static POOL_STATUS add_lock_target(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char* table); + static bool has_lock_target(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char* table, bool for_update); + static POOL_STATUS insert_oid_into_insert_lock(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char* table); /* timeout sec for pool_check_fd */ static int timeoutsec; *************** *** 2296,2301 **** --- 2299,2305 ---- static char nullmap[8192]; unsigned char mask = 0; + *result = NULL; res = malloc(sizeof(*res)); if (!res) { *************** *** 2305,2310 **** --- 2309,2316 ---- rowdesc = malloc(sizeof(*rowdesc)); if (!rowdesc) { + if (res) + free(res); pool_error("pool_query: malloc failed"); return POOL_ERROR; } *************** *** 2624,2630 **** * Return values are: * 0: lock is not neccessary * 1: table lock is required ! * 2: row lock is required (SERIAL is used) */ int need_insert_lock(POOL_CONNECTION_POOL *backend, char *query, Node *node) { --- 2630,2637 ---- * Return values are: * 0: lock is not neccessary * 1: table lock is required ! * 2: row lock against sequence table is required ! * 3: row lock against insert_lock table is required */ int need_insert_lock(POOL_CONNECTION_POOL *backend, char *query, Node *node) { *************** *** 2705,2721 **** /* * Search relcache. */ result = pool_search_relcache(relcache, backend, str)==0?0:2; return result; } /* ! * Issue LOCK TABLE IN SHARE ROW EXCLUSIVE MODE if lock_kind == 1. ! * Issue row lock if lock_kind == 2. */ POOL_STATUS insert_lock(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char *query, InsertStmt *node, int lock_kind) { ! char *table; int len = 0; char qbuf[1024]; POOL_STATUS status; --- 2712,2742 ---- /* * Search relcache. */ + #ifdef USE_TABLE_LOCK + result = pool_search_relcache(relcache, backend, str)==0?0:1; + #elif USE_SEQUENCE_LOCK result = pool_search_relcache(relcache, backend, str)==0?0:2; + #else + result = pool_search_relcache(relcache, backend, str)==0?0:3; + #endif return result; } /* ! * insert lock to synchronize sequence number ! * lock_kind are: ! * 1: Issue LOCK TABLE IN SHARE ROW EXCLUSIVE MODE ! * 2: Issue row lock against sequence table ! * 3: Issue row lock against pgpool_catalog.insert_lock table ! * "lock_kind == 2" is deprecated because PostgreSQL disallows ! * SELECT FOR UPDATE/SHARE on sequence tables since 2011/06/03. ! * See following threads for more details: ! * [HACKERS] pgpool versus sequences ! * [ADMIN] 'SGT DETAIL: Could not open file "pg_clog/05DC": ... */ POOL_STATUS insert_lock(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char *query, InsertStmt *node, int lock_kind) { ! char *table, *p; int len = 0; char qbuf[1024]; POOL_STATUS status; *************** *** 2725,2734 **** #define SEQUENCETABLEQUERY2 "SELECT d.adsrc FROM pg_catalog.pg_class c, pg_catalog.pg_attribute a LEFT JOIN pg_catalog.pg_attrdef d ON (a.attrelid = d.adrelid AND a.attnum = d.adnum) WHERE c.oid = a.attrelid AND a.attnum >= 1 AND a.attisdropped = 'f' AND c.oid = pgpool_regclass('%s') AND d.adsrc ~ 'nextval'" ! #define MAX_SEQ_NAME 128 char *adsrc; ! char seq_rel_name[MAX_SEQ_NAME+1]; regex_t preg; size_t nmatch = 2; regmatch_t pmatch[nmatch]; --- 2746,2761 ---- #define SEQUENCETABLEQUERY2 "SELECT d.adsrc FROM pg_catalog.pg_class c, pg_catalog.pg_attribute a LEFT JOIN pg_catalog.pg_attrdef d ON (a.attrelid = d.adrelid AND a.attnum = d.adnum) WHERE c.oid = a.attrelid AND a.attnum >= 1 AND a.attisdropped = 'f' AND c.oid = pgpool_regclass('%s') AND d.adsrc ~ 'nextval'" ! /* query to lock a row by only the specified table name without regard to the schema */ ! #define ROWLOCKQUERY "SELECT 1 FROM pgpool_catalog.insert_lock WHERE reloid = (SELECT oid FROM pg_catalog.pg_class WHERE relname = '%s' ORDER BY oid LIMIT 1) FOR UPDATE" ! ! #define ROWLOCKQUERY2 "SELECT 1 FROM pgpool_catalog.insert_lock WHERE reloid = pgpool_regclass('%s') FOR UPDATE" ! ! #define MAX_NAME_LEN 128 char *adsrc; ! char seq_rel_name[MAX_NAME_LEN+1]; ! char rel_name[MAX_NAME_LEN+1]; regex_t preg; size_t nmatch = 2; regmatch_t pmatch[nmatch]; *************** *** 2741,2754 **** /* get table name */ table = get_insert_command_table_name(node); /* could not get table name. probably wrong SQL command */ if (table == NULL) { return POOL_CONTINUE; } /* row lock for sequence table? */ ! if (lock_kind == 2) { /* * If relcache does not exist, create it. --- 2768,2796 ---- /* get table name */ table = get_insert_command_table_name(node); + /* trim quotes */ + p = table; + for (i=0; *p; p++) + { + if (*p != '"') + rel_name[i++] = *p; + } + rel_name[i] = '\0'; + /* could not get table name. probably wrong SQL command */ if (table == NULL) { return POOL_CONTINUE; } + /* table lock for insert target table? */ + if (lock_kind == 1) + { + /* Issue lock table command */ + snprintf(qbuf, sizeof(qbuf), "LOCK TABLE %s IN SHARE ROW EXCLUSIVE MODE", table); + } /* row lock for sequence table? */ ! else if (lock_kind == 2) { /* * If relcache does not exist, create it. *************** *** 2814,2823 **** pool_debug("seq rel name: %s", seq_rel_name); snprintf(qbuf, sizeof(qbuf), "SELECT 1 FROM %s FOR UPDATE", seq_rel_name); } else { ! /* Issue lock table command */ ! snprintf(qbuf, sizeof(qbuf), "LOCK TABLE %s IN SHARE ROW EXCLUSIVE MODE", table); } per_node_statement_log(backend, MASTER_NODE_ID, qbuf); --- 2856,2877 ---- pool_debug("seq rel name: %s", seq_rel_name); snprintf(qbuf, sizeof(qbuf), "SELECT 1 FROM %s FOR UPDATE", seq_rel_name); } + /* row lock for insert_lock table? */ else { ! if (pool_has_insert_lock()) ! { ! if (pool_has_pgpool_regclass()) ! snprintf(qbuf, sizeof(qbuf), ROWLOCKQUERY2, rel_name); ! else ! snprintf(qbuf, sizeof(qbuf), ROWLOCKQUERY, rel_name); ! } ! else ! { ! /* issue lock table command if insert_lock table does not exist */ ! lock_kind = 1; ! snprintf(qbuf, sizeof(qbuf), "LOCK TABLE %s IN SHARE ROW EXCLUSIVE MODE", table); ! } } per_node_statement_log(backend, MASTER_NODE_ID, qbuf); *************** *** 2827,2839 **** --- 2881,2937 ---- status = do_command(frontend, MASTER(backend), qbuf, MAJOR(backend), MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); } + else if (lock_kind == 2) + { + POOL_SELECT_RESULT *result; + status = do_query(MASTER(backend), qbuf, &result, MAJOR(backend)); + if (result) + free_select_result(result); + } else { POOL_SELECT_RESULT *result; + /* issue row lock command */ status = do_query(MASTER(backend), qbuf, &result, MAJOR(backend)); + if (status == POOL_CONTINUE) + { + /* does oid exist in insert_lock table? */ + if (result && result->numrows == 0) + { + free_select_result(result); + result = NULL; + + /* insert a lock target row into insert_lock table */ + status = add_lock_target(frontend, backend, rel_name); + if (status == POOL_CONTINUE) + { + per_node_statement_log(backend, MASTER_NODE_ID, qbuf); + + /* issue row lock command */ + status = do_query(MASTER(backend), qbuf, &result, MAJOR(backend)); + if (status == POOL_CONTINUE) + { + if (!(result && result->data[0] && !strcmp(result->data[0], "1"))) + status = POOL_ERROR; + } + } + } + } + if (result) free_select_result(result); + + if (status != POOL_CONTINUE) + { + /* try to lock table finally, if row lock failed */ + lock_kind = 1; + snprintf(qbuf, sizeof(qbuf), "LOCK TABLE %s IN SHARE ROW EXCLUSIVE MODE", table); + per_node_statement_log(backend, MASTER_NODE_ID, qbuf); + status = do_command(frontend, MASTER(backend), qbuf, MAJOR(backend), MASTER_CONNECTION(backend)->pid, + MASTER_CONNECTION(backend)->key, 0); + } } + if (status == POOL_END) { return POOL_END; *************** *** 2850,2865 **** MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); else { - per_node_statement_log(backend, i, qbuf); - if (lock_kind == 1) { status = do_command(frontend, CONNECTION(backend, i), qbuf, PROTO_MAJOR_V3, MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); } ! else { POOL_SELECT_RESULT *result; status = do_query(CONNECTION(backend,i), qbuf, &result, MAJOR(backend)); if (result) free_select_result(result); --- 2948,2963 ---- MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); else { if (lock_kind == 1) { + per_node_statement_log(backend, i, qbuf); status = do_command(frontend, CONNECTION(backend, i), qbuf, PROTO_MAJOR_V3, MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); } ! else if (lock_kind == 2) { POOL_SELECT_RESULT *result; + per_node_statement_log(backend, i, qbuf); status = do_query(CONNECTION(backend,i), qbuf, &result, MAJOR(backend)); if (result) free_select_result(result); *************** *** 2876,2881 **** --- 2974,3140 ---- return POOL_CONTINUE; } + /* + * Judge if we have insert_lock table or not. + */ + static bool pool_has_insert_lock(void) + { + /* + * Query to know if insert_lock table exists. + */ + #define HASINSERT_LOCKQUERY "SELECT count(*) FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON (c.relnamespace = n.oid) WHERE nspname = 'pgpool_catalog' AND relname = '%s'" + bool result; + static POOL_RELCACHE *relcache; + POOL_CONNECTION_POOL *backend; + + backend = pool_get_session_context()->backend; + + if (!relcache) + { + relcache = pool_create_relcache(32, HASINSERT_LOCKQUERY, + int_register_func, int_unregister_func, + false); + if (relcache == NULL) + { + pool_error("pool_has_insert_lock: pool_create_relcache error"); + return false; + } + } + + result = pool_search_relcache(relcache, backend, "insert_lock")==0?0:1; + return result; + } + + /* + * Insert a lock target row into insert_lock table. + * This function is called after the transaction has been started. + * Protocol is V3 only. + * Return POOL_CONTINUE if the row is inserted successfully + * or the row already exists, the others return POOL_ERROR. + */ + static POOL_STATUS add_lock_target(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char* table) + { + POOL_STATUS status; + + /* lock the row where reloid is 0 to avoid "duplicate key violates..." error when insert new oid */ + if (!has_lock_target(frontend, backend, NULL, true)) + { + pool_log("add_lock_target: could not lock the row where reloid is 0"); + + per_node_statement_log(backend, MASTER_NODE_ID, "LOCK TABLE pgpool_catalog.insert_lock IN SHARE ROW EXCLUSIVE MODE"); + status = do_command(frontend, MASTER(backend), "LOCK TABLE pgpool_catalog.insert_lock IN SHARE ROW EXCLUSIVE MODE", + PROTO_MAJOR_V3, MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); + if (status == POOL_CONTINUE) + { + if (has_lock_target(frontend, backend, NULL, false)) + { + pool_debug("add_lock_target: reloid 0 already exists in insert_lock table"); + } + else + { + if (insert_oid_into_insert_lock(frontend, backend, NULL) != POOL_CONTINUE) + { + pool_log("add_lock_target: could not insert 0 into insert_lock table"); + return POOL_ERROR; + } + } + } + else + { + return POOL_ERROR; + } + } + + /* does insert_lock table contain the oid of the table? */ + if (has_lock_target(frontend, backend, table, false)) + { + pool_debug("add_lock_target: \"%s\" oid already exists in insert_lock table", table); + return POOL_CONTINUE; + } + + /* insert the oid of the table into insert_lock table */ + if (insert_oid_into_insert_lock(frontend, backend, table) != POOL_CONTINUE) + { + pool_log("add_lock_target: could not insert \"%s\" oid into insert_lock table", table); + return POOL_ERROR; + } + + return POOL_CONTINUE; + } + + /* + * Judge if insert_lock table contains the oid of the specified table or not. + * If the table name is NULL, this function checks whether oid 0 exists. + * If lock is true, this function locks the row of the table oid. + */ + static bool has_lock_target(POOL_CONNECTION *frontend, + POOL_CONNECTION_POOL *backend, + char* table, bool lock) + { + char *suffix; + char qbuf[QUERY_STRING_BUFFER_LEN]; + POOL_STATUS status; + POOL_SELECT_RESULT *result; + + suffix = lock ? " FOR UPDATE" : ""; + + if (table) + { + if (pool_has_pgpool_regclass()) + snprintf(qbuf, sizeof(qbuf), "SELECT 1 FROM pgpool_catalog.insert_lock WHERE reloid = pgpool_regclass('%s')%s", table, suffix); + else + snprintf(qbuf, sizeof(qbuf), "SELECT 1 FROM pgpool_catalog.insert_lock WHERE reloid = (SELECT oid FROM pg_catalog.pg_class WHERE relname = '%s' ORDER BY oid LIMIT 1)%s", table, suffix); + } + else + { + snprintf(qbuf, sizeof(qbuf), "SELECT 1 FROM pgpool_catalog.insert_lock WHERE reloid = 0%s", suffix); + } + + per_node_statement_log(backend, MASTER_NODE_ID, qbuf); + status = do_query(MASTER(backend), qbuf, &result, MAJOR(backend)); + if (status == POOL_CONTINUE) + { + if (result && result->data[0] && !strcmp(result->data[0], "1")) + { + free_select_result(result); + return true; + } + } + + if (result) + free_select_result(result); + + return false; + } + + /* + * Insert the oid of the specified table into insert_lock table. + */ + static POOL_STATUS insert_oid_into_insert_lock(POOL_CONNECTION *frontend, + POOL_CONNECTION_POOL *backend, + char* table) + { + char qbuf[QUERY_STRING_BUFFER_LEN]; + POOL_STATUS status; + + if (table) + { + if (pool_has_pgpool_regclass()) + snprintf(qbuf, sizeof(qbuf), "INSERT INTO pgpool_catalog.insert_lock VALUES (pgpool_regclass('%s'))", table); + else + snprintf(qbuf, sizeof(qbuf), "INSERT INTO pgpool_catalog.insert_lock SELECT oid FROM pg_catalog.pg_class WHERE relname = '%s' ORDER BY oid LIMIT 1", table); + } + else + { + snprintf(qbuf, sizeof(qbuf), "INSERT INTO pgpool_catalog.insert_lock VALUES (0)"); + } + + per_node_statement_log(backend, MASTER_NODE_ID, qbuf); + status = do_command(frontend, MASTER(backend), qbuf, PROTO_MAJOR_V3, + MASTER_CONNECTION(backend)->pid, MASTER_CONNECTION(backend)->key, 0); + return status; + } + bool is_partition_table(POOL_CONNECTION_POOL *backend, Node *node) { DistDefInfo *info = NULL;