external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add source_lsn and target_lsn fields into get_rebalance_progress (#6364) 

DESCRIPTION: Adds source_lsn and target_lsn fields into
get_rebalance_progress

Adding two fields named `source_lsn` and `target_lsn` to the function
`get_rebalance_progress`.
Target lsn data is fetched in `GetShardStatistics`, by expanding the
query sent to workers (joining with pg_subscription_rel and
pg_stat_subscription). Then put into the hashmap, for each shard.
Source lsn data is fetched in `BuildWorkerShardStatististicsHash`, in
the loop that iterate each node, by sending a pg_current_wal_lsn query
to each node. Then put into the hashmap, for each node.
pull/6387/head
Ahmet Gedemenli 2022-10-05 11:12:24 +03:00 committed by GitHub
parent e0f8666131
commit e36890ce55
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 256 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
src/backend/distributed/operations/shard_rebalancer.c
View File

@ -39,6 +39,7 @@
#include "distributed/metadata_cache.h" #include "distributed/metadata_cache.h"
#include "distributed/metadata_utility.h" #include "distributed/metadata_utility.h"
#include "distributed/multi_client_executor.h" #include "distributed/multi_client_executor.h"
#include "distributed/multi_logical_replication.h"
#include "distributed/multi_progress.h" #include "distributed/multi_progress.h"
#include "distributed/multi_server_executor.h" #include "distributed/multi_server_executor.h"
#include "distributed/pg_dist_rebalance_strategy.h" #include "distributed/pg_dist_rebalance_strategy.h"
@ -60,6 +61,7 @@
#include "utils/json.h" #include "utils/json.h"
#include "utils/lsyscache.h" #include "utils/lsyscache.h"
#include "utils/memutils.h" #include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/syscache.h" #include "utils/syscache.h"
#include "common/hashfn.h" #include "common/hashfn.h"
@ -166,6 +168,7 @@ typedef struct ShardStatistics
/* The shard its size in bytes. */ /* The shard its size in bytes. */
uint64 totalSize; uint64 totalSize;
XLogRecPtr shardLSN;
} ShardStatistics; } ShardStatistics;
/* /*
@ -175,6 +178,7 @@ typedef struct ShardStatistics
typedef struct WorkerShardStatistics typedef struct WorkerShardStatistics
{ {
WorkerHashKey worker; WorkerHashKey worker;
XLogRecPtr workerLSN;
/* /*
* Statistics for each shard on this worker: * Statistics for each shard on this worker:
@ -246,6 +250,10 @@ static HTAB * GetMovedShardIdsByWorker(PlacementUpdateEventProgress *steps,
int stepCount, bool fromSource); int stepCount, bool fromSource);
static uint64 WorkerShardSize(HTAB *workerShardStatistics, static uint64 WorkerShardSize(HTAB *workerShardStatistics,
char *workerName, int workerPort, uint64 shardId); char *workerName, int workerPort, uint64 shardId);
static XLogRecPtr WorkerShardLSN(HTAB *workerShardStatisticsHash, char *workerName,
int workerPort, uint64 shardId);
static XLogRecPtr WorkerLSN(HTAB *workerShardStatisticsHash,
char *workerName, int workerPort);
static void AddToWorkerShardIdSet(HTAB *shardsByWorker, char *workerName, int workerPort, static void AddToWorkerShardIdSet(HTAB *shardsByWorker, char *workerName, int workerPort,
uint64 shardId); uint64 shardId);
static HTAB * BuildShardSizesHash(ProgressMonitorData *monitor, HTAB *shardStatistics); static HTAB * BuildShardSizesHash(ProgressMonitorData *monitor, HTAB *shardStatistics);
@ -1232,6 +1240,11 @@ get_rebalance_progress(PG_FUNCTION_ARGS)
uint64 targetSize = WorkerShardSize(shardStatistics, step->targetName, uint64 targetSize = WorkerShardSize(shardStatistics, step->targetName,
step->targetPort, shardId); step->targetPort, shardId);
XLogRecPtr sourceLSN = WorkerLSN(shardStatistics, step->sourceName,
step->sourcePort);
XLogRecPtr targetLSN = WorkerShardLSN(shardStatistics, step->targetName,
step->targetPort, shardId);
uint64 shardSize = 0; uint64 shardSize = 0;
ShardStatistics *shardSizesStat = ShardStatistics *shardSizesStat =
hash_search(shardSizes, &shardId, HASH_FIND, NULL); hash_search(shardSizes, &shardId, HASH_FIND, NULL);
@ -1240,8 +1253,8 @@ get_rebalance_progress(PG_FUNCTION_ARGS)
shardSize = shardSizesStat->totalSize; shardSize = shardSizesStat->totalSize;
} }
Datum values[12]; Datum values[14];
bool nulls[12]; bool nulls[14];
memset(values, 0, sizeof(values)); memset(values, 0, sizeof(values));
memset(nulls, 0, sizeof(nulls)); memset(nulls, 0, sizeof(nulls));
@ -1259,6 +1272,8 @@ get_rebalance_progress(PG_FUNCTION_ARGS)
values[10] = UInt64GetDatum(targetSize); values[10] = UInt64GetDatum(targetSize);
values[11] = PointerGetDatum( values[11] = PointerGetDatum(
cstring_to_text(PlacementUpdateTypeNames[step->updateType])); cstring_to_text(PlacementUpdateTypeNames[step->updateType]));
values[12] = LSNGetDatum(sourceLSN);
values[13] = LSNGetDatum(targetLSN);
tuplestore_putvalues(tupstore, tupdesc, values, nulls); tuplestore_putvalues(tupstore, tupdesc, values, nulls);
} }
@ -1392,6 +1407,59 @@ WorkerShardSize(HTAB *workerShardStatisticsHash, char *workerName, int workerPor
} }
/*
* WorkerShardLSN returns the LSN of a shard on a worker, based on
* the workerShardStatisticsHash. If there is no LSN data in the
* statistics object, returns InvalidXLogRecPtr.
*/
static XLogRecPtr
WorkerShardLSN(HTAB *workerShardStatisticsHash, char *workerName, int workerPort,
uint64 shardId)
{
WorkerHashKey workerKey = { 0 };
strlcpy(workerKey.hostname, workerName, MAX_NODE_LENGTH);
workerKey.port = workerPort;
WorkerShardStatistics *workerStats =
hash_search(workerShardStatisticsHash, &workerKey, HASH_FIND, NULL);
if (!workerStats)
{
return InvalidXLogRecPtr;
}
ShardStatistics *shardStats =
hash_search(workerStats->statistics, &shardId, HASH_FIND, NULL);
if (!shardStats)
{
return InvalidXLogRecPtr;
}
return shardStats->shardLSN;
}
/*
* WorkerLSN returns the LSN of a worker, based on the workerShardStatisticsHash.
* If there is no LSN data in the statistics object, returns InvalidXLogRecPtr.
*/
static XLogRecPtr
WorkerLSN(HTAB *workerShardStatisticsHash, char *workerName, int workerPort)
{
WorkerHashKey workerKey = { 0 };
strlcpy(workerKey.hostname, workerName, MAX_NODE_LENGTH);
workerKey.port = workerPort;
WorkerShardStatistics *workerStats =
hash_search(workerShardStatisticsHash, &workerKey, HASH_FIND, NULL);
if (!workerStats)
{
return InvalidXLogRecPtr;
}
return workerStats->workerLSN;
}
/* /*
* BuildWorkerShardStatisticsHash returns a shard id -> shard statistics hash containing * BuildWorkerShardStatisticsHash returns a shard id -> shard statistics hash containing
* sizes of shards on the source node and destination node. * sizes of shards on the source node and destination node.
@ -1430,6 +1498,7 @@ BuildWorkerShardStatisticsHash(PlacementUpdateEventProgress *steps, int stepCoun
WorkerShardStatistics *moveStat = WorkerShardStatistics *moveStat =
hash_search(workerShardStatistics, &entry->worker, HASH_ENTER, NULL); hash_search(workerShardStatistics, &entry->worker, HASH_ENTER, NULL);
moveStat->statistics = statistics; moveStat->statistics = statistics;
moveStat->workerLSN = GetRemoteLogPosition(connection);
} }
return workerShardStatistics; return workerShardStatistics;
@ -1481,15 +1550,17 @@ GetShardStatistics(MultiConnection *connection, HTAB *shardIds)
appendStringInfoString(query, "))"); appendStringInfoString(query, "))");
appendStringInfoString( appendStringInfoString(
query, query,
" SELECT shard_id, coalesce(pg_total_relation_size(tables.relid),0)" " SELECT shard_id, coalesce(pg_total_relation_size(tables.relid),0), tables.lsn"
/* for each shard in shardIds */ /* for each shard in shardIds */
" FROM shard_names" " FROM shard_names"
/* check if its name can be found in pg_class, if so return size */ /* check if its name can be found in pg_class, if so return size */
" LEFT JOIN" " LEFT JOIN"
" (SELECT c.oid AS relid, c.relname, n.nspname" " (SELECT c.oid AS relid, c.relname, n.nspname, ss.latest_end_lsn AS lsn"
" FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace) tables" " FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
" LEFT JOIN pg_subscription_rel sr ON sr.srrelid = c.oid "
" LEFT JOIN pg_stat_subscription ss ON sr.srsubid = ss.subid) tables"
" ON tables.relname = shard_names.table_name AND" " ON tables.relname = shard_names.table_name AND"
" tables.nspname = shard_names.schema_name "); " tables.nspname = shard_names.schema_name ");
@ -1530,6 +1601,17 @@ GetShardStatistics(MultiConnection *connection, HTAB *shardIds)
ShardStatistics *statistics = ShardStatistics *statistics =
hash_search(shardStatistics, &shardId, HASH_ENTER, NULL); hash_search(shardStatistics, &shardId, HASH_ENTER, NULL);
statistics->totalSize = totalSize; statistics->totalSize = totalSize;
if (PQgetisnull(result, rowIndex, 2))
{
statistics->shardLSN = InvalidXLogRecPtr;
}
else
{
char *LSNString = PQgetvalue(result, rowIndex, 2);
Datum LSNDatum = DirectFunctionCall1(pg_lsn_in, CStringGetDatum(LSNString));
statistics->shardLSN = DatumGetLSN(LSNDatum);
}
} }
PQclear(result); PQclear(result);

