FDW bug fixes

Author: knizhnik

contrib/pg_shard/bench/dtmbench.cpp

@@ -158,7 +158,7 @@ void initializeDatabase()
 	exec(txn, "CREATE EXTENSION pg_shard");
 	exec(txn, "create table t(u int primary key, v int)");
 	exec(txn, "SELECT master_create_distributed_table(table_name := 't', partition_column := 'u')");
-	exec(txn, "SELECT master_create_worker_shards(table_name := 't', shard_count := 100, replication_factor := 1)");
+	exec(txn, "SELECT master_create_worker_shards(table_name := 't', shard_count := 9, replication_factor := 1)");
 	for (int i = 0; i < cfg.nAccounts; i++) { 
 		exec(txn, "insert into t values (%d,0)", i);
 	}

contrib/postgres_fdw/connection.c

@@ -650,8 +650,8 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 	{
 		switch (event)
 		{
-		  case XACT_EVENT_PARALLEL_COMMIT:
-		  case XACT_EVENT_COMMIT:
+		  case XACT_EVENT_PARALLEL_PRE_COMMIT:
+		  case XACT_EVENT_PRE_COMMIT:
 		  {
 			  csn_t maxCSN = 0;
 
@@ -668,160 +668,134 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 			  {
 				  RunDtmCommand(psprintf("ROLLBACK PREPARED '%d.%d'", 
 										 MyProcPid, currentLocalTransactionId));
-			  }
-			  break;
+				  ereport(ERROR,
+						  (errcode(ERRCODE_TRANSACTION_ROLLBACK),
+						   errmsg("transaction was aborted at one of the shards")));
+				  break;
+			  } 
+			  return;
 		  }
-		  case XACT_EVENT_ABORT:
-			  RunDtmCommand("ROLLBACK"); 
-			  break;
-		  case XACT_EVENT_PRE_PREPARE:
-  			  ereport(ERROR,
-					  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					   errmsg("cannot prepare a transaction that modified remote tables")));
-			  break;
 		  default:
-  			  break;
+   			  break;
 		}
