ckdb - use psql cursors for larger tables

src/ckdb.h

@@ -55,7 +55,7 @@
 
 #define DB_VLOCK "1"
 #define DB_VERSION "1.0.4"
-#define CKDB_VERSION DB_VERSION"-1.403"
+#define CKDB_VERSION DB_VERSION"-1.500"
 
 #define WHERE_FFL " - from %s %s() line %d"
 #define WHERE_FFL_HERE __FILE__, __func__, __LINE__

src/ckdb_dbio.c

@@ -41,8 +41,13 @@ char *pqerrmsg(PGconn *conn)
 		if (__col == -1) { \
 			LOGERR("%s(): Unknown field '%s' row %d", __func__, __name, __row); \
 			__ok = false; \
-		} else \
+		} else { \
 			__fld = PQgetvalue(__res, __row, __col); \
+			if (__fld == NULL) { \
+				LOGERR("%s(): Invalid field '%s' or row %d", __func__, __name, __row); \
+				__ok = false; \
+			}\
+		} \
 	} while (0)
 
 // HISTORY FIELDS
@@ -1561,26 +1566,44 @@ bool workers_fill(PGconn *conn)
 	ExecStatusType rescode;
 	PGresult *res;
 	K_ITEM *item;
-	int n, i;
+	int n, t, i;
 	WORKERS *row;
 	char *field;
 	char *sel;
 	int fields = 7;
-	bool ok;
+	bool ok = false;
 
 	LOGDEBUG("%s(): select", __func__);
 
-	sel = "select "
+	sel = "declare wk cursor for select "
 		"userid,workername,difficultydefault,"
 		"idlenotificationenabled,idlenotificationtime,workerbits"
 		HISTORYDATECONTROL
 		",workerid from workers";