4
src/backend/distributed/sql/citus--11.1-1--11.2-1.sql
View File

@ -1,3 +1 @@
-- citus--11.1-1--11.2-1 #include "udfs/get_rebalance_progress/11.2-1.sql"
-- bump version to 11.2-1

2
src/backend/distributed/sql/downgrades/citus--11.2-1--11.1-1.sql
View File

@ -1 +1 @@
-- citus--11.2-1--11.1-1 #include "../udfs/get_rebalance_progress/11.1-1.sql"

22
src/backend/distributed/sql/udfs/get_rebalance_progress/11.2-1.sql generated Normal file
View File

@ -0,0 +1,22 @@
DROP FUNCTION pg_catalog.get_rebalance_progress();
CREATE OR REPLACE FUNCTION pg_catalog.get_rebalance_progress()
RETURNS TABLE(sessionid integer,
table_name regclass,
shardid bigint,
shard_size bigint,
sourcename text,
sourceport int,
targetname text,
targetport int,
progress bigint,
source_shard_size bigint,
target_shard_size bigint,
operation_type text,
source_lsn pg_lsn,
target_lsn pg_lsn
)
AS 'MODULE_PATHNAME'
LANGUAGE C STRICT;
COMMENT ON FUNCTION pg_catalog.get_rebalance_progress()
IS 'provides progress information about the ongoing rebalance operations';

