@@ -226,7 +226,6 @@ typedef struct ApplyErrorCallbackArg
226
226
/* Remote node information */
227
227
int remote_attnum ; /* -1 if invalid */
228
228
TransactionId remote_xid ;
229
- TimestampTz ts ; /* commit, rollback, or prepare timestamp */
230
229
} ApplyErrorCallbackArg ;
231
230
232
231
static ApplyErrorCallbackArg apply_error_callback_arg =
@@ -235,7 +234,6 @@ static ApplyErrorCallbackArg apply_error_callback_arg =
235
234
.rel = NULL ,
236
235
.remote_attnum = -1 ,
237
236
.remote_xid = InvalidTransactionId ,
238
- .ts = 0 ,
239
237
};
240
238
241
239
static MemoryContext ApplyMessageContext = NULL ;
@@ -334,7 +332,7 @@ static void apply_spooled_messages(TransactionId xid, XLogRecPtr lsn);
334
332
335
333
/* Functions for apply error callback */
336
334
static void apply_error_callback (void * arg );
337
- static inline void set_apply_error_context_xact (TransactionId xid , TimestampTz ts );
335
+ static inline void set_apply_error_context_xact (TransactionId xid );
338
336
static inline void reset_apply_error_context_info (void );
339
337
340
338
/*
@@ -787,7 +785,7 @@ apply_handle_begin(StringInfo s)
787
785
LogicalRepBeginData begin_data ;
788
786
789
787
logicalrep_read_begin (s , & begin_data );
790
- set_apply_error_context_xact (begin_data .xid , begin_data . committime );
788
+ set_apply_error_context_xact (begin_data .xid );
791
789
792
790
remote_final_lsn = begin_data .final_lsn ;
793
791
@@ -839,7 +837,7 @@ apply_handle_begin_prepare(StringInfo s)
839
837
errmsg_internal ("tablesync worker received a BEGIN PREPARE message" )));
840
838
841
839
logicalrep_read_begin_prepare (s , & begin_data );
842
- set_apply_error_context_xact (begin_data .xid , begin_data . prepare_time );
840
+ set_apply_error_context_xact (begin_data .xid );
843
841
844
842
remote_final_lsn = begin_data .prepare_lsn ;
845
843
@@ -938,7 +936,7 @@ apply_handle_commit_prepared(StringInfo s)
938
936
char gid [GIDSIZE ];
939
937
940
938
logicalrep_read_commit_prepared (s , & prepare_data );
941
- set_apply_error_context_xact (prepare_data .xid , prepare_data . commit_time );
939
+ set_apply_error_context_xact (prepare_data .xid );
942
940
943
941
/* Compute GID for two_phase transactions. */
944
942
TwoPhaseTransactionGid (MySubscription -> oid , prepare_data .xid ,
@@ -979,7 +977,7 @@ apply_handle_rollback_prepared(StringInfo s)
979
977
char gid [GIDSIZE ];
980
978
981
979
logicalrep_read_rollback_prepared (s , & rollback_data );
982
- set_apply_error_context_xact (rollback_data .xid , rollback_data . rollback_time );
980
+ set_apply_error_context_xact (rollback_data .xid );
983
981
984
982
/* Compute GID for two_phase transactions. */
985
983
TwoPhaseTransactionGid (MySubscription -> oid , rollback_data .xid ,
@@ -1044,7 +1042,7 @@ apply_handle_stream_prepare(StringInfo s)
1044
1042
errmsg_internal ("tablesync worker received a STREAM PREPARE message" )));
1045
1043
1046
1044
logicalrep_read_stream_prepare (s , & prepare_data );
1047
- set_apply_error_context_xact (prepare_data .xid , prepare_data . prepare_time );
1045
+ set_apply_error_context_xact (prepare_data .xid );
1048
1046
1049
1047
elog (DEBUG1 , "received prepare for streamed transaction %u" , prepare_data .xid );
1050
1048
@@ -1126,7 +1124,7 @@ apply_handle_stream_start(StringInfo s)
1126
1124
(errcode (ERRCODE_PROTOCOL_VIOLATION ),
1127
1125
errmsg_internal ("invalid transaction ID in streamed replication transaction" )));
1128
1126
1129
- set_apply_error_context_xact (stream_xid , 0 );
1127
+ set_apply_error_context_xact (stream_xid );
1130
1128
1131
1129
/*
1132
1130
* Initialize the worker's stream_fileset if we haven't yet. This will be
@@ -1215,7 +1213,7 @@ apply_handle_stream_abort(StringInfo s)
1215
1213
*/
1216
1214
if (xid == subxid )
1217
1215
{
1218
- set_apply_error_context_xact (xid , 0 );
1216
+ set_apply_error_context_xact (xid );
1219
1217
stream_cleanup_files (MyLogicalRepWorker -> subid , xid );
1220
1218
}
1221
1219
else
@@ -1241,7 +1239,7 @@ apply_handle_stream_abort(StringInfo s)
1241
1239
bool found = false;
1242
1240
char path [MAXPGPATH ];
1243
1241
1244
- set_apply_error_context_xact (subxid , 0 );
1242
+ set_apply_error_context_xact (subxid );
1245
1243
1246
1244
subidx = -1 ;
1247
1245
begin_replication_step ();
@@ -1426,7 +1424,7 @@ apply_handle_stream_commit(StringInfo s)
1426
1424
errmsg_internal ("STREAM COMMIT message without STREAM STOP" )));
1427
1425
1428
1426
xid = logicalrep_read_stream_commit (s , & commit_data );
1429
- set_apply_error_context_xact (xid , commit_data . committime );
1427
+ set_apply_error_context_xact (xid );
1430
1428
1431
1429
elog (DEBUG1 , "received commit for streamed transaction %u" , xid );
1432
1430
@@ -3648,46 +3646,41 @@ IsLogicalWorker(void)
3648
3646
static void
3649
3647
apply_error_callback (void * arg )
3650
3648
{
3651
- StringInfoData buf ;
3652
3649
ApplyErrorCallbackArg * errarg = & apply_error_callback_arg ;
3653
3650
3654
3651
if (apply_error_callback_arg .command == 0 )
3655
3652
return ;
3656
3653
3657
- initStringInfo (& buf );
3658
- appendStringInfo (& buf , _ ("processing remote data during \"%s\"" ),
3659
- logicalrep_message_type (errarg -> command ));
3660
-
3661
- /* append relation information */
3662
- if (errarg -> rel )
3663
- {
3664
- appendStringInfo (& buf , _ (" for replication target relation \"%s.%s\"" ),
3665
- errarg -> rel -> remoterel .nspname ,
3666
- errarg -> rel -> remoterel .relname );
3667
- if (errarg -> remote_attnum >= 0 )
3668
- appendStringInfo (& buf , _ (" column \"%s\"" ),
3669
- errarg -> rel -> remoterel .attnames [errarg -> remote_attnum ]);
3670
- }
3671
-
3672
- /* append transaction information */
3673
- if (TransactionIdIsNormal (errarg -> remote_xid ))
3654
+ if (errarg -> rel == NULL )
3674
3655
{
3675
- appendStringInfo (& buf , _ (" in transaction %u" ), errarg -> remote_xid );
3676
- if (errarg -> ts != 0 )
3677
- appendStringInfo (& buf , _ (" at %s" ),
3678
- timestamptz_to_str (errarg -> ts ));
3656
+ if (!TransactionIdIsValid (errarg -> remote_xid ))
3657
+ errcontext ("processing remote data during \"%s\"" ,
3658
+ logicalrep_message_type (errarg -> command ));
3659
+ else
3660
+ errcontext ("processing remote data during \"%s\" in transaction %u" ,
3661
+ logicalrep_message_type (errarg -> command ),
3662
+ errarg -> remote_xid );
3679
3663
}
3680
-
3681
- errcontext ("%s" , buf .data );
3682
- pfree (buf .data );
3664
+ else if (errarg -> remote_attnum < 0 )
3665
+ errcontext ("processing remote data during \"%s\" for replication target relation \"%s.%s\" in transaction %u" ,
3666
+ logicalrep_message_type (errarg -> command ),
3667
+ errarg -> rel -> remoterel .nspname ,
3668
+ errarg -> rel -> remoterel .relname ,
3669
+ errarg -> remote_xid );
3670
+ else
3671
+ errcontext ("processing remote data during \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u" ,
3672
+ logicalrep_message_type (errarg -> command ),
3673
+ errarg -> rel -> remoterel .nspname ,
3674
+ errarg -> rel -> remoterel .relname ,
3675
+ errarg -> rel -> remoterel .attnames [errarg -> remote_attnum ],
3676
+ errarg -> remote_xid );
3683
3677
}
3684
3678
3685
3679
/* Set transaction information of apply error callback */
3686
3680
static inline void
3687
- set_apply_error_context_xact (TransactionId xid , TimestampTz ts )
3681
+ set_apply_error_context_xact (TransactionId xid )
3688
3682
{
3689
3683
apply_error_callback_arg .remote_xid = xid ;
3690
- apply_error_callback_arg .ts = ts ;
3691
3684
}
3692
3685
3693
3686
/* Reset all information of apply error callback */
@@ -3697,5 +3690,5 @@ reset_apply_error_context_info(void)
3697
3690
apply_error_callback_arg .command = 0 ;
3698
3691
apply_error_callback_arg .rel = NULL ;
3699
3692
apply_error_callback_arg .remote_attnum = -1 ;
3700
- set_apply_error_context_xact (InvalidTransactionId , 0 );
3693
+ set_apply_error_context_xact (InvalidTransactionId );
3701
3694
}
0 commit comments