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

Calling ErrorOnConcurrentOperation correctly

pull/7013/head
Shabnam Khan 2023-06-19 16:27:32 +05:30
parent 18ef6862ba
commit 27389e8f58
1 changed files with 16 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/backend/distributed/operations/auto_shard_split.c
View File

@ -1,21 +1,21 @@
#include "postgres.h" #include "postgres.h"
#include "libpq-fe.h" #include "libpq-fe.h"
#include "executor/spi.h" #include "executor/spi.h"
#include "distributed/lock_graph.h" // #include "distributed/lock_graph.h"
#include "distributed/coordinator_protocol.h" #include "distributed/coordinator_protocol.h"
#include "distributed/metadata_cache.h" // #include "distributed/metadata_cache.h"
#include "distributed/metadata_utility.h" // #include "distributed/metadata_utility.h"
#include "distributed/multi_logical_replication.h" // #include "distributed/multi_logical_replication.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"
#include "distributed/pg_dist_shard.h" // #include "distributed/pg_dist_shard.h"
#include "distributed/reference_table_utils.h" // #include "distributed/reference_table_utils.h"
#include "distributed/remote_commands.h" // #include "distributed/remote_commands.h"
#include "distributed/resource_lock.h" // #include "distributed/resource_lock.h"
#include "distributed/tuplestore.h" // #include "distributed/tuplestore.h"
#include "distributed/utils/array_type.h" // #include "distributed/utils/array_type.h"
#include "distributed/worker_protocol.h" // #include "distributed/worker_protocol.h"
#include "nodes/pg_list.h" // #include "nodes/pg_list.h"
#include "postmaster/postmaster.h" #include "postmaster/postmaster.h"
#include "distributed/distribution_column.h" #include "distributed/distribution_column.h"
#include "utils/builtins.h" #include "utils/builtins.h"
@ -110,7 +110,6 @@ ExecuteSplitBackgroundJob(int64 jobid, ShardInfo shardinfo, List *SplitPoints)
int32 nodesInvolved[1]; int32 nodesInvolved[1];
nodesInvolved[0] = shardinfo->nodeid; nodesInvolved[0] = shardinfo->nodeid;
Oid superUserId = CitusExtensionOwner(); Oid superUserId = CitusExtensionOwner();
ErrorOnConcurrentOperation();
BackgroundTask *task = ScheduleBackgroundTask(jobid, superUserId, splitQuery->data, 0, BackgroundTask *task = ScheduleBackgroundTask(jobid, superUserId, splitQuery->data, 0,
NULL, 1, nodesInvolved); NULL, 1, nodesInvolved);
} }
@ -255,6 +254,7 @@ ScheduleShardSplit(ShardInfo shardinfo)
List *SplitPoints = FindShardSplitPoints(shardinfo); List *SplitPoints = FindShardSplitPoints(shardinfo);
if (list_length(SplitPoints) > 0) if (list_length(SplitPoints) > 0)
{ {
ErrorOnConcurrentOperation();
int64 jobId = CreateBackgroundJob("Automatic Shard Split", int64 jobId = CreateBackgroundJob("Automatic Shard Split",
"Split using SplitPoints List"); "Split using SplitPoints List");
ExecuteSplitBackgroundJob(jobId, shardinfo, SplitPoints); ExecuteSplitBackgroundJob(jobId, shardinfo, SplitPoints);
@ -290,7 +290,7 @@ citus_auto_shard_split_start(PG_FUNCTION_ARGS)
" FROM citus_shards cs JOIN pg_dist_shard ps USING(shardid)" " FROM citus_shards cs JOIN pg_dist_shard ps USING(shardid)"
" WINDOW w AS (PARTITION BY colocation_id, shardminvalue ORDER BY shard_size DESC) )as t where total_sum >= %ld )" " WINDOW w AS (PARTITION BY colocation_id, shardminvalue ORDER BY shard_size DESC) )as t where total_sum >= %ld )"
" AS max_sizes ON cs.shardid=max_sizes.shardid AND cs.shard_size = max_sizes.max_size JOIN citus_tables ct ON cs.table_name = ct.table_name AND pd.shardminvalue <> pd.shardmaxvalue AND pd.shardminvalue <> ''", " AS max_sizes ON cs.shardid=max_sizes.shardid AND cs.shard_size = max_sizes.max_size JOIN citus_tables ct ON cs.table_name = ct.table_name AND pd.shardminvalue <> pd.shardmaxvalue AND pd.shardminvalue <> ''",
MaxShardSize 0
); );
ereport(LOG, errmsg("%s", query->data)); ereport(LOG, errmsg("%s", query->data));