processMessage processes a logical replication message as appropriate. A couple important aspects: 1. Relation messages describe tables being replicated and are used to build a type map for decoding tuples 2. INSERT/UPDATE/DELETE messages describe changes to rows that must be applied to the replica.
( xld pglogrepl.XLogData, state *replicationState, )
| 682 | // |
| 683 | // Returns a boolean true if the message was a commit that should be acknowledged, and an error if one occurred. |
| 684 | func (r *LogicalReplicator) processMessage( |
| 685 | xld pglogrepl.XLogData, |
| 686 | state *replicationState, |
| 687 | ) (bool, error) { |
| 688 | walData := xld.WALData |
| 689 | logicalMsg, err := pglogrepl.ParseV2(walData, state.inStream) |
| 690 | if err != nil { |
| 691 | return false, err |
| 692 | } |
| 693 | |
| 694 | r.logger.Debugf("XLogData (%T) => WALStart %s ServerWALEnd %s ServerTime %s", logicalMsg, xld.WALStart, xld.ServerWALEnd, xld.ServerTime) |
| 695 | |
| 696 | // Update the last received LSN |
| 697 | if xld.ServerWALEnd > state.lastReceivedLSN { |
| 698 | state.lastReceivedLSN = xld.ServerWALEnd |
| 699 | } |
| 700 | |
| 701 | switch logicalMsg := logicalMsg.(type) { |
| 702 | case *pglogrepl.RelationMessageV2: |
| 703 | _, exists := state.relations[logicalMsg.RelationID] |
| 704 | if exists { |
| 705 | // This means schema changes have occurred, so we need to |
| 706 | // commit any buffered ongoing batch transactions. |
| 707 | err := r.commitOngoingTxn(state, delta.DDLStmtFlushReason) |
| 708 | if err != nil { |
| 709 | return false, err |
| 710 | } |
| 711 | } |
| 712 | |
| 713 | state.relations[logicalMsg.RelationID] = logicalMsg |
| 714 | |
| 715 | schema := make(sql.Schema, len(logicalMsg.Columns)) |
| 716 | var keys []uint16 |
| 717 | for i, col := range logicalMsg.Columns { |
| 718 | pgType, err := pgtypes.NewPostgresType(state.typeMap, col.DataType, col.TypeModifier) |
| 719 | if err != nil { |
| 720 | return false, err |
| 721 | } |
| 722 | schema[i] = &sql.Column{ |
| 723 | Name: col.Name, |
| 724 | Type: pgType, |
| 725 | PrimaryKey: col.Flags == 1, |
| 726 | } |
| 727 | if col.Flags == 1 { |
| 728 | keys = append(keys, uint16(i)) |
| 729 | } |
| 730 | } |
| 731 | state.schemas[logicalMsg.RelationID] = schema |
| 732 | state.keys[logicalMsg.RelationID] = keys |
| 733 | |
| 734 | // Create the table if it doesn't exist |
| 735 | if ddl, err := generateCreateTableStmt(logicalMsg); err != nil { |
| 736 | return false, err |
| 737 | } else if _, err := adapter.ExecCatalog(state.replicaCtx, ddl); err != nil { |
| 738 | return false, err |
| 739 | } |
| 740 | |
| 741 | case *pglogrepl.BeginMessage: |
no test coverage detected