Rename use -> shouldUse

Because setting the flag doesn't necessarily mean that we'll
use 2PC. If connections are read-only, we will not use 2PC.
In other words, we'll use 2PC only for connections that modified
any placements.
pull/4806/head
Onder Kalaci 2021-03-10 19:52:05 +01:00
parent 6a7ed7b309
commit e65e72130d
12 changed files with 96 additions and 86 deletions

View File

@ -2244,7 +2244,7 @@ CitusCopyDestReceiverStartup(DestReceiver *dest, int operation,
if (cacheEntry->replicationModel == REPLICATION_MODEL_2PC || if (cacheEntry->replicationModel == REPLICATION_MODEL_2PC ||
MultiShardCommitProtocol == COMMIT_PROTOCOL_2PC) MultiShardCommitProtocol == COMMIT_PROTOCOL_2PC)
{ {
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
} }
/* define how tuples will be serialised */ /* define how tuples will be serialised */

View File

@ -1223,7 +1223,7 @@ StartDistributedExecution(DistributedExecution *execution)
if (xactProperties->requires2PC) if (xactProperties->requires2PC)
{ {
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
} }
/* /*
@ -3180,7 +3180,7 @@ Activate2PCIfModifyingTransactionExpandsToNewNode(WorkerSession *session)
* just opened, which means we're now going to make modifications * just opened, which means we're now going to make modifications
* over multiple connections. Activate 2PC! * over multiple connections. Activate 2PC!
*/ */
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
} }
} }

View File

@ -244,7 +244,7 @@ ExecuteLocalTaskListExtended(List *taskList,
* queries are also ReadOnly, our 2PC logic is smart enough to skip sending * queries are also ReadOnly, our 2PC logic is smart enough to skip sending
* PREPARE to those connections. * PREPARE to those connections.
*/ */
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
} }
LogLocalCommand(task); LogLocalCommand(task);

View File