-
-		currentGlobalTransactionId = 0;
-
-		hash_seq_init(&scan, ConnectionHash);
-		while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
-		{
-			/* Ignore cache entry if no open connection right now */
-			if (entry->conn == NULL)
-				continue;
-			/* Reset state to show we're out of a transaction */
-			entry->xact_depth = 0;
-			
-			/*
-			 * If the connection isn't in a good idle state, discard it to
-			 * recover. Next GetConnection will open a new connection.
-			 */
-			if (PQstatus(entry->conn) != CONNECTION_OK ||
-				PQtransactionStatus(entry->conn) != PQTRANS_IDLE)
-			{
-				elog(DEBUG3, "discarding connection %p, conn status=%d, trans status=%d", entry->conn, PQstatus(entry->conn), PQtransactionStatus(entry->conn));
-				PQfinish(entry->conn);
-				entry->conn = NULL;
-			}
-		}
-	} else { 
-		/*
-		 * Scan all connection cache entries to find open remote transactions, and
-		 * close them.
-		 */
-		hash_seq_init(&scan, ConnectionHash);
-		while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
-		{
-			PGresult   *res;
+	}
+	/*
+	 * Scan all connection cache entries to find open remote transactions, and
+	 * close them.
+	 */
+	hash_seq_init(&scan, ConnectionHash);
+	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+	{
+		PGresult   *res;
 
-			/* Ignore cache entry if no open connection right now */
-			if (entry->conn == NULL)
-				continue;
+		/* Ignore cache entry if no open connection right now */
+		if (entry->conn == NULL)
+			continue;
 
-			/* If it has an open remote transaction, try to close it */
-			if (entry->xact_depth > 0)
-			{
-				elog(DEBUG3, "closing remote transaction on connection %p event %d",
-					 entry->conn, event);
+		/* If it has an open remote transaction, try to close it */
+		if (entry->xact_depth > 0)
+		{
+			elog(DEBUG3, "closing remote transaction on connection %p event %d",
+				 entry->conn, event);
 
-				switch (event)
-				{
-				    case XACT_EVENT_PARALLEL_PRE_COMMIT:
-   				    case XACT_EVENT_PRE_COMMIT:
-					    /* Commit all remote transactions during pre-commit */
-						do_sql_send_command(entry->conn, "COMMIT TRANSACTION");
-						continue;
+			switch (event)
+			{
+			    case XACT_EVENT_PARALLEL_PRE_COMMIT:
+			    case XACT_EVENT_PRE_COMMIT:
+				    /* Commit all remote transactions during pre-commit */
+					do_sql_send_command(entry->conn, "COMMIT TRANSACTION");
+					continue;
 
-			        case XACT_EVENT_PRE_PREPARE:
-  					    /*
-						 * We disallow remote transactions that modified anything,
-						 * since it's not very reasonable to hold them open until
-						 * the prepared transaction is committed.  For the moment,
-						 * throw error unconditionally; later we might allow
-						 * read-only cases.  Note that the error will cause us to
-						 * come right back here with event == XACT_EVENT_ABORT, so
-						 * we'll clean up the connection state at that point.
-						 */
-					  ereport(ERROR,
-							  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-							   errmsg("cannot prepare a transaction that modified remote tables")));
-					  break;
-
-				    case XACT_EVENT_PARALLEL_COMMIT:
-				    case XACT_EVENT_COMMIT:
-				    case XACT_EVENT_PREPARE:
-					    do_sql_wait_command(entry->conn, "COMMIT TRANSACTION");
-					    /*
-						 * If there were any errors in subtransactions, and we
-						 * made prepared statements, do a DEALLOCATE ALL to make
-						 * sure we get rid of all prepared statements. This is
-						 * annoying and not terribly bulletproof, but it's
-						 * probably not worth trying harder.
-						 *
-						 * DEALLOCATE ALL only exists in 8.3 and later, so this
-						 * constrains how old a server postgres_fdw can
-						 * communicate with.  We intentionally ignore errors in
-						 * the DEALLOCATE, so that we can hobble along to some
-						 * extent with older servers (leaking prepared statements
-						 * as we go; but we don't really support update operations
-						 * pre-8.3 anyway).
-						 */
+		        case XACT_EVENT_PRE_PREPARE:
+				    /*
+					 * We disallow remote transactions that modified anything,
+					 * since it's not very reasonable to hold them open until
+					 * the prepared transaction is committed.  For the moment,
+					 * throw error unconditionally; later we might allow
+					 * read-only cases.  Note that the error will cause us to
+					 * come right back here with event == XACT_EVENT_ABORT, so
+					 * we'll clean up the connection state at that point.
+					 */
+  				    ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("cannot prepare a transaction that modified remote tables")));
+					break;
+
+			    case XACT_EVENT_PARALLEL_COMMIT:
+				case XACT_EVENT_COMMIT:
+			    case XACT_EVENT_PREPARE:
+				    if (!currentGlobalTransactionId)
+					{
+						do_sql_wait_command(entry->conn, "COMMIT TRANSACTION");
+					}
+					/*
+					 * If there were any errors in subtransactions, and we
+					 * made prepared statements, do a DEALLOCATE ALL to make
+					 * sure we get rid of all prepared statements. This is
+					 * annoying and not terribly bulletproof, but it's
+					 * probably not worth trying harder.
+					 *
+					 * DEALLOCATE ALL only exists in 8.3 and later, so this
+					 * constrains how old a server postgres_fdw can
+					 * communicate with.  We intentionally ignore errors in
+					 * the DEALLOCATE, so that we can hobble along to some
+					 * extent with older servers (leaking prepared statements
+					 * as we go; but we don't really support update operations
+					 * pre-8.3 anyway).
+					 */
+					if (entry->have_prep_stmt && entry->have_error)
+					{
+						res = PQexec(entry->conn, "DEALLOCATE ALL");
+						PQclear(res);
+					}
+					entry->have_prep_stmt = false;
+					entry->have_error = false;
+					break;
+					
+			    case XACT_EVENT_PARALLEL_ABORT:
+			    case XACT_EVENT_ABORT:
+				    /* Assume we might have lost track of prepared statements */
+				    entry->have_error = true;
+					/* If we're aborting, abort all remote transactions too */
+					res = PQexec(entry->conn, "ABORT TRANSACTION");
+					/* Note: can't throw ERROR, it would be infinite loop */
+					if (PQresultStatus(res) != PGRES_COMMAND_OK)
+						pgfdw_report_error(WARNING, res, entry->conn, true,
+										   "ABORT TRANSACTION");
+					else
+					{
+						PQclear(res);
+						/* As above, make sure to clear any prepared stmts */
 						if (entry->have_prep_stmt && entry->have_error)
 						{
 							res = PQexec(entry->conn, "DEALLOCATE ALL");
 							PQclear(res);
 						}
 						entry->have_prep_stmt = false;
 						entry->have_error = false;
-						break;
-						
-  				    case XACT_EVENT_PARALLEL_ABORT:
-				    case XACT_EVENT_ABORT:
-					    /* Assume we might have lost track of prepared statements */
-					    entry->have_error = true;
-						/* If we're aborting, abort all remote transactions too */
-						res = PQexec(entry->conn, "ABORT TRANSACTION");
-						/* Note: can't throw ERROR, it would be infinite loop */
-						if (PQresultStatus(res) != PGRES_COMMAND_OK)
-							pgfdw_report_error(WARNING, res, entry->conn, true,
-											   "ABORT TRANSACTION");
-						else
-						{
-							PQclear(res);
-							/* As above, make sure to clear any prepared stmts */
-							if (entry->have_prep_stmt && entry->have_error)
-							{
-								res = PQexec(entry->conn, "DEALLOCATE ALL");
-								PQclear(res);
-							}
-							entry->have_prep_stmt = false;
-							entry->have_error = false;
-						}
-						break;
-
- 				    case XACT_EVENT_START:
-				    case XACT_EVENT_ABORT_PREPARED:
-				    case XACT_EVENT_COMMIT_PREPARED:
-					   break;					
-				}
-			}
-			/* Reset state to show we're out of a transaction */
-			entry->xact_depth = 0;
-			
-			/*
-			 * If the connection isn't in a good idle state, discard it to
-			 * recover. Next GetConnection will open a new connection.
-			 */
-			if (PQstatus(entry->conn) != CONNECTION_OK ||
-				PQtransactionStatus(entry->conn) != PQTRANS_IDLE)
-			{
-				elog(DEBUG3, "discarding connection %p, conn status=%d, trans status=%d", entry->conn, PQstatus(entry->conn), PQtransactionStatus(entry->conn));
-				PQfinish(entry->conn);
-				entry->conn = NULL;
+					}
+					break;
+					
+ 				case XACT_EVENT_START:
+			    case XACT_EVENT_ABORT_PREPARED:
+			    case XACT_EVENT_COMMIT_PREPARED:
+				    break;					
 			}
 		}
