Fix bugs in postgres_fdw

Author: knizhnik

contrib/pg_tsdtm/pg_dtm.c

@@ -72,8 +72,8 @@ typedef struct
 } DtmTransId;
 
 
-#define DTM_TRACE(x) 
-//#define DTM_TRACE(x) fprintf x
+//#define DTM_TRACE(x) 
+#define DTM_TRACE(x) fprintf x
 
 static shmem_startup_hook_type prev_shmem_startup_hook;
 static HTAB* xid2status;
@@ -602,10 +602,10 @@ void DtmInitialize()
 
 void DtmLocalBegin(DtmCurrentTrans* x)
 {
-    if (x->xid == InvalidTransactionId) { 
+    if (!TransactionIdIsValid(x->xid)) { 
         SpinLockAcquire(&local->lock);
         x->xid = GetCurrentTransactionId();
-        Assert(x->xid != InvalidTransactionId);
+        Assert(TransactionIdIsValid(x->xid));
         x->cid = INVALID_CID;
         x->is_global = false;
         x->is_prepared = false;
@@ -648,7 +648,7 @@ cid_t DtmLocalAccess(DtmCurrentTrans* x, GlobalTransactionId gtid, cid_t global_
     }
     SpinLockRelease(&local->lock);
 	if (global_cid < local_cid - DtmVacuumDelay*USEC) { 
-		elog(ERROR, "Too old snapshot");
+		elog(ERROR, "Too old snapshot: requested %ld, current %ld", global_cid, local_cid);
 	}
     return global_cid;
 }
@@ -662,7 +662,7 @@ void DtmLocalBeginPrepare(GlobalTransactionId gtid)
 
         id = (DtmTransId*)hash_search(gtid2xid, gtid, HASH_FIND, NULL);
         Assert(id != NULL);
-
+		Assert(TransactionIdIsValid(id->xid));
         ts = (DtmTransStatus*)hash_search(xid2status, &id->xid, HASH_ENTER, NULL);
         ts->status = TRANSACTION_STATUS_IN_PROGRESS;
         ts->cid = dtm_get_cid();
@@ -743,9 +743,11 @@ void DtmLocalCommitPrepared(DtmCurrentTrans* x, GlobalTransactionId gtid)
 void DtmLocalCommit(DtmCurrentTrans* x)
 {
     SpinLockAcquire(&local->lock);
+	if (TransactionIdIsValid(x->xid))
     {
         bool found;
-        DtmTransStatus* ts = (DtmTransStatus*)hash_search(xid2status, &x->xid, HASH_ENTER, &found);
+        DtmTransStatus* ts;
+		ts = (DtmTransStatus*)hash_search(xid2status, &x->xid, HASH_ENTER, &found);
         ts->status = TRANSACTION_STATUS_COMMITTED;
         if (x->is_prepared) { 
             int i;
@@ -795,7 +797,9 @@ void DtmLocalAbort(DtmCurrentTrans* x)
     SpinLockAcquire(&local->lock);
     { 
         bool found;
-        DtmTransStatus* ts = (DtmTransStatus*)hash_search(xid2status, &x->xid, HASH_ENTER, &found);
+        DtmTransStatus* ts;
+		Assert(TransactionIdIsValid(x->xid));
+		ts = (DtmTransStatus*)hash_search(xid2status, &x->xid, HASH_ENTER, &found);
         if (x->is_prepared) { 
             Assert(found);
             Assert(x->is_global);
@@ -865,7 +869,9 @@ static void DtmAddSubtransactions(DtmTransStatus* ts, TransactionId* subxids, in
     int i;
     for (i = 0; i < nSubxids; i++) { 
         bool found;
-        DtmTransStatus* sts = (DtmTransStatus*)hash_search(xid2status, &subxids[i], HASH_ENTER, &found);
+		DtmTransStatus* sts;
+		Assert(TransactionIdIsValid(subxids[i]));
+        sts = (DtmTransStatus*)hash_search(xid2status, &subxids[i], HASH_ENTER, &found);
         Assert(!found);
         sts->status = ts->status;
         sts->cid = ts->cid;

contrib/postgres_fdw/connection.c

@@ -370,6 +370,7 @@ do_sql_send_command(PGconn *conn, const char *sql)
 {
 	if (PQsendQuery(conn, sql) != PGRES_COMMAND_OK) {
 		PGresult *res = PQgetResult(conn);
+		elog(WARNING, "Failed to send command %s", sql);
 		pgfdw_report_error(ERROR, res, conn, true, sql);
 		PQclear(res);
 	}
@@ -587,20 +588,27 @@ static bool RunDtmStatement(char const* sql, unsigned expectedStatus, DtmCommand
 	hash_seq_init(&scan, ConnectionHash);
 	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
 	{
-		do_sql_send_command(entry->conn, sql);
+		if (entry->xact_depth > 0)
+		{
+			do_sql_send_command(entry->conn, sql);
+		}
 	}
 
 	hash_seq_init(&scan, ConnectionHash);
 	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
 	{
-		PGresult *result = PQgetResult(entry->conn);
-		if (PQresultStatus(result) != expectedStatus || (handler && !handler(result, arg)))
+		if (entry->xact_depth > 0)
 		{
-			pgfdw_report_error(ERROR, result, entry->conn, true, sql);
-			allOk = false;
+			PGresult *result = PQgetResult(entry->conn);
+			if (PQresultStatus(result) != expectedStatus || (handler && !handler(result, arg)))
+			{
+				elog(WARNING, "Failed command %s: status=%d, expected status=%d", sql, PQresultStatus(result), expectedStatus);
+				pgfdw_report_error(ERROR, result, entry->conn, true, sql);
+				allOk = false;
+			}
+			PQclear(result);
+			PQgetResult(entry->conn); /* consume NULL result */
 		}
-		PQclear(result);
-		PQgetResult(entry->conn); /* consume NULL result */
 	}
 	return allOk;
 }

contrib/postgres_fdw/tests/dtmbench.cpp

@@ -156,14 +156,14 @@ void* writer(void* arg)
 void initializeDatabase()
 {
     connection conn(cfg.connection);
-    work txn(conn);
 	int accountsPerShard = (cfg.nAccounts + cfg.nShards - 1)/cfg.nShards;
 	for (int i = 0; i < cfg.nShards; i++) 
 	{ 
+		work txn(conn);
 		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();
 	}
-    txn.commit();
 }
 
 int main (int argc, char* argv[])