Method readTableColumns

driver/mysql/extractor.go:568–614

readTableColumns reads table columns on applier

()

// readTableColumns reads table columns on applier
func (e *Extractor) readTableColumns() (err error) {
	e.logger.Info("Examining table structure on extractor")

	// map parent -> child
	fkParentMap := make(map[common.SchemaTable]map[common.SchemaTable]struct{})

	for _, doDb := range e.replicateDoDb {
		for _, tbCtx := range doDb.TableMap {
			doTb := tbCtx.Table
			tableColumns, fkParentTables, err := base.GetTableColumnsSqle(e.sqleContext, doTb.TableSchema, doTb.TableName)
			if err != nil {
				return err
			}
			doTb.OriginalTableColumns = tableColumns
			doTb.ColumnMap, err = mysqlconfig.BuildColumnMapIndex(doTb.ColumnMapFrom, doTb.OriginalTableColumns.Ordinals)
			if err != nil {
				return err
			}

			childST := common.SchemaTable{Schema: doTb.TableSchema, Table: doTb.TableName}
			for _, fkpt := range fkParentTables {
				schema := g.StringElse(fkpt.Schema.String(), doTb.TableSchema)
				parentST := common.SchemaTable{Schema: schema, Table: fkpt.Name.String()}
				if m, ok := fkParentMap[parentST]; ok {
					m[childST] = struct{}{}
				} else {
					fkParentMap[parentST] = map[common.SchemaTable]struct{}{
						childST: {},
					}
				}
			}
		}
	}

	for _, db := range e.replicateDoDb {
		for _, tbCtx := range db.TableMap {
			tb := tbCtx.Table
			st := common.SchemaTable{Schema: tb.TableSchema, Table: tb.TableName}
			if m, ok := fkParentMap[st]; ok {
				tbCtx.FKChildren = m
				e.logger.Info("fk parent", "len", len(m), "schema", tb.TableSchema, "table", tb.TableName)
			}
		}
	}

	return nil
}

func (e *Extractor) initNatsPubClient(natsAddr string) (err error) {
	e.logger.Debug("begin Connect nats server", "NatAddr", natsAddr)