@ -332,7 +332,7 @@ DropShards(Oid relationId, char *schemaName, char *relationName,
*/ */
if (MultiShardCommitProtocol == COMMIT_PROTOCOL_2PC) if (MultiShardCommitProtocol == COMMIT_PROTOCOL_2PC)
{ {
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
} }
List *dropTaskList = DropTaskList(relationId, schemaName, relationName, List *dropTaskList = DropTaskList(relationId, schemaName, relationName,

View File

@ -24,7 +24,7 @@
static Size MemoryContextTotalSpace(MemoryContext context); static Size MemoryContextTotalSpace(MemoryContext context);
PG_FUNCTION_INFO_V1(top_transaction_context_size); PG_FUNCTION_INFO_V1(top_transaction_context_size);
PG_FUNCTION_INFO_V1(coordinated_transaction_uses_2PC); PG_FUNCTION_INFO_V1(coordinated_transaction_should_use_2PC);
/* /*
@ -62,17 +62,17 @@ MemoryContextTotalSpace(MemoryContext context)
/* /*
* coordinated_transaction_uses_2PC returns true if the transaction is in a * coordinated_transaction_should_use_2PC returns true if the transaction is in a
* coordinated transaction and uses 2PC. If the transaction is nott in a * coordinated transaction and uses 2PC. If the transaction is nott in a
* coordinated transaction, the function throws an error. * coordinated transaction, the function throws an error.
*/ */
Datum Datum
coordinated_transaction_uses_2PC(PG_FUNCTION_ARGS) coordinated_transaction_should_use_2PC(PG_FUNCTION_ARGS)
{ {
if (!InCoordinatedTransaction()) if (!InCoordinatedTransaction())
{ {
ereport(ERROR, (errmsg("The transaction is not a coordinated transaction"))); ereport(ERROR, (errmsg("The transaction is not a coordinated transaction")));
} }
PG_RETURN_BOOL(GetCoordinatedTransactionUses2PC()); PG_RETURN_BOOL(GetCoordinatedTransactionShouldUse2PC());
} }

View File

@ -96,9 +96,16 @@ MemoryContext CommitContext = NULL;
/* /*
* Should this coordinated transaction use 2PC? Set by * Should this coordinated transaction use 2PC? Set by
* CoordinatedTransactionUse2PC(), e.g. if DDL was issued and * CoordinatedTransactionUse2PC(), e.g. if DDL was issued and
* MultiShardCommitProtocol was set to 2PC. * MultiShardCommitProtocol was set to 2PC. But, even if this
* flag is set, the transaction manager is smart enough to only
* do 2PC on the remote connections that did a modification.
*
* As a variable name ShouldCoordinatedTransactionUse2PC could
* be improved. We use CoordinatedTransactionShouldUse2PC() as the
* public API function, hence couldn't come up with a better name
* for the underlying variable at the moment.
*/ */
bool CoordinatedTransactionUses2PC = false; bool ShouldCoordinatedTransactionUse2PC = false;
/* if disabled, distributed statements in a function may run as separate transactions */ /* if disabled, distributed statements in a function may run as separate transactions */
bool FunctionOpensTransactionBlock = true; bool FunctionOpensTransactionBlock = true;
@ -183,26 +190,29 @@ InCoordinatedTransaction(void)
/* /*
* CoordinatedTransactionUse2PC() signals that the current coordinated * CoordinatedTransactionShouldUse2PC() signals that the current coordinated
* transaction should use 2PC to commit. * transaction should use 2PC to commit.
*
* Note that even if 2PC is enabled, it is only used for connections that make
* modification (DML or DDL).
*/ */
void void
CoordinatedTransactionUse2PC(void) CoordinatedTransactionShouldUse2PC(void)
{ {
Assert(InCoordinatedTransaction()); Assert(InCoordinatedTransaction());
CoordinatedTransactionUses2PC = true; ShouldCoordinatedTransactionUse2PC = true;
} }
/* /*
* GetCoordinatedTransactionUses2PC is a wrapper function to read the value * GetCoordinatedTransactionShouldUse2PC is a wrapper function to read the value
* of CoordinatedTransactionUses2PC. * of CoordinatedTransactionShouldUse2PCFlag.
*/ */
bool bool
GetCoordinatedTransactionUses2PC(void) GetCoordinatedTransactionShouldUse2PC(void)
{ {
return CoordinatedTransactionUses2PC; return ShouldCoordinatedTransactionUse2PC;
} }
@ -436,7 +446,7 @@ CoordinatedTransactionCallback(XactEvent event, void *arg)
*/ */
MarkFailedShardPlacements(); MarkFailedShardPlacements();
if (CoordinatedTransactionUses2PC) if (ShouldCoordinatedTransactionUse2PC)
{ {
CoordinatedRemoteTransactionsPrepare(); CoordinatedRemoteTransactionsPrepare();
CurrentCoordinatedTransactionState = COORD_TRANS_PREPARED; CurrentCoordinatedTransactionState = COORD_TRANS_PREPARED;
@ -464,7 +474,7 @@ CoordinatedTransactionCallback(XactEvent event, void *arg)
* Check again whether shards/placement successfully * Check again whether shards/placement successfully
* committed. This handles failure at COMMIT/PREPARE time. * committed. This handles failure at COMMIT/PREPARE time.
*/ */
PostCommitMarkFailedShardPlacements(CoordinatedTransactionUses2PC); PostCommitMarkFailedShardPlacements(ShouldCoordinatedTransactionUse2PC);
break; break;
} }
@ -496,7 +506,7 @@ ResetGlobalVariables()
FreeSavedExplainPlan(); FreeSavedExplainPlan();
dlist_init(&InProgressTransactions); dlist_init(&InProgressTransactions);
activeSetStmts = NULL; activeSetStmts = NULL;
CoordinatedTransactionUses2PC = false; ShouldCoordinatedTransactionUse2PC = false;
TransactionModifiedNodeMetadata = false; TransactionModifiedNodeMetadata = false;
MetadataSyncOnCommit = false; MetadataSyncOnCommit = false;
ResetWorkerErrorIndication(); ResetWorkerErrorIndication();

View File

@ -96,7 +96,7 @@ SendCommandToWorkerAsUser(const char *nodeName, int32 nodePort, const char *node
uint32 connectionFlags = 0; uint32 connectionFlags = 0;
UseCoordinatedTransaction(); UseCoordinatedTransaction();
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
MultiConnection *transactionConnection = GetNodeUserDatabaseConnection( MultiConnection *transactionConnection = GetNodeUserDatabaseConnection(
connectionFlags, nodeName, connectionFlags, nodeName,
@ -404,7 +404,7 @@ SendCommandToWorkersParamsInternal(TargetWorkerSet targetWorkerSet, const char *
List *workerNodeList = TargetWorkerSetNodeList(targetWorkerSet, ShareLock); List *workerNodeList = TargetWorkerSetNodeList(targetWorkerSet, ShareLock);
UseCoordinatedTransaction(); UseCoordinatedTransaction();
CoordinatedTransactionUse2PC(); CoordinatedTransactionShouldUse2PC();
/* open connections in parallel */ /* open connections in parallel */
WorkerNode *workerNode = NULL; WorkerNode *workerNode = NULL;

View File

@ -111,8 +111,8 @@ extern bool TransactionModifiedNodeMetadata;
*/ */
extern void UseCoordinatedTransaction(void); extern void UseCoordinatedTransaction(void);
extern bool InCoordinatedTransaction(void); extern bool InCoordinatedTransaction(void);
extern void CoordinatedTransactionUse2PC(void); extern void CoordinatedTransactionShouldUse2PC(void);
extern bool GetCoordinatedTransactionUses2PC(void); extern bool GetCoordinatedTransactionShouldUse2PC(void);
extern bool IsMultiStatementTransaction(void); extern bool IsMultiStatementTransaction(void);
extern void EnsureDistributedTransactionId(void); extern void EnsureDistributedTransactionId(void);

View File

@ -895,11 +895,11 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM (SELECT in
0 0
(1 row) (1 row)
-- a helper function which return true if the coordinated trannsaction uses -- a helper function which return true if the coordinated
-- 2PC -- trannsaction uses 2PC
CREATE OR REPLACE FUNCTION coordinated_transaction_uses_2PC() CREATE OR REPLACE FUNCTION coordinated_transaction_should_use_2PC()
RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus', RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus',
$$coordinated_transaction_uses_2PC$$; $$coordinated_transaction_should_use_2PC$$;
-- a local SELECT followed by remote SELECTs -- a local SELECT followed by remote SELECTs
-- does not trigger 2PC -- does not trigger 2PC
BEGIN; BEGIN;
@ -933,8 +933,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM coordinato
0 0
(1 row) (1 row)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
f f
(1 row) (1 row)
@ -973,8 +973,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM coordinato
0 0
(1 row) (1 row)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
f f
(1 row) (1 row)
@ -992,8 +992,8 @@ NOTICE: executing the command locally: SELECT y FROM coordinator_shouldhaveshar
UPDATE test SET y = y +1; UPDATE test SET y = y +1;
NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503000 test SET y = (y OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503000 test SET y = (y OPERATOR(pg_catalog.+) 1)
NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503003 test SET y = (y OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503003 test SET y = (y OPERATOR(pg_catalog.+) 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1012,8 +1012,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM coordinato
1 1
(1 row) (1 row)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1027,8 +1027,8 @@ NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards
UPDATE test SET y = y +1; UPDATE test SET y = y +1;
NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503000 test SET y = (y OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503000 test SET y = (y OPERATOR(pg_catalog.+) 1)
NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503003 test SET y = (y OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE coordinator_shouldhaveshards.test_1503003 test SET y = (y OPERATOR(pg_catalog.+) 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1040,8 +1040,8 @@ BEGIN;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1) NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1)
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1053,8 +1053,8 @@ BEGIN;
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1) NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1073,8 +1073,8 @@ BEGIN;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1) NOTICE: executing the command locally: INSERT INTO coordinator_shouldhaveshards.test_1503000 (x, y) VALUES (1, 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1094,8 +1094,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM coordinato
(1 row) (1 row)
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
f f
(1 row) (1 row)

View File

@ -1856,9 +1856,9 @@ NOTICE: executing the command locally: SELECT bool_and((z IS NULL)) AS bool_and
RESET citus.local_copy_flush_threshold; RESET citus.local_copy_flush_threshold;
RESET citus.local_copy_flush_threshold; RESET citus.local_copy_flush_threshold;
CREATE OR REPLACE FUNCTION coordinated_transaction_uses_2PC() CREATE OR REPLACE FUNCTION coordinated_transaction_should_use_2PC()
RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus', RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus',
$$coordinated_transaction_uses_2PC$$; $$coordinated_transaction_should_use_2PC$$;
-- a multi-shard/single-shard select that is failed over to local -- a multi-shard/single-shard select that is failed over to local
-- execution doesn't start a 2PC -- execution doesn't start a 2PC
BEGIN; BEGIN;
@ -1899,8 +1899,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM (SELECT an
1 1
(1 row) (1 row)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
f f
(1 row) (1 row)
@ -1908,7 +1908,7 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM (SELECT an
ROLLBACK; ROLLBACK;
-- same without a transaction block -- same without a transaction block
WITH cte_1 AS (SELECT count(*) as cnt FROM another_schema_table LIMIT 1000), WITH cte_1 AS (SELECT count(*) as cnt FROM another_schema_table LIMIT 1000),
cte_2 AS (SELECT coordinated_transaction_uses_2PC() as enabled_2pc) cte_2 AS (SELECT coordinated_transaction_should_use_2PC() as enabled_2pc)
SELECT cnt, enabled_2pc FROM cte_1, cte_2; SELECT cnt, enabled_2pc FROM cte_1, cte_2;
NOTICE: executing the command locally: SELECT count(*) AS cnt FROM single_node.another_schema_table_90630511 another_schema_table WHERE true LIMIT '1000'::bigint NOTICE: executing the command locally: SELECT count(*) AS cnt FROM single_node.another_schema_table_90630511 another_schema_table WHERE true LIMIT '1000'::bigint
NOTICE: executing the command locally: SELECT count(*) AS cnt FROM single_node.another_schema_table_90630512 another_schema_table WHERE true LIMIT '1000'::bigint NOTICE: executing the command locally: SELECT count(*) AS cnt FROM single_node.another_schema_table_90630512 another_schema_table WHERE true LIMIT '1000'::bigint
@ -1928,8 +1928,8 @@ NOTICE: executing the command locally: UPDATE single_node.another_schema_table_
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630512 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630512 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1)
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630513 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630513 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1)
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630514 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630514 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1950,8 +1950,8 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM (SELECT in
10001 10001
(1 row) (1 row)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1959,13 +1959,13 @@ NOTICE: executing the command locally: SELECT count(*) AS count FROM (SELECT in
ROLLBACK; ROLLBACK;
-- same without transaction block -- same without transaction block
WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *) WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630512 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630512 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630513 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630513 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630514 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630514 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) RETURNING a, b
NOTICE: executing the command locally: SELECT single_node.coordinated_transaction_uses_2pc() AS coordinated_transaction_uses_2pc NOTICE: executing the command locally: SELECT single_node.coordinated_transaction_should_use_2pc() AS coordinated_transaction_should_use_2pc
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1975,8 +1975,8 @@ NOTICE: executing the command locally: SELECT single_node.coordinated_transacti
BEGIN; BEGIN;
UPDATE another_schema_table SET b = b + 1 WHERE a = 1; UPDATE another_schema_table SET b = b + 1 WHERE a = 1;
NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) WHERE (a OPERATOR(pg_catalog.=) 1) NOTICE: executing the command locally: UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (b OPERATOR(pg_catalog.+) 1) WHERE (a OPERATOR(pg_catalog.=) 1)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
coordinated_transaction_uses_2pc coordinated_transaction_should_use_2pc
--------------------------------------------------------------------- ---------------------------------------------------------------------
t t
(1 row) (1 row)
@ -1984,8 +1984,8 @@ NOTICE: executing the command locally: UPDATE single_node.another_schema_table_
ROLLBACK; ROLLBACK;
-- same without transaction block -- same without transaction block
WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 WHERE a = 1 RETURNING *) WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 WHERE a = 1 RETURNING *)
SELECT coordinated_transaction_uses_2PC() FROM cte_1; SELECT coordinated_transaction_should_use_2PC() FROM cte_1;
NOTICE: executing the command locally: WITH cte_1 AS (UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (another_schema_table.b OPERATOR(pg_catalog.+) 1) WHERE (another_schema_table.a OPERATOR(pg_catalog.=) 1) RETURNING another_schema_table.a, another_schema_table.b) SELECT single_node.coordinated_transaction_uses_2pc() AS coordinated_transaction_uses_2pc FROM cte_1 NOTICE: executing the command locally: WITH cte_1 AS (UPDATE single_node.another_schema_table_90630511 another_schema_table SET b = (another_schema_table.b OPERATOR(pg_catalog.+) 1) WHERE (another_schema_table.a OPERATOR(pg_catalog.=) 1) RETURNING another_schema_table.a, another_schema_table.b) SELECT single_node.coordinated_transaction_should_use_2pc() AS coordinated_transaction_should_use_2pc FROM cte_1
ERROR: The transaction is not a coordinated transaction ERROR: The transaction is not a coordinated transaction
-- if the local execution is disabled, we cannot failover to -- if the local execution is disabled, we cannot failover to
-- local execution and the queries would fail -- local execution and the queries would fail

View File

@ -372,11 +372,11 @@ inserts AS (
RETURNING * RETURNING *
) SELECT count(*) FROM inserts; ) SELECT count(*) FROM inserts;
-- a helper function which return true if the coordinated trannsaction uses -- a helper function which return true if the coordinated
-- 2PC -- trannsaction uses 2PC
CREATE OR REPLACE FUNCTION coordinated_transaction_uses_2PC() CREATE OR REPLACE FUNCTION coordinated_transaction_should_use_2PC()
RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus', RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus',
$$coordinated_transaction_uses_2PC$$; $$coordinated_transaction_should_use_2PC$$;
-- a local SELECT followed by remote SELECTs -- a local SELECT followed by remote SELECTs
-- does not trigger 2PC -- does not trigger 2PC
@ -385,7 +385,7 @@ BEGIN;
WITH cte_1 AS (SELECT y FROM test WHERE x = 1 LIMIT 5) SELECT count(*) FROM test; WITH cte_1 AS (SELECT y FROM test WHERE x = 1 LIMIT 5) SELECT count(*) FROM test;
SELECT count(*) FROM test; SELECT count(*) FROM test;
WITH cte_1 as (SELECT * FROM test LIMIT 5) SELECT count(*) FROM test; WITH cte_1 as (SELECT * FROM test LIMIT 5) SELECT count(*) FROM test;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- remote SELECTs followed by local SELECTs -- remote SELECTs followed by local SELECTs
@ -395,7 +395,7 @@ BEGIN;
WITH cte_1 as (SELECT * FROM test LIMIT 5) SELECT count(*) FROM test; WITH cte_1 as (SELECT * FROM test LIMIT 5) SELECT count(*) FROM test;
SELECT y FROM test WHERE x = 1; SELECT y FROM test WHERE x = 1;
WITH cte_1 AS (SELECT y FROM test WHERE x = 1 LIMIT 5) SELECT count(*) FROM test; WITH cte_1 AS (SELECT y FROM test WHERE x = 1 LIMIT 5) SELECT count(*) FROM test;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a local SELECT followed by a remote Modify -- a local SELECT followed by a remote Modify
@ -403,7 +403,7 @@ COMMIT;
BEGIN; BEGIN;
SELECT y FROM test WHERE x = 1; SELECT y FROM test WHERE x = 1;
UPDATE test SET y = y +1; UPDATE test SET y = y +1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a local modify followed by a remote SELECT -- a local modify followed by a remote SELECT
@ -411,7 +411,7 @@ COMMIT;
BEGIN; BEGIN;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
SELECT count(*) FROM test; SELECT count(*) FROM test;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a local modify followed by a remote MODIFY -- a local modify followed by a remote MODIFY
@ -419,7 +419,7 @@ COMMIT;
BEGIN; BEGIN;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
UPDATE test SET y = y +1; UPDATE test SET y = y +1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a local modify followed by a remote single shard MODIFY -- a local modify followed by a remote single shard MODIFY
@ -427,7 +427,7 @@ COMMIT;
BEGIN; BEGIN;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a remote single shard modify followed by a local single -- a remote single shard modify followed by a local single
@ -435,7 +435,7 @@ COMMIT;
BEGIN; BEGIN;
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
COMMIT; COMMIT;
-- a remote single shard select followed by a local single -- a remote single shard select followed by a local single
@ -445,7 +445,7 @@ COMMIT;
BEGIN; BEGIN;
SELECT count(*) FROM test WHERE x = 3; SELECT count(*) FROM test WHERE x = 3;
INSERT INTO test VALUES (1,1); INSERT INTO test VALUES (1,1);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
SET LOCAL citus.log_remote_commands TO ON; SET LOCAL citus.log_remote_commands TO ON;
COMMIT; COMMIT;
@ -454,7 +454,7 @@ COMMIT;
BEGIN; BEGIN;
SELECT count(*) FROM test WHERE x = 1; SELECT count(*) FROM test WHERE x = 1;
INSERT INTO test VALUES (3,3); INSERT INTO test VALUES (3,3);
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
SET LOCAL citus.log_remote_commands TO ON; SET LOCAL citus.log_remote_commands TO ON;
COMMIT; COMMIT;

View File

@ -909,9 +909,9 @@ RESET citus.local_copy_flush_threshold;
RESET citus.local_copy_flush_threshold; RESET citus.local_copy_flush_threshold;
CREATE OR REPLACE FUNCTION coordinated_transaction_uses_2PC() CREATE OR REPLACE FUNCTION coordinated_transaction_should_use_2PC()
RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus', RETURNS BOOL LANGUAGE C STRICT VOLATILE AS 'citus',
$$coordinated_transaction_uses_2PC$$; $$coordinated_transaction_should_use_2PC$$;
-- a multi-shard/single-shard select that is failed over to local -- a multi-shard/single-shard select that is failed over to local
-- execution doesn't start a 2PC -- execution doesn't start a 2PC
@ -922,19 +922,19 @@ BEGIN;
SELECT count(*) FROM cte_1; SELECT count(*) FROM cte_1;
WITH cte_1 as (SELECT * FROM another_schema_table WHERE a = 1 LIMIT 10) WITH cte_1 as (SELECT * FROM another_schema_table WHERE a = 1 LIMIT 10)
SELECT count(*) FROM cte_1; SELECT count(*) FROM cte_1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
ROLLBACK; ROLLBACK;
-- same without a transaction block -- same without a transaction block
WITH cte_1 AS (SELECT count(*) as cnt FROM another_schema_table LIMIT 1000), WITH cte_1 AS (SELECT count(*) as cnt FROM another_schema_table LIMIT 1000),
cte_2 AS (SELECT coordinated_transaction_uses_2PC() as enabled_2pc) cte_2 AS (SELECT coordinated_transaction_should_use_2PC() as enabled_2pc)
SELECT cnt, enabled_2pc FROM cte_1, cte_2; SELECT cnt, enabled_2pc FROM cte_1, cte_2;
-- a multi-shard modification that is failed over to local -- a multi-shard modification that is failed over to local
-- execution starts a 2PC -- execution starts a 2PC
BEGIN; BEGIN;
UPDATE another_schema_table SET b = b + 1; UPDATE another_schema_table SET b = b + 1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
ROLLBACK; ROLLBACK;
-- a multi-shard modification that is failed over to local -- a multi-shard modification that is failed over to local
@ -942,23 +942,23 @@ ROLLBACK;
BEGIN; BEGIN;
WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *) WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *)
SELECT count(*) FROM cte_1; SELECT count(*) FROM cte_1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
ROLLBACK; ROLLBACK;
-- same without transaction block -- same without transaction block
WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *) WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 RETURNING *)
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
-- a single-shard modification that is failed over to local -- a single-shard modification that is failed over to local
-- starts 2PC execution -- starts 2PC execution
BEGIN; BEGIN;
UPDATE another_schema_table SET b = b + 1 WHERE a = 1; UPDATE another_schema_table SET b = b + 1 WHERE a = 1;
SELECT coordinated_transaction_uses_2PC(); SELECT coordinated_transaction_should_use_2PC();
ROLLBACK; ROLLBACK;
-- same without transaction block -- same without transaction block
WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 WHERE a = 1 RETURNING *) WITH cte_1 AS (UPDATE another_schema_table SET b = b + 1 WHERE a = 1 RETURNING *)
SELECT coordinated_transaction_uses_2PC() FROM cte_1; SELECT coordinated_transaction_should_use_2PC() FROM cte_1;
-- if the local execution is disabled, we cannot failover to -- if the local execution is disabled, we cannot failover to
-- local execution and the queries would fail -- local execution and the queries would fail