+	res = PQexec(conn, "Begin", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	PQclear(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Begin", rescode, conn);
+		return false;
+	}
+
 	res = PQexec(conn, sel, CKPQ_READ);
 	rescode = PQresultStatus(res);
+	PQclear(res);
 	if (!PGOK(rescode)) {
-		PGLOGERR("Select", rescode, conn);
+		PGLOGERR("Declare", rescode, conn);
+		goto flail;
+	}
+
+	LOGDEBUG("%s(): fetching ...", __func__);
+
+	res = PQexec(conn, "fetch 1 in wk", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Fetch first", rescode, conn);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
 	n = PQnfields(res);
@@ -1588,14 +1611,14 @@ bool workers_fill(PGconn *conn)
 		LOGERR("%s(): Invalid field count - should be %d, but is %d",
 			__func__, fields + HISTORYDATECOUNT, n);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
-	n = PQntuples(res);
+	n = 0;
-	LOGDEBUG("%s(): tree build count %d", __func__, n);
 	ok = true;
 	K_WLOCK(workers_free);
-	for (i = 0; i < n; i++) {
+	while ((t = PQntuples(res)) > 0) {
+		for (i = 0; i < t; i++) {
 			item = k_unlink_head(workers_free);
 			DATA_WORKERS(row, item);
 			bzero(row, sizeof(*row));
@@ -1653,16 +1676,30 @@ bool workers_fill(PGconn *conn)
 			 *  other tables */
 			find_create_workerstatus(false, row->userid, row->workername,
 						 __FILE__, __func__, __LINE__);
+			tick();
+			n++;
+		}
+		PQclear(res);
+		res = PQexec(conn, "fetch 9999 in wk", CKPQ_READ);
+		rescode = PQresultStatus(res);
+		if (!PGOK(rescode)) {
+			PGLOGERR("Fetch next", rescode, conn);
+			ok = false;
+			break;
+		}
 	}
 	if (!ok)
 		k_add_head(workers_free, item);
 
 	K_WUNLOCK(workers_free);
 	PQclear(res);
+flail:
+	res = PQexec(conn, "Commit", CKPQ_READ);
+	PQclear(res);
 
 	if (ok) {
 		LOGDEBUG("%s(): built", __func__);
-		LOGWARNING("%s(): loaded %d workers records", __func__, n);
+		LOGWARNING("%s(): fetched %d workers records", __func__, n);
 	}
 
 	return ok;
@@ -2160,25 +2197,43 @@ bool payments_fill(PGconn *conn)
 	PGresult *res;
 	K_ITEM *item;
 	PAYMENTS *row;
-	int n, i;
+	int n, t, i;
 	char *field;
 	char *sel;
 	int fields = 11;
-	bool ok;
+	bool ok = false;
 
 	LOGDEBUG("%s(): select", __func__);
 
-	sel = "select "
+	sel = "declare ps cursor for select "
 		"paymentid,payoutid,userid,subname,paydate,payaddress,"
 		"originaltxn,amount,diffacc,committxn,commitblockhash"
 		HISTORYDATECONTROL
 		" from payments";
+	res = PQexec(conn, "Begin", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	PQclear(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Begin", rescode, conn);
+		return false;
+	}
+
 	res = PQexec(conn, sel, CKPQ_READ);
 	rescode = PQresultStatus(res);
+	PQclear(res);
 	if (!PGOK(rescode)) {
-		PGLOGERR("Select", rescode, conn);
+		PGLOGERR("Declare", rescode, conn);
+		goto flail;
+	}
+
+	LOGDEBUG("%s(): fetching ...", __func__);
+
+	res = PQexec(conn, "fetch 1 in ps", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Fetch first", rescode, conn);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
 	n = PQnfields(res);
@@ -2186,14 +2241,14 @@ bool payments_fill(PGconn *conn)
 		LOGERR("%s(): Invalid field count - should be %d, but is %d",
 			__func__, fields + HISTORYDATECOUNT, n);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
-	n = PQntuples(res);
+	n = 0;
-	LOGDEBUG("%s(): tree build count %d", __func__, n);
 	ok = true;
 	K_WLOCK(payments_free);
-	for (i = 0; i < n; i++) {
+	while ((t = PQntuples(res)) > 0) {
+		for (i = 0; i < t; i++) {
 			item = k_unlink_head(payments_free);
 			DATA_PAYMENTS(row, item);
 			bzero(row, sizeof(*row));
@@ -2264,16 +2319,30 @@ bool payments_fill(PGconn *conn)
 
 			add_to_ktree(payments_root, item);
 			k_add_head(payments_store, item);
+			tick();
+			n++;
+		}
+		PQclear(res);
+		res = PQexec(conn, "fetch 9999 in ps", CKPQ_READ);
+		rescode = PQresultStatus(res);
+		if (!PGOK(rescode)) {
+			PGLOGERR("Fetch next", rescode, conn);
+			ok = false;
+			break;
+		}
 	}
 	if (!ok)
 		k_add_head(payments_free, item);
 
 	K_WUNLOCK(payments_free);
 	PQclear(res);
+flail:
+	res = PQexec(conn, "Commit", CKPQ_READ);
+	PQclear(res);
 
 	if (ok) {
 		LOGDEBUG("%s(): built", __func__);
-		LOGWARNING("%s(): loaded %d payments records", __func__, n);
+		LOGWARNING("%s(): fetched %d payments records", __func__, n);
 	}
 
 	return ok;
@@ -2772,12 +2841,12 @@ bool workinfo_fill(PGconn *conn)
 	K_ITEM *item;
 	WORKINFO *row;
 	char *params[3];
-	int n, i, par = 0;
+	int n, t, i, par = 0;
 	char *field;
 	char *sel = NULL;
 	size_t len, off;
 	int fields = 10;
-	bool ok;
+	bool ok = false;
 
 	LOGDEBUG("%s(): select", __func__);
 
@@ -2790,7 +2859,7 @@ bool workinfo_fill(PGconn *conn)
 
 	APPEND_REALLOC_INIT(sel, off, len);
 	APPEND_REALLOC(sel, off, len,
-			"select "
+			"declare wi cursor for select "
 //			"workinfoid,poolinstance,transactiontree,merklehash,prevhash,"
 			"workinfoid,poolinstance,merklehash,prevhash,"
 			"coinbase1,coinbase2,version,bits,ntime,reward"
@@ -2820,12 +2889,31 @@ bool workinfo_fill(PGconn *conn)
 		params[par++] = bigint_to_buf(dbload_workinfoid_start, NULL, 0);
 	params[par++] = bigint_to_buf(dbload_workinfoid_finish, NULL, 0);
 	PARCHK(par, params);
+
+	res = PQexec(conn, "Begin", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	PQclear(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Begin", rescode, conn);
+		return false;
+	}
+
 	res = PQexecParams(conn, sel, par, NULL, (const char **)params, NULL, NULL, 0, CKPQ_READ);
 	rescode = PQresultStatus(res);
+	PQclear(res);
 	if (!PGOK(rescode)) {
-		PGLOGERR("Select", rescode, conn);
+		PGLOGERR("Declare", rescode, conn);
+		goto flail;
+	}
+
+	LOGDEBUG("%s(): fetching ...", __func__);
+
+	res = PQexec(conn, "fetch 1 in wi", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Fetch first", rescode, conn);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
 	n = PQnfields(res);
@@ -2833,14 +2921,14 @@ bool workinfo_fill(PGconn *conn)
 		LOGERR("%s(): Invalid field count - should be %d, but is %d",
 			__func__, fields + HISTORYDATECOUNT, n);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
-	n = PQntuples(res);
+	n = 0;
-	LOGDEBUG("%s(): tree build count %d", __func__, n);
 	ok = true;
 	//K_WLOCK(workinfo_free);
-	for (i = 0; i < n; i++) {
+	while ((t = PQntuples(res)) > 0) {
+		for (i = 0; i < t; i++) {
 			item = k_unlink_head(workinfo_free);
 			DATA_WORKINFO(row, item);
 			bzero(row, sizeof(*row));
@@ -2925,14 +3013,23 @@ bool workinfo_fill(PGconn *conn)
 				dbstatus.newest_workinfoid = row->workinfoid;
 			}
 
-		if (i == 0 || ((i+1) % 100000) == 0) {
+			if (n == 0 || ((n+1) % 100000) == 0) {
 				printf(TICK_PREFIX"wi ");
-			pcom(i+1);
+				pcom(n+1);
 				putchar('\r');
 				fflush(stdout);
 			}
-
 			tick();
+			n++;
+		}
+		PQclear(res);
+		res = PQexec(conn, "fetch 9999 in wi", CKPQ_READ);
+		rescode = PQresultStatus(res);
+		if (!PGOK(rescode)) {
+			PGLOGERR("Fetch next", rescode, conn);
+			ok = false;
+			break;
+		}
 	}
 	if (!ok) {
 		free_workinfo_data(item);
@@ -2942,10 +3039,14 @@ bool workinfo_fill(PGconn *conn)
 
 	//K_WUNLOCK(workinfo_free);
 	PQclear(res);
+flail:
+	res = PQexec(conn, "Commit", CKPQ_READ);
+	PQclear(res);
+
 
 	if (ok) {
 		LOGDEBUG("%s(): built", __func__);
-		LOGWARNING("%s(): loaded %d workinfo records", __func__, n);
+		LOGWARNING("%s(): fetched %d workinfo records", __func__, n);
 	}
 
 	return ok;
@@ -5223,24 +5324,42 @@ bool miningpayouts_fill(PGconn *conn)
 	PGresult *res;
 	K_ITEM *item;
 	MININGPAYOUTS *row;
-	int n, i;
+	int n, t, i;
 	char *field;
 	char *sel;
 	int fields = 4;
-	bool ok;
+	bool ok = false;
 
 	LOGDEBUG("%s(): select", __func__);
 
-	sel = "select "
+	sel = "declare mp cursor for select "
 		"payoutid,userid,diffacc,amount"
 		HISTORYDATECONTROL
 		" from miningpayouts";
+	res = PQexec(conn, "Begin", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	PQclear(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Begin", rescode, conn);
+		return false;
+	}
+
 	res = PQexec(conn, sel, CKPQ_READ);
 	rescode = PQresultStatus(res);
+	PQclear(res);
 	if (!PGOK(rescode)) {
-		PGLOGERR("Select", rescode, conn);
+		PGLOGERR("Declare", rescode, conn);
+		goto flail;
+	}
+
+	LOGDEBUG("%s(): fetching ...", __func__);
+
+	res = PQexec(conn, "fetch 1 in mp", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Fetch first", rescode, conn);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
 	n = PQnfields(res);
@@ -5248,14 +5367,14 @@ bool miningpayouts_fill(PGconn *conn)
 		LOGERR("%s(): Invalid field count - should be %d, but is %d",
 			__func__, fields + HISTORYDATECOUNT, n);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
-	n = PQntuples(res);
+	n = 0;
-	LOGDEBUG("%s(): tree build count %d", __func__, n);
 	ok = true;
 	K_WLOCK(miningpayouts_free);
-	for (i = 0; i < n; i++) {
+	while ((t = PQntuples(res)) > 0) {
+		for (i = 0; i < t; i++) {
 			item = k_unlink_head(miningpayouts_free);
 			DATA_MININGPAYOUTS(row, item);
 			bzero(row, sizeof(*row));
@@ -5291,18 +5410,30 @@ bool miningpayouts_fill(PGconn *conn)
 
 			add_to_ktree(miningpayouts_root, item);
 			k_add_head(miningpayouts_store, item);
-
 			tick();
+			n++;
+		}
+		PQclear(res);
+		res = PQexec(conn, "fetch 9999 in mp", CKPQ_READ);
+		rescode = PQresultStatus(res);
+		if (!PGOK(rescode)) {
+			PGLOGERR("Fetch next", rescode, conn);
+			ok = false;
+			break;
+		}
 	}
 	if (!ok)
 		k_add_head(miningpayouts_free, item);
 
 	K_WUNLOCK(miningpayouts_free);
 	PQclear(res);
+flail:
+	res = PQexec(conn, "Commit", CKPQ_READ);
+	PQclear(res);
 
 	if (ok) {
 		LOGDEBUG("%s(): built", __func__);
-		LOGWARNING("%s(): loaded %d miningpayout records", __func__, n);
+		LOGWARNING("%s(): fetched %d miningpayout records", __func__, n);
 	}
 
 	return ok;
@@ -6521,12 +6652,12 @@ bool markersummary_fill(PGconn *conn)
 	ExecStatusType rescode;
 	PGresult *res;
 	K_ITEM *item, *p_item;
-	int n, i, p_n;
+	int n, t, i, p_n;
 	MARKERSUMMARY *row, *p_row;
 	char *field;
 	char *sel;
 	int fields = 20;
-	bool ok;
+	bool ok = false;
 
 	LOGDEBUG("%s(): select", __func__);
 
@@ -6534,19 +6665,37 @@ bool markersummary_fill(PGconn *conn)
 	fflush(stdout);
 
 	// TODO: limit how far back
-	sel = "select "
+	sel = "declare ws cursor for select "
 		"markerid,userid,workername,diffacc,diffsta,diffdup,diffhi,"
 		"diffrej,shareacc,sharesta,sharedup,sharehi,sharerej,"
 		"sharecount,errorcount,firstshare,lastshare,firstshareacc,"
 		"lastshareacc,lastdiffacc"
 		MODIFYDATECONTROL
 		" from markersummary";
+	res = PQexec(conn, "Begin", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	PQclear(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Begin", rescode, conn);
+		return false;
+	}
+
 	res = PQexec(conn, sel, CKPQ_READ);
 	rescode = PQresultStatus(res);
+	PQclear(res);
 	if (!PGOK(rescode)) {
-		PGLOGERR("Select", rescode, conn);
+		PGLOGERR("Declare", rescode, conn);
+		goto flail;
+	}
+
+	LOGDEBUG("%s(): fetching ...", __func__);
+
+	res = PQexec(conn, "fetch 1 in ws", CKPQ_READ);
+	rescode = PQresultStatus(res);
+	if (!PGOK(rescode)) {
+		PGLOGERR("Fetch first", rescode, conn);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
 	n = PQnfields(res);
@@ -6554,14 +6703,14 @@ bool markersummary_fill(PGconn *conn)
 		LOGERR("%s(): Invalid field count - should be %d, but is %d",
 			__func__, fields + MODIFYDATECOUNT, n);
 		PQclear(res);
-		return false;
+		goto flail;
 	}
 
-	n = PQntuples(res);
+	n = 0;
-	LOGDEBUG("%s(): tree build count %d", __func__, n);
 	ok = true;
 	//K_WLOCK(markersummary_free);
-	for (i = 0; i < n; i++) {
+	while ((t = PQntuples(res)) > 0) {
+		for (i = 0; i < t; i++) {
 			item = k_unlink_head(markersummary_free);
 			DATA_MARKERSUMMARY(row, item);
 			bzero(row, sizeof(*row));
@@ -6709,14 +6858,23 @@ bool markersummary_fill(PGconn *conn)
 
 			_userinfo_update(NULL, NULL, row, false, false);
 
-		if (i == 0 || ((i+1) % 100000) == 0) {
+			if (n == 0 || ((n+1) % 100000) == 0) {
 				printf(TICK_PREFIX"ms ");
-			pcom(i+1);
+				pcom(n+1);
 				putchar('\r');
 				fflush(stdout);
 			}
-
 			tick();
+			n++;
+		}
+		PQclear(res);
+		res = PQexec(conn, "fetch 9999 in ws", CKPQ_READ);
+		rescode = PQresultStatus(res);
+		if (!PGOK(rescode)) {
+			PGLOGERR("Fetch next", rescode, conn);
+			ok = false;
+			break;
+		}
 	}
 	if (!ok) {
 		free_markersummary_data(item);
@@ -6727,10 +6885,13 @@ bool markersummary_fill(PGconn *conn)
 
 	//K_WUNLOCK(markersummary_free);
 	PQclear(res);
+flail:
+	res = PQexec(conn, "Commit", CKPQ_READ);
+	PQclear(res);
 
 	if (ok) {
 		LOGDEBUG("%s(): built", __func__);
-		LOGWARNING("%s(): loaded %d markersummary records", __func__, n);
+		LOGWARNING("%s(): fetched %d markersummary records", __func__, n);
 		LOGWARNING("%s(): created %d markersummary pool records", __func__, p_n);
 	}