+		/* Reset state to show we're out of a transaction */
+		entry->xact_depth = 0;
+
+		/*
+		 * If the connection isn't in a good idle state, discard it to
+		 * recover. Next GetConnection will open a new connection.
+		 */
+		if (PQstatus(entry->conn) != CONNECTION_OK ||
+			PQtransactionStatus(entry->conn) != PQTRANS_IDLE)
+		{
+			elog(WARNING, "discarding connection %p, conn status=%d, trans status=%d", entry->conn, PQstatus(entry->conn), PQtransactionStatus(entry->conn));
+			PQfinish(entry->conn);
+			entry->conn = NULL;
+		}
 	}
 	if (event != XACT_EVENT_PARALLEL_PRE_COMMIT && event != XACT_EVENT_PRE_COMMIT) { 
 		/*
@@ -833,6 +807,8 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 
 		/* Also reset cursor numbering for next transaction */
 		cursor_number = 0;
+
+		currentGlobalTransactionId = 0;
 	}
 }
 

contrib/postgres_fdw/tests/dtmbench.cpp

@@ -160,8 +160,8 @@ void initializeDatabase()
 	int accountsPerShard = (cfg.nAccounts + cfg.nShards - 1)/cfg.nShards;
 	for (int i = 0; i < cfg.nShards; i++) 
 	{ 
-		exec(txn, "alter table t_fdw%i add check (u between %d and %d)", i+1, accountsPerShard*i, accountsPerShard-1);
-		exec(txn, "insert into t_fdw%i (select generate_series(%d,%d), %d)", i+1, accountsPerShard*i, accountsPerShard-1, 0);
+		exec(txn, "alter table t_fdw%i add check (u between %d and %d)", i+1, accountsPerShard*i, accountsPerShard*(i+1)-1);
+		exec(txn, "insert into t_fdw%i (select generate_series(%d,%d), %d)", i+1, accountsPerShard*i, accountsPerShard*(i+1)-1, 0);
 	}
     txn.commit();
 }