4
src/backend/distributed/sql/udfs/get_rebalance_progress/latest.sql
View File

@ -12,7 +12,9 @@ CREATE OR REPLACE FUNCTION pg_catalog.get_rebalance_progress()
progress bigint, progress bigint,
source_shard_size bigint, source_shard_size bigint,
target_shard_size bigint, target_shard_size bigint,
operation_type text operation_type text,
source_lsn pg_lsn,
target_lsn pg_lsn
) )
AS 'MODULE_PATHNAME' AS 'MODULE_PATHNAME'
LANGUAGE C STRICT; LANGUAGE C STRICT;

204
src/test/regress/expected/isolation_shard_rebalancer_progress.out
View File

@ -30,15 +30,18 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move |t |t |f
colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move |t |t |f
colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move |t |t |f
(4 rows) (4 rows)
step s2-unlock-1-start: step s2-unlock-1-start:
@ -75,10 +78,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -115,15 +121,18 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 0|localhost | 57638| 50000| 2|move colocated1|1500001| 50000|localhost | 57637| 0|localhost | 57638| 50000| 2|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 0|localhost | 57638| 400000| 2|move colocated2|1500005| 400000|localhost | 57637| 0|localhost | 57638| 400000| 2|move |t |t |f
colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 1|move colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 1|move |t |t |f
colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 1|move colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 1|move |t |t |f
(4 rows) (4 rows)
step s3-unlock-2-start: step s3-unlock-2-start:
@ -160,10 +169,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -212,15 +224,18 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move |t |t |f
colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move |t |t |f
colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move |t |t |f
(4 rows) (4 rows)
step s7-release-lock: step s7-release-lock:
@ -257,10 +272,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -301,15 +319,18 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|move |t |t |f
colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move colocated1|1500002| 200000|localhost | 57637| 200000|localhost | 57638| 0| 0|move |t |t |f
colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move colocated2|1500006| 8000|localhost | 57637| 8000|localhost | 57638| 0| 0|move |t |t |f
(4 rows) (4 rows)
step s6-release-advisory-lock: step s6-release-advisory-lock:
@ -351,10 +372,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -391,13 +415,16 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move |t |t |f
(2 rows) (2 rows)
step s2-unlock-1-start: step s2-unlock-1-start:
@ -426,10 +453,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -477,13 +507,16 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move |t |t |f
(2 rows) (2 rows)
step s7-release-lock: step s7-release-lock:
@ -512,10 +545,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -553,13 +589,16 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|copy colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|copy |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|copy colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|copy |t |t |f
(2 rows) (2 rows)
step s2-unlock-1-start: step s2-unlock-1-start:
@ -607,13 +646,16 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|move |t |t |f
(2 rows) (2 rows)
step s6-release-advisory-lock: step s6-release-advisory-lock:
@ -647,10 +689,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -691,13 +736,16 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|copy colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 8000| 1|copy |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|copy colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 8000| 1|copy |t |t |f
(2 rows) (2 rows)
step s6-release-advisory-lock: step s6-release-advisory-lock:
@ -751,14 +799,17 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 0| 1|move |t |t |f
separate |1500009| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move separate |1500009| 50000|localhost | 57637| 50000|localhost | 57638| 0| 1|move |t |t |f
(3 rows) (3 rows)
step s2-unlock-1-start: step s2-unlock-1-start:
@ -796,10 +847,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -851,14 +905,17 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move colocated1|1500001| 50000|localhost | 57637| 50000|localhost | 57638| 50000| 1|move |t |t |f
colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move colocated2|1500005| 400000|localhost | 57637| 400000|localhost | 57638| 400000| 1|move |t |t |f
separate |1500009| 50000|localhost | 57637| 50000|localhost | 57638| 200000| 1|move separate |1500009| 50000|localhost | 57637| 50000|localhost | 57638| 200000| 1|move |t |t |f
(3 rows) (3 rows)
step s7-release-lock: step s7-release-lock:
@ -896,10 +953,13 @@ step s7-get-progress:
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type table_name|shardid|shard_size|sourcename|sourceport|source_shard_size|targetname|targetport|target_shard_size|progress|operation_type|lsn_sanity_check|source_lsn_available|target_lsn_available
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)

