Consider SHOW command as kind of a read query.

In streaming replication mode, if SHOW is issued then subsequent
SELECTs are sent to the primary node in an explicit transaction. This
is not a reasonable and unnecessary limitation.

Also fix hang when parse command returns error.

diff --git a/src/context/pool_query_context.c b/src/context/pool_query_context.c
index 365429a3f4b7201603636243435e8b25568f208a..7b6d7da7466f5a380cca41f157ee62fcfb28e94c 100644 (file)
--- a/src/context/pool_query_context.c
+++ b/src/context/pool_query_context.c
@@ -1387,6 +1387,13 @@ static POOL_DEST send_to_where(Node *node, char *query)
                         */
                        return POOL_PRIMARY; 
                }
+               /*
+                * SHOW
+                */
+               else if (IsA(node, VariableShowStmt))
+               {
+                       return POOL_EITHER;
+               }
 
                /*
                 * Other statements are sent to primary

diff --git a/src/protocol/CommandComplete.c b/src/protocol/CommandComplete.c
index 8974c9e830a8c789e597f50d5b29863960fd1180..ac80ce1991a1e4e24af0ad32f6df33b549c92c47 100644 (file)
--- a/src/protocol/CommandComplete.c
+++ b/src/protocol/CommandComplete.c
@@ -68,6 +68,7 @@ POOL_STATUS CommandComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
        /*
         * If operated in streaming replication mode and doing an extended query,
         * read backend message according to the query context.
+        * Also we set the transaction state at this point.
         */
        if (STREAM && pool_is_doing_extended_query_message())
        {
@@ -89,6 +90,14 @@ POOL_STATUS CommandComplete(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *bac
                                        return POOL_END;
                                p1 = palloc(len);
                                memcpy(p1, p, len);
+
+                               if (session_context->query_context->parse_tree &&
+                                       is_start_transaction_query(session_context->query_context->parse_tree))
+                                       TSTATE(backend, i) ='T';                /* we are inside a transaction */
+
+                                       ereport(DEBUG1,
+                                                       (errmsg("processing command complete"),
+                                                        errdetail("set transaction state to T")));
                        }
                }
        }

diff --git a/src/protocol/pool_process_query.c b/src/protocol/pool_process_query.c
index 93356ebae371ecc5814167c882e3db9e3f13d988..567ff06517c78973157cce5090e2b1d64b7e7c44 100644 (file)
--- a/src/protocol/pool_process_query.c
+++ b/src/protocol/pool_process_query.c
@@ -1080,6 +1080,7 @@ static int
  * - COPY TO STDOUT
  * - EXPLAIN
  * - EXPLAIN ANALYZE and query is SELECT not including writing functions
+ * - SHOW
  *
  * Note that for SELECT this function returns false.
  */
@@ -1200,6 +1201,9 @@ bool is_select_query(Node *node, char *sql)
                                return true;
                }
        }
+       else if (IsA(node, VariableShowStmt))           /* SHOW command */
+               return true;
+
        return false;
 }
 

diff --git a/src/protocol/pool_proto_modules.c b/src/protocol/pool_proto_modules.c
index 8c541b26d9c1bec076aef7c22de630b87e822e52..00350dfe2928d67cd67a82363089422f4404bbac 100644 (file)
--- a/src/protocol/pool_proto_modules.c
+++ b/src/protocol/pool_proto_modules.c
@@ -2639,10 +2639,30 @@ POOL_STATUS ProcessBackendResponse(POOL_CONNECTION *frontend,
                                        pool_set_ignore_till_sync();
                                        pool_unset_query_in_progress();
 
-                                       /* Remove all pending messages */
-                                       while (pool_pending_message_pull_out())
-                                               ;
-                                       pool_pending_message_reset_previous_message();
+                                       if (STREAM)
+                                       {
+                                               POOL_PENDING_MESSAGE *pmsg;
+                                               int i;
+
+                                               /* Remove all pending messages */
+                                               do
+                                               {
+                                                       pmsg = pool_pending_message_pull_out();
+                                                       pool_pending_message_free_pending_message(pmsg);
+                                               }
+                                               while (pmsg);
+
+                                               pool_pending_message_reset_previous_message();
+
+                                               /* Discard read buffer */
+                                               for (i=0;i<NUM_BACKENDS;i++)
+                                               {
+                                                       if (VALID_BACKEND(i))
+                                                       {
+                                                               pool_discard_read_buffer(CONNECTION(backend, i));
+                                                       }
+                                               }
+                                       }
                                }
                                break;