| 1034 | } |
| 1035 | |
| 1036 | func (r *LogicalReplicator) append(state *replicationState, relationID uint32, tuple []*pglogrepl.TupleDataColumn, actionType, eventType binlog.RowEventType, onlyKeys bool) error { |
| 1037 | rel, ok := state.relations[relationID] |
| 1038 | if !ok { |
| 1039 | return fmt.Errorf("unknown relation ID %d", relationID) |
| 1040 | } |
| 1041 | appender, err := state.deltas.GetDeltaAppender(rel.Namespace, rel.RelationName, state.schemas[relationID]) |
| 1042 | if err != nil { |
| 1043 | return err |
| 1044 | } |
| 1045 | |
| 1046 | if len(tuple) == 0 { |
| 1047 | // The only case where we can have an empty tuple is when |
| 1048 | // we're deleting+inserting a row and the key columns are unchanged. |
| 1049 | if eventType == binlog.UpdateRowEvent && actionType == binlog.DeleteRowEvent { |
| 1050 | appender.ObserveEvents(binlog.UpdateRowEvent, 1) |
| 1051 | return nil |
| 1052 | } |
| 1053 | return fmt.Errorf("empty tuple data") |
| 1054 | } |
| 1055 | |
| 1056 | fields := appender.Fields() |
| 1057 | actions := appender.Action() |
| 1058 | txnTags := appender.TxnTag() |
| 1059 | txnServers := appender.TxnServer() |
| 1060 | txnGroups := appender.TxnGroup() |
| 1061 | txnSeqNumbers := appender.TxnSeqNumber() |
| 1062 | txnStmtOrdinals := appender.TxnStmtOrdinal() |
| 1063 | |
| 1064 | actions.Append(int8(actionType)) |
| 1065 | txnTags.AppendNull() |
| 1066 | txnServers.Append([]byte("")) |
| 1067 | txnGroups.AppendNull() |
| 1068 | txnSeqNumbers.Append(uint64(state.currentTransactionLSN)) |
| 1069 | txnStmtOrdinals.Append(state.inTxnStmtID) |
| 1070 | |
| 1071 | size := 0 |
| 1072 | idx := 0 |
| 1073 | |
| 1074 | for i, metadata := range rel.Columns { |
| 1075 | builder := fields[i] |
| 1076 | var col *pglogrepl.TupleDataColumn |
| 1077 | if onlyKeys { |
| 1078 | if metadata.Flags != 1 { // not a key column |
| 1079 | builder.AppendNull() |
| 1080 | continue |
| 1081 | } |
| 1082 | col = tuple[idx] |
| 1083 | idx++ |
| 1084 | } else { |
| 1085 | col = tuple[i] |
| 1086 | } |
| 1087 | switch col.DataType { |
| 1088 | case pglogrepl.TupleDataTypeNull: |
| 1089 | builder.AppendNull() |
| 1090 | case pglogrepl.TupleDataTypeText, pglogrepl.TupleDataTypeBinary: |
| 1091 | length, err := decodeToArrow(state.typeMap, metadata, col.Data, tupleDataFormat(col.DataType), builder) |
| 1092 | if err != nil { |
| 1093 | return err |