6
src/test/regress/expected/multi_extension.out
View File

@ -1199,9 +1199,11 @@ SELECT * FROM multi_extension.print_extension_changes();
-- Snapshot of state at 11.2-1 -- Snapshot of state at 11.2-1
ALTER EXTENSION citus UPDATE TO '11.2-1'; ALTER EXTENSION citus UPDATE TO '11.2-1';
SELECT * FROM multi_extension.print_extension_changes(); SELECT * FROM multi_extension.print_extension_changes();
previous_object | current_object previous_object | current_object
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) function get_rebalance_progress() TABLE(sessionid integer, table_name regclass, shardid bigint, shard_size bigint, sourcename text, sourceport integer, targetname text, targetport integer, progress bigint, source_shard_size bigint, target_shard_size bigint, operation_type text) |
| function get_rebalance_progress() TABLE(sessionid integer, table_name regclass, shardid bigint, shard_size bigint, sourcename text, sourceport integer, targetname text, targetport integer, progress bigint, source_shard_size bigint, target_shard_size bigint, operation_type text, source_lsn pg_lsn, target_lsn pg_lsn)
(2 rows)
DROP TABLE multi_extension.prev_objects, multi_extension.extension_diff; DROP TABLE multi_extension.prev_objects, multi_extension.extension_diff;
-- show running version -- show running version

