| blob | 6a6d152fbfb45a281df54d37b891dc388098a27f |
1 /*-------------------------------------------------------------------------
2 * worker.c
3 * PostgreSQL logical replication worker (apply)
4 *
5 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
6 *
7 * IDENTIFICATION
8 * src/backend/replication/logical/worker.c
9 *
10 * NOTES
11 * This file contains the worker which applies logical changes as they come
12 * from remote logical replication stream.
13 *
14 * The main worker (apply) is started by logical replication worker
15 * launcher for every enabled subscription in a database. It uses
16 * walsender protocol to communicate with publisher.
17 *
18 * This module includes server facing code and shares libpqwalreceiver
19 * module with walreceiver for providing the libpq specific functionality.
20 *
21 *
22 * STREAMED TRANSACTIONS
23 * ---------------------
24 * Streamed transactions (large transactions exceeding a memory limit on the
25 * upstream) are not applied immediately, but instead, the data is written
26 * to temporary files and then applied at once when the final commit arrives.
27 *
28 * Unlike the regular (non-streamed) case, handling streamed transactions has
29 * to handle aborts of both the toplevel transaction and subtransactions. This
30 * is achieved by tracking offsets for subtransactions, which is then used
31 * to truncate the file with serialized changes.
32 *
33 * The files are placed in tmp file directory by default, and the filenames
34 * include both the XID of the toplevel transaction and OID of the
35 * subscription. This is necessary so that different workers processing a
36 * remote transaction with the same XID doesn't interfere.
37 *
38 * We use BufFiles instead of using normal temporary files because (a) the
39 * BufFile infrastructure supports temporary files that exceed the OS file size
40 * limit, (b) provides a way for automatic clean up on the error and (c) provides
41 * a way to survive these files across local transactions and allow to open and
42 * close at stream start and close. We decided to use FileSet
43 * infrastructure as without that it deletes the files on the closure of the
44 * file and if we decide to keep stream files open across the start/stop stream
45 * then it will consume a lot of memory (more than 8K for each BufFile and
46 * there could be multiple such BufFiles as the subscriber could receive
47 * multiple start/stop streams for different transactions before getting the
48 * commit). Moreover, if we don't use FileSet then we also need to invent
49 * a new way to pass filenames to BufFile APIs so that we are allowed to open
50 * the file we desired across multiple stream-open calls for the same
51 * transaction.
52 *
53 * TWO_PHASE TRANSACTIONS
54 * ----------------------
55 * Two phase transactions are replayed at prepare and then committed or
56 * rolled back at commit prepared and rollback prepared respectively. It is
57 * possible to have a prepared transaction that arrives at the apply worker
58 * when the tablesync is busy doing the initial copy. In this case, the apply
59 * worker skips all the prepared operations [e.g. inserts] while the tablesync
60 * is still busy (see the condition of should_apply_changes_for_rel). The
61 * tablesync worker might not get such a prepared transaction because say it
62 * was prior to the initial consistent point but might have got some later
63 * commits. Now, the tablesync worker will exit without doing anything for the
64 * prepared transaction skipped by the apply worker as the sync location for it
65 * will be already ahead of the apply worker's current location. This would lead
66 * to an "empty prepare", because later when the apply worker does the commit
67 * prepare, there is nothing in it (the inserts were skipped earlier).
68 *
69 * To avoid this, and similar prepare confusions the subscription's two_phase
70 * commit is enabled only after the initial sync is over. The two_phase option
71 * has been implemented as a tri-state with values DISABLED, PENDING, and
72 * ENABLED.
73 *
74 * Even if the user specifies they want a subscription with two_phase = on,
75 * internally it will start with a tri-state of PENDING which only becomes
76 * ENABLED after all tablesync initializations are completed - i.e. when all
77 * tablesync workers have reached their READY state. In other words, the value
78 * PENDING is only a temporary state for subscription start-up.
79 *
80 * Until the two_phase is properly available (ENABLED) the subscription will
81 * behave as if two_phase = off. When the apply worker detects that all
82 * tablesyncs have become READY (while the tri-state was PENDING) it will
83 * restart the apply worker process. This happens in
84 * process_syncing_tables_for_apply.
85 *
86 * When the (re-started) apply worker finds that all tablesyncs are READY for a
87 * two_phase tri-state of PENDING it start streaming messages with the
88 * two_phase option which in turn enables the decoding of two-phase commits at
89 * the publisher. Then, it updates the tri-state value from PENDING to ENABLED.
90 * Now, it is possible that during the time we have not enabled two_phase, the
91 * publisher (replication server) would have skipped some prepares but we
92 * ensure that such prepares are sent along with commit prepare, see
93 * ReorderBufferFinishPrepared.
94 *
95 * If the subscription has no tables then a two_phase tri-state PENDING is
96 * left unchanged. This lets the user still do an ALTER TABLE REFRESH
97 * PUBLICATION which might otherwise be disallowed (see below).
98 *
99 * If ever a user needs to be aware of the tri-state value, they can fetch it
100 * from the pg_subscription catalog (see column subtwophasestate).
101 *
102 * We don't allow to toggle two_phase option of a subscription because it can
103 * lead to an inconsistent replica. Consider, initially, it was on and we have
104 * received some prepare then we turn it off, now at commit time the server
105 * will send the entire transaction data along with the commit. With some more
106 * analysis, we can allow changing this option from off to on but not sure if
107 * that alone would be useful.
108 *
109 * Finally, to avoid problems mentioned in previous paragraphs from any
110 * subsequent (not READY) tablesyncs (need to toggle two_phase option from 'on'
111 * to 'off' and then again back to 'on') there is a restriction for
112 * ALTER SUBSCRIPTION REFRESH PUBLICATION. This command is not permitted when
113 * the two_phase tri-state is ENABLED, except when copy_data = false.
114 *
115 * We can get prepare of the same GID more than once for the genuine cases
116 * where we have defined multiple subscriptions for publications on the same
117 * server and prepared transaction has operations on tables subscribed to those
118 * subscriptions. For such cases, if we use the GID sent by publisher one of
119 * the prepares will be successful and others will fail, in which case the
120 * server will send them again. Now, this can lead to a deadlock if user has
121 * set synchronous_standby_names for all the subscriptions on subscriber. To
122 * avoid such deadlocks, we generate a unique GID (consisting of the
123 * subscription oid and the xid of the prepared transaction) for each prepare
124 * transaction on the subscriber.
125 *-------------------------------------------------------------------------
126 */
130 #include <sys/stat.h>
131 #include <unistd.h>
198 {
199 dlist_node node;
200 XLogRecPtr local_end;
201 XLogRecPtr remote_end;
207 {
213 /* These fields are used when the target relation is partitioned: */
218 /* Struct for saving and restoring apply errcontext information */
220 {
224 /* Remote node information */
226 TransactionId remote_xid;
231 {
237 };
242 /* per stream context for streaming transactions */
253 /* fields valid only when processing streamed transaction */
258 /* BufFile handle of the current streaming file */
262 {
268 /* Sub-transaction data for the current streaming transaction */
270 {
282 /*
283 * Information about subtransactions of a given toplevel transaction.
284 */
290 /*
291 * Serialize and deserialize changes for a toplevel transaction.
292 */
304 /* prototype needed because of stream_commit */
325 CmdType operation);
327 /* Compute GID for two_phase transactions */
330 /* Common streaming function to apply all the spooled messages */
333 /* Functions for apply error callback */
338 /*
339 * Should this worker apply changes for given relation.
340 *
341 * This is mainly needed for initial relation data sync as that runs in
342 * separate worker process running in parallel and we need some way to skip
343 * changes coming to the main apply worker during the sync of a table.
344 *
345 * Note we need to do smaller or equals comparison for SYNCDONE state because
346 * it might hold position of end of initial slot consistent point WAL
347 * record + 1 (ie start of next record) and next record can be COMMIT of
348 * transaction we are now processing (which is what we set remote_final_lsn
349 * to in apply_handle_begin).
350 */
351 static bool
353 {
356 else
360 }
362 /*
363 * Begin one step (one INSERT, UPDATE, etc) of a replication transaction.
364 *
365 * Start a transaction, if this is the first step (else we keep using the
366 * existing transaction).
367 * Also provide a global snapshot and ensure we run in ApplyMessageContext.
368 */
369 static void
371 {
375 {
378 }
383 }
385 /*
386 * Finish up one step of a replication transaction.
387 * Callers of begin_replication_step() must also call this.
388 *
389 * We don't close out the transaction here, but we should increment
390 * the command counter to make the effects of this step visible.
391 */
392 static void
394 {
398 }
400 /*
401 * Handle streamed transactions.
402 *
403 * If in streaming mode (receiving a block of streamed transaction), we
404 * simply redirect it to a file for the proper toplevel transaction.
405 *
406 * Returns true for streamed transactions, false otherwise (regular mode).
407 */
408 static bool
410 {
411 TransactionId xid;
413 /* not in streaming mode */
420 /*
421 * We should have received XID of the subxact as the first part of the
422 * message, so extract it.
423 */
431 /* Add the new subxact to the array (unless already there). */
434 /* write the change to the current file */
438 }
440 /*
441 * Executor state preparation for evaluation of constraint expressions,
442 * indexes and triggers for the specified relation.
443 *
444 * Note that the caller must open and close any indexes to be updated.
445 */
448 {
468 /*
469 * Use Relation opened by logicalrep_rel_open() instead of opening it
470 * again.
471 */
474 /*
475 * We put the ResultRelInfo in the es_opened_result_relations list, even
476 * though we don't populate the es_result_relations array. That's a bit
477 * bogus, but it's enough to make ExecGetTriggerResultRel() find them.
478 *
479 * ExecOpenIndices() is not called here either, each execution path doing
480 * an apply operation being responsible for that.
481 */
487 /* Prepare to catch AFTER triggers. */
490 /* other fields of edata remain NULL for now */
493 }
495 /*
496 * Finish any operations related to the executor state created by
497 * create_edata_for_relation().
498 */
499 static void
501 {
504 /* Handle any queued AFTER triggers. */
507 /* Shut down tuple routing, if any was done. */
511 /*
512 * Cleanup. It might seem that we should call ExecCloseResultRelations()
513 * here, but we intentionally don't. It would close the rel we added to
514 * es_opened_result_relations above, which is wrong because we took no
515 * corresponding refcount. We rely on ExecCleanupTupleRouting() to close
516 * any other relations opened during execution.
517 */
521 }
523 /*
524 * Executes default values for columns for which we can't map to remote
525 * relation columns.
526 *
527 * This allows us to support tables which have more columns on the downstream
528 * than on the upstream.
529 */
530 static void
533 {
545 /* We got all the data via replication, no need to evaluate anything. */
554 {
566 {
567 /* Run the expression through planner */
570 /* Initialize executable expression in copycontext */
573 num_defaults++;
574 }
576 }
581 }
583 /*
584 * Store tuple data into slot.
585 *
586 * Incoming data can be either text or binary format.
587 */
588 static void
591 {
597 /* Call the "in" function for each non-dropped, non-null attribute */
600 {
605 {
610 /* Set attnum for error callback */
614 {
615 Oid typinput;
616 Oid typioparam;
623 }
625 {
626 Oid typreceive;
627 Oid typioparam;
629 /*
630 * In some code paths we may be asked to re-parse the same
631 * tuple data. Reset the StringInfo's cursor so that works.
632 */
640 /* Trouble if it didn't eat the whole buffer */
647 }
648 else
649 {
650 /*
651 * NULL value from remote. (We don't expect to see
652 * LOGICALREP_COLUMN_UNCHANGED here, but if we do, treat it as
653 * NULL.)
654 */
657 }
659 /* Reset attnum for error callback */
661 }
662 else
663 {
664 /*
665 * We assign NULL to dropped attributes and missing values
666 * (missing values should be later filled using
667 * slot_fill_defaults).
668 */
671 }
672 }
675 }
677 /*
678 * Replace updated columns with data from the LogicalRepTupleData struct.
679 * This is somewhat similar to heap_modify_tuple but also calls the type
680 * input functions on the user data.
681 *
682 * "slot" is filled with a copy of the tuple in "srcslot", replacing
683 * columns provided in "tupleData" and leaving others as-is.
684 *
685 * Caution: unreplaced pass-by-ref columns in "slot" will point into the
686 * storage for "srcslot". This is OK for current usage, but someday we may
687 * need to materialize "slot" at the end to make it independent of "srcslot".
688 */
689 static void
693 {
697 /* We'll fill "slot" with a virtual tuple, so we must start with ... */
700 /*
701 * Copy all the column data from srcslot, so that we'll have valid values
702 * for unreplaced columns.
703 */
709 /* Call the "in" function for each replaced attribute */
712 {
722 {
725 /* Set attnum for error callback */
729 {
730 Oid typinput;
731 Oid typioparam;
738 }
740 {
741 Oid typreceive;
742 Oid typioparam;
744 /*
745 * In some code paths we may be asked to re-parse the same
746 * tuple data. Reset the StringInfo's cursor so that works.
747 */
755 /* Trouble if it didn't eat the whole buffer */
762 }
763 else
764 {
765 /* must be LOGICALREP_COLUMN_NULL */
768 }
770 /* Reset attnum for error callback */
772 }
773 }
775 /* And finally, declare that "slot" contains a valid virtual tuple */
777 }
779 /*
780 * Handle BEGIN message.
781 */
782 static void
784 {
785 LogicalRepBeginData begin_data;
795 }
797 /*
798 * Handle COMMIT message.
799 *
800 * TODO, support tracking of multiple origins
801 */
802 static void
804 {
805 LogicalRepCommitData commit_data;
818 /* Process any tables that are being synchronized in parallel. */
823 }
825 /*
826 * Handle BEGIN PREPARE message.
827 */
828 static void
830 {
831 LogicalRepPreparedTxnData begin_data;
833 /* Tablesync should never receive prepare. */
847 }
849 /*
850 * Common function to prepare the GID.
851 */
852 static void
854 {
857 /*
858 * Compute unique GID for two_phase transactions. We don't use GID of
859 * prepared transaction sent by server as that can lead to deadlock when
860 * we have multiple subscriptions from same node point to publications on
861 * the same node. See comments atop worker.c
862 */
866 /*
867 * BeginTransactionBlock is necessary to balance the EndTransactionBlock
868 * called within the PrepareTransactionBlock below.
869 */
873 /*
874 * Update origin state so we can restart streaming from correct position
875 * in case of crash.
876 */
881 }
883 /*
884 * Handle PREPARE message.
885 */
886 static void
888 {
889 LogicalRepPreparedTxnData prepare_data;
900 /*
901 * Unlike commit, here, we always prepare the transaction even though no
902 * change has happened in this transaction. It is done this way because at
903 * commit prepared time, we won't know whether we have skipped preparing a
904 * transaction because of no change.
905 *
906 * XXX, We can optimize such that at commit prepared time, we first check
907 * whether we have prepared the transaction or not but that doesn't seem
908 * worthwhile because such cases shouldn't be common.
909 */
922 /* Process any tables that are being synchronized in parallel. */
927 }
929 /*
930 * Handle a COMMIT PREPARED of a previously PREPARED transaction.
931 */
932 static void
934 {
935 LogicalRepCommitPreparedTxnData prepare_data;
941 /* Compute GID for two_phase transactions. */
945 /* There is no transaction when COMMIT PREPARED is called */
948 /*
949 * Update origin state so we can restart streaming from correct position
950 * in case of crash.
951 */
963 /* Process any tables that are being synchronized in parallel. */
968 }
970 /*
971 * Handle a ROLLBACK PREPARED of a previously PREPARED TRANSACTION.
972 */
973 static void
975 {
976 LogicalRepRollbackPreparedTxnData rollback_data;
982 /* Compute GID for two_phase transactions. */
986 /*
987 * It is possible that we haven't received prepare because it occurred
988 * before walsender reached a consistent point or the two_phase was still
989 * not enabled by that time, so in such cases, we need to skip rollback
990 * prepared.
991 */
994 {
995 /*
996 * Update origin state so we can restart streaming from correct
997 * position in case of crash.
998 */
1002 /* There is no transaction when ABORT/ROLLBACK PREPARED is called */
1007 }
1014 /* Process any tables that are being synchronized in parallel. */
1019 }
1021 /*
1022 * Handle STREAM PREPARE.
1023 *
1024 * Logic is in two parts:
1025 * 1. Replay all the spooled operations
1026 * 2. Mark the transaction as prepared
1027 */
1028 static void
1030 {
1031 LogicalRepPreparedTxnData prepare_data;
1038 /* Tablesync should never receive prepare. */
1049 /* Replay all the spooled operations. */
1052 /* Mark the transaction as prepared. */
1063 /* unlink the files with serialized changes and subxact info. */
1066 /* Process any tables that are being synchronized in parallel. */
1072 }
1074 /*
1075 * Handle ORIGIN message.
1076 *
1077 * TODO, support tracking of multiple origins
1078 */
1079 static void
1081 {
1082 /*
1083 * ORIGIN message can only come inside streaming transaction or inside
1084 * remote transaction and before any actual writes.
1085 */
1092 }
1094 /*
1095 * Handle STREAM START message.
1096 */
1097 static void
1099 {
1107 /*
1108 * Start a transaction on stream start, this transaction will be committed
1109 * on the stream stop unless it is a tablesync worker in which case it
1110 * will be committed after processing all the messages. We need the
1111 * transaction for handling the buffile, used for serializing the
1112 * streaming data and subxact info.
1113 */
1116 /* notify handle methods we're processing a remote transaction */
1119 /* extract XID of the top-level transaction */
1129 /*
1130 * Initialize the worker's stream_fileset if we haven't yet. This will be
1131 * used for the entire duration of the worker so create it in a permanent
1132 * context. We create this on the very first streaming message from any
1133 * transaction and then use it for this and other streaming transactions.
1134 * Now, we could create a fileset at the start of the worker as well but
1135 * then we won't be sure that it will ever be used.
1136 */
1138 {
1139 MemoryContext oldctx;
1147 }
1149 /* open the spool file for this transaction */
1152 /* if this is not the first segment, open existing subxact file */
1159 }
1161 /*
1162 * Handle STREAM STOP message.
1163 */
1164 static void
1166 {
1172 /*
1173 * Close the file with serialized changes, and serialize information about
1174 * subxacts for the toplevel transaction.
1175 */
1179 /* We must be in a valid transaction state */
1182 /* Commit the per-stream transaction */
1187 /* Reset per-stream context */
1192 }
1194 /*
1195 * Handle STREAM abort message.
1196 */
1197 static void
1199 {
1200 TransactionId xid;
1201 TransactionId subxid;
1210 /*
1211 * If the two XIDs are the same, it's in fact abort of toplevel xact, so
1212 * just delete the files with serialized info.
1213 */
1215 {
1218 }
1219 else
1220 {
1221 /*
1222 * OK, so it's a subxact. We need to read the subxact file for the
1223 * toplevel transaction, determine the offset tracked for the subxact,
1224 * and truncate the file with changes. We also remove the subxacts
1225 * with higher offsets (or rather higher XIDs).
1226 *
1227 * We intentionally scan the array from the tail, because we're likely
1228 * aborting a change for the most recent subtransactions.
1229 *
1230 * We can't use the binary search here as subxact XIDs won't
1231 * necessarily arrive in sorted order, consider the case where we have
1232 * released the savepoint for multiple subtransactions and then
1233 * performed rollback to savepoint for one of the earlier
1234 * sub-transaction.
1235 */
1236 int64 i;
1237 int64 subidx;
1249 {
1251 {
1255 }
1256 }
1258 /*
1259 * If it's an empty sub-transaction then we will not find the subxid
1260 * here so just cleanup the subxact info and return.
1261 */
1263 {
1264 /* Cleanup the subxact info */
1270 }
1272 /* open the changes file */
1277 /* OK, truncate the file at the right offset */
1282 /* discard the subxacts added later */
1285 /* write the updated subxact list */
1290 }
1293 }
1295 /*
1296 * Common spoolfile processing.
1297 */
1298 static void
1300 {
1301 StringInfoData s2;
1305 MemoryContext oldcxt;
1308 /* Make sure we have an open transaction */
1311 /*
1312 * Allocate file handle and memory required to process all the messages in
1313 * TopTransactionContext to avoid them getting reset after each message is
1314 * processed.
1315 */
1318 /* Open the spool file for the committed/prepared transaction */
1332 /*
1333 * Make sure the handle apply_dispatch methods are aware we're in a remote
1334 * transaction.
1335 */
1341 /*
1342 * Read the entries one by one and pass them through the same logic as in
1343 * apply_dispatch.
1344 */
1347 {
1353 /* read length of the on-disk record */
1356 /* have we reached end of the file? */
1360 /* do we have a correct length? */
1365 path)));
1371 /* make sure we have sufficiently large buffer */
1374 /* and finally read the data into the buffer */
1379 path)));
1381 /* copy the buffer to the stringinfo and call apply_dispatch */
1385 /* Ensure we are reading the data into our memory context. */
1394 nchanges++;
1399 }
1410 }
1412 /*
1413 * Handle STREAM COMMIT message.
1414 */
1415 static void
1417 {
1418 TransactionId xid;
1419 LogicalRepCommitData commit_data;
1435 /* unlink the files with serialized changes and subxact info */
1438 /* Process any tables that are being synchronized in parallel. */
1444 }
1446 /*
1447 * Helper function for apply_handle_commit and apply_handle_stream_commit.
1448 */
1449 static void
1451 {
1453 {
1454 /*
1455 * Update origin state so we can restart streaming from correct
1456 * position in case of crash.
1457 */
1465 }
1466 else
1467 {
1468 /* Process any invalidation messages that might have accumulated. */
1471 }
1474 }
1476 /*
1477 * Handle RELATION message.
1478 *
1479 * Note we don't do validation against local schema here. The validation
1480 * against local schema is postponed until first change for given relation
1481 * comes as we only care about it when applying changes for it anyway and we
1482 * do less locking this way.
1483 */
1484 static void
1486 {
1494 }
1496 /*
1497 * Handle TYPE message.
1498 *
1499 * This implementation pays no attention to TYPE messages; we expect the user
1500 * to have set things up so that the incoming data is acceptable to the input
1501 * functions for the locally subscribed tables. Hence, we just read and
1502 * discard the message.
1503 */
1504 static void
1506 {
1507 LogicalRepTyp typ;
1513 }
1515 /*
1516 * Get replica identity index or if it is not defined a primary key.
1517 *
1518 * If neither is defined, returns InvalidOid
1519 */
1520 static Oid
1522 {
1523 Oid idxoid;
1531 }
1533 /*
1534 * Handle INSERT message.
1535 */
1537 static void
1539 {
1541 LogicalRepTupleData newtup;
1542 LogicalRepRelId relid;
1546 MemoryContext oldctx;
1556 {
1557 /*
1558 * The relation can't become interesting in the middle of the
1559 * transaction so it's safe to unlock it.
1560 */
1564 }
1566 /* Set relation for error callback */
1569 /* Initialize the executor state. */
1576 /* Process and store remote tuple in the slot */
1582 /* For a partitioned table, insert the tuple into a partition. */
1586 else
1588 remoteslot);
1592 /* Reset relation for error callback */
1598 }
1600 /*
1601 * Workhorse for apply_handle_insert()
1602 * relinfo is for the relation we're actually inserting into
1603 * (could be a child partition of edata->targetRelInfo)
1604 */
1605 static void
1609 {
1612 /* We must open indexes here. */
1615 /* Do the insert. */
1618 /* Cleanup. */
1620 }
1622 /*
1623 * Check if the logical replication relation is updatable and throw
1624 * appropriate error if it isn't.
1625 */
1626 static void
1628 {
1629 /* Updatable, no error. */
1633 /*
1634 * We are in error mode so it's fine this is somewhat slow. It's better to
1635 * give user correct error.
1636 */
1638 {
1644 }
1649 "neither REPLICA IDENTITY index nor PRIMARY "
1650 "KEY and published relation does not have "
1653 }
1655 /*
1656 * Handle UPDATE message.
1657 *
1658 * TODO: FDW support
1659 */
1660 static void
1662 {
1664 LogicalRepRelId relid;
1667 LogicalRepTupleData oldtup;
1668 LogicalRepTupleData newtup;
1672 MemoryContext oldctx;
1683 {
1684 /*
1685 * The relation can't become interesting in the middle of the
1686 * transaction so it's safe to unlock it.
1687 */
1691 }
1693 /* Set relation for error callback */
1696 /* Check if we can do the update. */
1699 /* Initialize the executor state. */
1706 /*
1707 * Populate updatedCols so that per-column triggers can fire, and so
1708 * executor can correctly pass down indexUnchanged hint. This could
1709 * include more columns than were actually changed on the publisher
1710 * because the logical replication protocol doesn't contain that
1711 * information. But it would for example exclude columns that only exist
1712 * on the subscriber, since we are not touching those.
1713 */
1716 {
1721 {
1727 }
1728 }
1730 /* Also populate extraUpdatedCols, in case we have generated columns */
1733 /* Build the search tuple. */
1739 /* For a partitioned table, apply update to correct partition. */
1743 else
1749 /* Reset relation for error callback */
1755 }
1757 /*
1758 * Workhorse for apply_handle_update()
1759 * relinfo is for the relation we're actually updating in
1760 * (could be a child partition of edata->targetRelInfo)
1761 */
1762 static void
1767 {
1771 EPQState epqstate;
1774 MemoryContext oldctx;
1784 /*
1785 * Tuple found.
1786 *
1787 * Note this will fail if there are other conflicting unique indexes.
1788 */
1790 {
1791 /* Process and store remote tuple in the slot */
1798 /* Do the actual update. */
1800 remoteslot);
1801 }
1802 else
1803 {
1804 /*
1805 * The tuple to be updated could not be found. Do nothing except for
1806 * emitting a log message.
1807 *
1808 * XXX should this be promoted to ereport(LOG) perhaps?
1809 */
1811 "logical replication did not find row to be updated "
1814 }
1816 /* Cleanup. */
1819 }
1821 /*
1822 * Handle DELETE message.
1823 *
1824 * TODO: FDW support
1825 */
1826 static void
1828 {
1830 LogicalRepTupleData oldtup;
1831 LogicalRepRelId relid;
1835 MemoryContext oldctx;
1845 {
1846 /*
1847 * The relation can't become interesting in the middle of the
1848 * transaction so it's safe to unlock it.
1849 */
1853 }
1855 /* Set relation for error callback */
1858 /* Check if we can do the delete. */
1861 /* Initialize the executor state. */
1868 /* Build the search tuple. */
1873 /* For a partitioned table, apply delete to correct partition. */
1877 else
1879 remoteslot);
1883 /* Reset relation for error callback */
1889 }
1891 /*
1892 * Workhorse for apply_handle_delete()
1893 * relinfo is for the relation we're actually deleting from
1894 * (could be a child partition of edata->targetRelInfo)
1895 */
1896 static void
1900 {
1904 EPQState epqstate;
1914 /* If found delete it. */
1916 {
1919 /* Do the actual delete. */
1921 }
1922 else
1923 {
1924 /*
1925 * The tuple to be deleted could not be found. Do nothing except for
1926 * emitting a log message.
1927 *
1928 * XXX should this be promoted to ereport(LOG) perhaps?
1929 */
1931 "logical replication did not find row to be deleted "
1934 }
1936 /* Cleanup. */
1939 }
1941 /*
1942 * Try to find a tuple received from the publication side (in 'remoteslot') in
1943 * the corresponding local relation using either replica identity index,
1944 * primary key or if needed, sequential scan.
1945 *
1946 * Local tuple, if found, is returned in '*localslot'.
1947 */
1948 static bool
1953 {
1954 Oid idxoid;
1965 LockTupleExclusive,
1967 else
1972 }
1974 /*
1975 * This handles insert, update, delete on a partitioned table.
1976 */
1977 static void
1981 CmdType operation)
1982 {
1990 Relation partrel;
1993 MemoryContext oldctx;
1995 /* ModifyTableState is needed for ExecFindPartition(). */
2002 /* ... as is PartitionTupleRouting. */
2005 /*
2006 * Find the partition to which the "search tuple" belongs.
2007 */
2015 /*
2016 * To perform any of the operations below, the tuple must match the
2017 * partition's rowtype. Convert if needed or just copy, using a dedicated
2018 * slot to store the tuple in any case.
2019 */
2026 remoteslot_part);
2027 else
2028 {
2031 }
2035 {
2038 remoteslot_part);
2043 remoteslot_part);
2048 /*
2049 * For UPDATE, depending on whether or not the updated tuple
2050 * satisfies the partition's constraint, perform a simple UPDATE
2051 * of the partition or move the updated tuple into a different
2052 * suitable partition.
2053 */
2054 {
2062 attrmap);
2064 /* Get the matching local tuple from the partition. */
2069 {
2070 /*
2071 * The tuple to be updated could not be found. Do nothing
2072 * except for emitting a log message.
2073 *
2074 * XXX should this be promoted to ereport(LOG) perhaps?
2075 */
2077 "logical replication did not find row to be updated "
2081 }
2083 /*
2084 * Apply the update to the local tuple, putting the result in
2085 * remoteslot_part.
2086 */
2089 newtup);
2092 /*
2093 * Does the updated tuple still satisfy the current
2094 * partition's constraint?
2095 */
2099 {
2100 /*
2101 * Yes, so simply UPDATE the partition. We don't call
2102 * apply_handle_update_internal() here, which would
2103 * normally do the following work, to avoid repeating some
2104 * work already done above to find the local tuple in the
2105 * partition.
2106 */
2107 EPQState epqstate;
2117 }
2118 else
2119 {
2120 /* Move the tuple into the new partition. */
2122 /*
2123 * New partition will be found using tuple routing, which
2124 * can only occur via the parent table. We might need to
2125 * convert the tuple to the parent's rowtype. Note that
2126 * this is the tuple found in the partition, not the
2127 * original search tuple received by this function.
2128 */
2130 {
2135 remoteslot =
2138 }
2139 else
2140 {
2143 }
2146 /* Find the new partition. */
2150 estate);
2154 /* DELETE old tuple found in the old partition. */
2156 localslot);
2158 /* INSERT new tuple into the new partition. */
2160 /*
2161 * Convert the replacement tuple to match the destination
2162 * partition rowtype.
2163 */
2172 {
2174 remoteslot,
2175 remoteslot_part);
2176 }
2177 else
2178 {
2180 remoteslot);
2182 }
2185 remoteslot_part);
2186 }
2187 }
2193 }
2194 }
2196 /*
2197 * Handle TRUNCATE message.
2198 *
2199 * TODO: FDW support
2200 */
2201 static void
2203 {
2223 {
2229 {
2230 /*
2231 * The relation can't become interesting in the middle of the
2232 * transaction so it's safe to unlock it.
2233 */
2236 }
2244 /*
2245 * Truncate partitions if we got a message to truncate a partitioned
2246 * table.
2247 */
2249 {
2252 lockmode,
2253 NULL);
2256 {
2258 Relation childrel;
2263 /* find_all_inheritors already got lock */
2266 /*
2267 * Ignore temp tables of other backends. See similar code in
2268 * ExecuteTruncate().
2269 */
2271 {
2274 }
2279 /* Log this relation only if needed for logical decoding */
2282 }
2283 }
2284 }
2286 /*
2287 * Even if we used CASCADE on the upstream primary we explicitly default
2288 * to replaying changes without further cascading. This might be later
2289 * changeable with a user specified option.
2290 */
2292 relids,
2293 relids_logged,
2294 DROP_RESTRICT,
2295 restart_seqs);
2297 {
2301 }
2303 {
2307 }
2310 }
2313 /*
2314 * Logical replication protocol message dispatcher.
2315 */
2316 static void
2318 {
2320 LogicalRepMsgType saved_command;
2322 /*
2323 * Set the current command being applied. Since this function can be
2324 * called recusively when applying spooled changes, save the current
2325 * command.
2326 */
2331 {
2370 /*
2371 * Logical replication does not use generic logical messages yet.
2372 * Although, it could be used by other applications that use this
2373 * output plugin.
2374 */
2417 }
2419 /* Reset the current command */
2421 }
2423 /*
2424 * Figure out which write/flush positions to report to the walsender process.
2425 *
2426 * We can't simply report back the last LSN the walsender sent us because the
2427 * local transaction might not yet be flushed to disk locally. Instead we
2428 * build a list that associates local with remote LSNs for every commit. When
2429 * reporting back the flush position to the sender we iterate that list and
2430 * check which entries on it are already locally flushed. Those we can report
2431 * as having been flushed.
2432 *
2433 * The have_pending_txes is true if there are outstanding transactions that
2434 * need to be flushed.
2435 */
2436 static void
2439 {
2440 dlist_mutable_iter iter;
2447 {
2454 {
2458 }
2459 else
2460 {
2461 /*
2462 * Don't want to uselessly iterate over the rest of the list which
2463 * could potentially be long. Instead get the last element and
2464 * grab the write position from there.
2465 */
2471 }
2472 }
2475 }
2477 /*
2478 * Store current remote/local lsn pair in the tracking list.
2479 */
2480 static void
2482 {
2485 /* Need to do this in permanent context */
2488 /* Track commit lsn */
2495 }
2498 /* Update statistics of the worker. */
2499 static void
2501 {
2506 {
2509 }
2510 }
2512 /*
2513 * Apply main loop.
2514 */
2515 static void
2517 {
2520 TimeLineID tli;
2521 ErrorContextCallback errcallback;
2523 /*
2524 * Init the ApplyMessageContext which we clean up after each replication
2525 * protocol message.
2526 */
2529 ALLOCSET_DEFAULT_SIZES);
2531 /*
2532 * This memory context is used for per-stream data when the streaming mode
2533 * is enabled. This context is reset on each stream stop.
2534 */
2537 ALLOCSET_DEFAULT_SIZES);
2539 /* mark as idle, before starting to loop */
2542 /*
2543 * Push apply error context callback. Fields will be filled during
2544 * applying a change.
2545 */
2550 /* This outer loop iterates once per wait. */
2552 {
2567 {
2568 /* Loop to process all available data (without blocking). */
2570 {
2574 {
2576 }
2578 {
2583 }
2584 else
2585 {
2587 StringInfoData s;
2589 /* Reset timeout. */
2593 /* Ensure we are reading the data into our memory context. */
2604 {
2605 XLogRecPtr start_lsn;
2606 XLogRecPtr end_lsn;
2607 TimestampTz send_time;
2622 }
2624 {
2625 XLogRecPtr end_lsn;
2626 TimestampTz timestamp;
2638 }
2639 /* other message types are purposefully ignored */
2642 }
2645 }
2646 }
2648 /* confirm all writes so far */
2652 {
2653 /*
2654 * If we didn't get any transactions for a while there might be
2655 * unconsumed invalidation messages in the queue, consume them
2656 * now.
2657 */
2661 /* Process any table synchronization changes. */
2663 }
2665 /* Cleanup the memory. */
2669 /* Check if we need to exit the streaming loop. */
2673 /*
2674 * Wait for more data or latch. If we have unflushed transactions,
2675 * wake up after WalWriterDelay to see if they've been flushed yet (in
2676 * which case we should send a feedback message). Otherwise, there's
2677 * no particular urgency about waking up unless we get data or a
2678 * signal.
2679 */
2682 else
2689 WAIT_EVENT_LOGICAL_APPLY_MAIN);
2692 {
2695 }
2698 {
2701 }
2704 {
2705 /*
2706 * We didn't receive anything new. If we haven't heard anything
2707 * from the server for more than wal_receiver_timeout / 2, ping
2708 * the server. Also, if it's been longer than
2709 * wal_receiver_status_interval since the last update we sent,
2710 * send a status update to the primary anyway, to report any
2711 * progress in applying WAL.
2712 */
2715 /*
2716 * Check if time since last receive from primary has reached the
2717 * configured limit.
2718 */
2720 {
2722 TimestampTz timeout;
2724 timeout =
2726 wal_receiver_timeout);
2733 /* Check to see if it's time for a ping. */
2735 {
2739 {
2742 }
2743 }
2744 }
2747 }
2748 }
2750 /* Pop the error context stack */
2753 /* All done */
2755 }
2757 /*
2758 * Send a Standby Status Update message to server.
2759 *
2760 * 'recvpos' is the latest LSN we've received data to, force is set if we need
2761 * to send a response to avoid timeouts.
2762 */
2763 static void
2765 {
2773 XLogRecPtr writepos;
2774 XLogRecPtr flushpos;
2775 TimestampTz now;
2778 /*
2779 * If the user doesn't want status to be reported to the publisher, be
2780 * sure to exit before doing anything at all.
2781 */
2785 /* It's legal to not pass a recvpos */
2791 /*
2792 * No outstanding transactions to flush, we can report the latest received
2793 * position. This is important for synchronous replication.
2794 */
2806 /* if we've already reported everything we're good */
2816 {
2821 }
2822 else
2833 force,
2847 }
2849 /*
2850 * Reread subscription info if needed. Most changes will be exit.
2851 */
2852 static void
2854 {
2855 MemoryContext oldctx;
2859 /* When cache state is valid there is nothing to do here. */
2863 /* This function might be called inside or outside of transaction. */
2865 {
2868 }
2870 /* Ensure allocations in permanent context. */
2875 /*
2876 * Exit if the subscription was removed. This normally should not happen
2877 * as the worker gets killed during DROP SUBSCRIPTION.
2878 */
2880 {
2887 }
2889 /*
2890 * Exit if the subscription was disabled. This normally should not happen
2891 * as the worker gets killed during ALTER SUBSCRIPTION ... DISABLE.
2892 */
2894 {
2901 }
2903 /* !slotname should never happen when enabled is true. */
2906 /* two-phase should not be altered */
2909 /*
2910 * Exit if any parameter that affects the remote connection was changed.
2911 * The launcher will start a new worker.
2912 */
2919 {
2921 (errmsg("logical replication apply worker for subscription \"%s\" will restart because of a parameter change",
2925 }
2927 /* Check for other changes that should never happen too. */
2929 {
2932 }
2934 /* Clean old subscription info and switch to new one. */
2940 /* Change synchronous commit according to the user's wishes */
2948 }
2950 /*
2951 * Callback from subscription syscache invalidation.
2952 */
2953 static void
2955 {
2957 }
2959 /*
2960 * subxact_info_write
2961 * Store information about subxacts for a toplevel transaction.
2962 *
2963 * For each subxact we store offset of it's first change in the main file.
2964 * The file is always over-written as a whole.
2965 *
2966 * XXX We should only store subxacts that were not aborted yet.
2967 */
2968 static void
2970 {
2972 Size len;
2977 /* construct the subxact filename */
2980 /* Delete the subxacts file, if exists. */
2982 {
2987 }
2989 /*
2990 * Create the subxact file if it not already created, otherwise open the
2991 * existing file.
2992 */
3000 /* Write the subxact count and subxact info */
3006 /* free the memory allocated for subxact info */
3008 }
3010 /*
3011 * subxact_info_read
3012 * Restore information about subxacts of a streamed transaction.
3013 *
3014 * Read information about subxacts into the structure subxact_data that can be
3015 * used later.
3016 */
3017 static void
3019 {
3021 Size len;
3023 MemoryContext oldctx;
3029 /*
3030 * If the subxact file doesn't exist that means we don't have any subxact
3031 * info.
3032 */
3039 /* read number of subxact items */
3046 path)));
3050 /* we keep the maximum as a power of 2 */
3053 /*
3054 * Allocate subxact information in the logical streaming context. We need
3055 * this information during the complete stream so that we can add the sub
3056 * transaction info to this. On stream stop we will flush this information
3057 * to the subxact file and reset the logical streaming context.
3058 */
3068 path)));
3071 }
3073 /*
3074 * subxact_info_add
3075 * Add information about a subxact (offset in the main file).
3076 */
3077 static void
3079 {
3081 int64 i;
3083 /* We must have a valid top level stream xid and a stream fd. */
3087 /*
3088 * If the XID matches the toplevel transaction, we don't want to add it.
3089 */
3093 /*
3094 * In most cases we're checking the same subxact as we've already seen in
3095 * the last call, so make sure to ignore it (this change comes later).
3096 */
3100 /* OK, remember we're processing this XID. */
3103 /*
3104 * Check if the transaction is already present in the array of subxact. We
3105 * intentionally scan the array from the tail, because we're likely adding
3106 * a change for the most recent subtransactions.
3107 *
3108 * XXX Can we rely on the subxact XIDs arriving in sorted order? That
3109 * would allow us to use binary search here.
3110 */
3112 {
3113 /* found, so we're done */
3116 }
3118 /* This is a new subxact, so we need to add it to the array. */
3120 {
3121 MemoryContext oldctx;
3125 /*
3126 * Allocate this memory for subxacts in per-stream context, see
3127 * subxact_info_read.
3128 */
3132 }
3134 {
3138 }
3142 /*
3143 * Get the current offset of the stream file and store it as offset of
3144 * this subxact.
3145 */
3152 }
3154 /* format filename for file containing the info about subxacts */
3157 {
3159 }
3161 /* format filename for file containing serialized changes */
3164 {
3166 }
3168 /*
3169 * stream_cleanup_files
3170 * Cleanup files for a subscription / toplevel transaction.
3171 *
3172 * Remove files with serialized changes and subxact info for a particular
3173 * toplevel transaction. Each subscription has a separate set of files
3174 * for any toplevel transaction.
3175 */
3176 static void
3178 {
3181 /* Delete the changes file. */
3185 /* Delete the subxact file, if it exists. */
3188 }
3190 /*
3191 * stream_open_file
3192 * Open a file that we'll use to serialize changes for a toplevel
3193 * transaction.
3194 *
3195 * Open a file for streamed changes from a toplevel transaction identified
3196 * by stream_xid (global variable). If it's the first chunk of streamed
3197 * changes for this transaction, create the buffile, otherwise open the
3198 * previously created file.
3199 *
3200 * This can only be called at the beginning of a "streaming" block, i.e.
3201 * between stream_start/stream_stop messages from the upstream.
3202 */
3203 static void
3205 {
3207 MemoryContext oldcxt;
3218 /*
3219 * Create/open the buffiles under the logical streaming context so that we
3220 * have those files until stream stop.
3221 */
3224 /*
3225 * If this is the first streamed segment, create the changes file.
3226 * Otherwise, just open the file for writing, in append mode.
3227 */
3230 path);
3231 else
3232 {
3233 /*
3234 * Open the file and seek to the end of the file because we always
3235 * append the changes file.
3236 */
3240 }
3243 }
3245 /*
3246 * stream_close_file
3247 * Close the currently open file with streamed changes.
3248 *
3249 * This can only be called at the end of a streaming block, i.e. at stream_stop
3250 * message from the upstream.
3251 */
3252 static void
3254 {
3263 }
3265 /*
3266 * stream_write_change
3267 * Serialize a change to a file for the current toplevel transaction.
3268 *
3269 * The change is serialized in a simple format, with length (not including
3270 * the length), action code (identifying the message type) and message
3271 * contents (without the subxact TransactionId value).
3272 */
3273 static void
3275 {
3282 /* total on-disk size, including the action type character */
3285 /* first write the size */
3288 /* then the action */
3291 /* and finally the remaining part of the buffer (after the XID) */
3295 }
3297 /*
3298 * Cleanup the memory for subxacts and reset the related variables.
3299 */
3302 {
3310 }
3312 /*
3313 * Form the prepared transaction GID for two_phase transactions.
3314 *
3315 * Return the GID in the supplied buffer.
3316 */
3317 static void
3319 {
3328 }
3330 /* Logical Replication Apply worker entry point */
3331 void
3333 {
3336 MemoryContext oldctx;
3338 XLogRecPtr origin_startpos;
3340 WalRcvStreamOptions options;
3343 /* Attach to slot */
3346 /* Setup signal handling */
3351 /*
3352 * We don't currently need any ResourceOwner in a walreceiver process, but
3353 * if we did, we could call CreateAuxProcessResourceOwner here.
3354 */
3356 /* Initialise stats to a sanish value */
3360 /* Load the libpq-specific functions */
3363 /* Run as replica session replication role. */
3367 /* Connect to our database. */
3372 /*
3373 * Set always-secure search path, so malicious users can't redirect user
3374 * code (e.g. pg_index.indexprs).
3375 */
3378 /* Load the subscription into persistent memory context. */
3381 ALLOCSET_DEFAULT_SIZES);
3387 {
3393 }
3399 {
3406 }
3408 /* Setup synchronous commit according to the user's wishes */
3412 /* Keep us informed about subscription changes. */
3414 subscription_change_cb,
3419 (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
3421 else
3428 /* Connect to the origin and start the replication. */
3433 {
3437 {
3438 /* This is table synchronization worker, call initial sync. */
3440 }
3442 {
3446 /*
3447 * Report the table sync error. There is no corresponding message
3448 * type for table synchronization.
3449 */
3454 InvalidTransactionId,
3458 }
3461 /* allocate slot name in long-lived context */
3465 }
3466 else
3467 {
3468 /* This is main apply worker */
3469 RepOriginId originid;
3470 TimeLineID startpointTLI;
3475 /*
3476 * This shouldn't happen if the subscription is enabled, but guard
3477 * against DDL bugs or manual catalog changes. (libpqwalreceiver will
3478 * crash if slot is NULL.)
3479 */
3485 /* Setup replication origin tracking. */
3503 /*
3504 * We don't really use the output identify_system for anything but it
3505 * does some initializations on the upstream so let's still call it.
3506 */
3508 }
3510 /*
3511 * Setup callback for syscache so that we know when something changes in
3512 * the subscription relation state.
3513 */
3515 invalidate_syncing_table_states,
3518 /* Build logical replication streaming options. */
3527 LOGICALREP_PROTO_VERSION_NUM;
3535 {
3536 /*
3537 * Even when the two_phase mode is requested by the user, it remains
3538 * as the tri-state PENDING until all tablesyncs have reached READY
3539 * state. Only then, can it become ENABLED.
3540 *
3541 * Note: If the subscription has no tables then leave the state as
3542 * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
3543 * work.
3544 */
3547 {
3548 /* Start streaming with two_phase enabled */
3556 }
3557 else
3558 {
3560 }
3569 }
3570 else
3571 {
3572 /* Start normal logical streaming replication. */
3574 }
3576 /* Run the main loop. */
3578 {
3580 }
3582 {
3583 /* report the apply error */
3585 {
3598 }
3601 }
3605 }
3607 /*
3608 * Is current process a logical replication worker?
3609 */
3610 bool
3612 {
3614 }
3616 /* Error callback to give more context info about the change being applied */
3617 static void
3619 {
3620 StringInfoData buf;
3630 /* append relation information */
3632 {
3639 }
3641 /* append transaction information */
3643 {
3648 }
3652 }
3654 /* Set transaction information of apply error callback */
3657 {
3660 }
3662 /* Reset all information of apply error callback */
3665 {
3670 }