4
src/test/regress/expected/shard_rebalancer.out
View File

@ -1044,7 +1044,7 @@ SELECT * FROM get_rebalance_table_shards_plan('colocated_rebalance_test', rebala
-- Check that we can call this function -- Check that we can call this function
SELECT * FROM get_rebalance_progress(); SELECT * FROM get_rebalance_progress();
sessionid | table_name | shardid | shard_size | sourcename | sourceport | targetname | targetport | progress | source_shard_size | target_shard_size | operation_type sessionid | table_name | shardid | shard_size | sourcename | sourceport | targetname | targetport | progress | source_shard_size | target_shard_size | operation_type | source_lsn | target_lsn
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)
@ -1058,7 +1058,7 @@ SELECT * FROM rebalance_table_shards('colocated_rebalance_test', threshold := 0,
CALL citus_cleanup_orphaned_shards(); CALL citus_cleanup_orphaned_shards();
-- Check that we can call this function without a crash -- Check that we can call this function without a crash
SELECT * FROM get_rebalance_progress(); SELECT * FROM get_rebalance_progress();
sessionid | table_name | shardid | shard_size | sourcename | sourceport | targetname | targetport | progress | source_shard_size | target_shard_size | operation_type sessionid | table_name | shardid | shard_size | sourcename | sourceport | targetname | targetport | progress | source_shard_size | target_shard_size | operation_type | source_lsn | target_lsn
--------------------------------------------------------------------- ---------------------------------------------------------------------
(0 rows) (0 rows)

5
src/test/regress/spec/isolation_shard_rebalancer_progress.spec
View File

@ -172,7 +172,10 @@ step "s7-get-progress"
targetport, targetport,
( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size, ( SELECT size FROM possible_sizes WHERE ABS(size - target_shard_size) = (SELECT MIN(ABS(size - target_shard_size)) FROM possible_sizes )) target_shard_size,
progress, progress,
operation_type operation_type,
source_lsn >= target_lsn as lsn_sanity_check,
source_lsn > '0/0' as source_lsn_available,
target_lsn > '0/0' as target_lsn_available
FROM get_rebalance_progress(); FROM get_rebalance_progress();
} }