schema

Documentation

Index

• Constants
• Variables
• type ColumnType
• type MongoTables
  • func (mts *MongoTables) AddTable(db string, table string) (*Table, error)
  • func (mts *MongoTables) AddTableForMsg(msg *msg.Msg) error
  • func (mts *MongoTables) Close()
  • func (mts *MongoTables) Configure(pipelineName string, configInput map[string]interface{}) error
  • func (mts *MongoTables) DelTable(db string, table string) error
  • func (mts *MongoTables) GetTable(db string, table string) (*Table, error)
  • func (mts *MongoTables) NewSchemaTables(conf *config.BaseConfig, pluginConfig map[string]interface{}, startPos string, ...)
  • func (mts *MongoTables) RefreshTable(db string, table string)
  • func (mts *MongoTables) SaveMeta(data string) error
  • func (mts *MongoTables) UpdateTable(db string, table string, args interface{}, pos string, index int) (err error)
• type MysqlTables
  • func (mts *MysqlTables) AddTable(db string, table string) (*Table, error)
  • func (mts *MysqlTables) AddTableForMsg(msg *msg.Msg) error
  • func (mts *MysqlTables) Close()
  • func (mts *MysqlTables) Configure(pipelineName string, configInput map[string]interface{}) error
  • func (mts *MysqlTables) DelTable(db string, table string) (err error)
  • func (mts *MysqlTables) ExecuteSQL(cmd string, args ...interface{}) (rr *mysql.Result, err error)
  • func (mts *MysqlTables) ExecuteSQLForMemDB(cmd string, args ...interface{}) (rr *mysql.Result, err error)
  • func (mts *MysqlTables) ExecuteSQLForMetaDB(cmd string, args ...interface{}) (rr *mysql.Result, err error)
  • func (mts *MysqlTables) GetColumnTypeFromRawType(rawType string) int
  • func (mts *MysqlTables) GetTable(db string, table string) (*Table, error)
  • func (mts *MysqlTables) GetTableCreateDDL(db string, table string) (string, error)
  • func (mts *MysqlTables) LoadMetaFromDB(rulesMap map[string]interface{}) map[string]interface{}
  • func (mts *MysqlTables) LoadSyncTableMetaFromDB() map[string]interface{}
  • func (mts *MysqlTables) NewSchemaTables(conf *config.BaseConfig, pluginConfig map[string]interface{}, startPos string, ...)
  • func (mts *MysqlTables) RefreshTable(db string, table string)
  • func (mts *MysqlTables) SaveMeta(tablesMeta string) error
  • func (mts *MysqlTables) StartTimerSaveMeta()
  • func (mts *MysqlTables) UpdateTable(db string, table string, ddl interface{}, pos string, index int) (err error)
• type MysqlTablesMeta
• type Table
  • func (t *Table) DelColumn(name string) error
  • func (t *Table) FindColumn(name string) int
  • func (t *Table) GetTableColumnsName() []string
• type TableColumn

Constants

const MongoName = "mongo"

const MysqlName = "mysql"

Variables

var MemDbHost = "localhost"

var MemDbPort = 5166 // default 5166

Types

type ColumnType

type ColumnType = int

const (
	TypeNumber    ColumnType = iota + 1 // tinyint, smallint, int, bigint, year
	TypeMediumInt                       // medium int
	TypeFloat                           // float, double
	TypeEnum                            // enum
	TypeSet                             // set
	TypeString                          // other
	TypeDatetime                        // datetime
	TypeTimestamp                       // timestamp
	TypeDate                            // date
	TypeTime                            // time
	TypeBit                             // bit
	TypeJson                            // json
	TypeDecimal                         // decimal
)

type MongoTables

type MongoTables struct {
	*config.MongoConfig
	// contains filtered or unexported fields
}

func (*MongoTables) AddTable

func (mts *MongoTables) AddTable(db string, table string) (*Table, error)

func (*MongoTables) AddTableForMsg

func (mts *MongoTables) AddTableForMsg(msg *msg.Msg) error

func (*MongoTables) Close

func (mts *MongoTables) Close()

func (*MongoTables) Configure

func (mts *MongoTables) Configure(pipelineName string, configInput map[string]interface{}) error

func (*MongoTables) DelTable

func (mts *MongoTables) DelTable(db string, table string) error

func (*MongoTables) GetTable

func (mts *MongoTables) GetTable(db string, table string) (*Table, error)

func (*MongoTables) NewSchemaTables

func (mts *MongoTables) NewSchemaTables(conf *config.BaseConfig, pluginConfig map[string]interface{}, startPos string, rulesMap map[string]interface{})

func (*MongoTables) RefreshTable

func (mts *MongoTables) RefreshTable(db string, table string)

func (*MongoTables) SaveMeta

func (mts *MongoTables) SaveMeta(data string) error

func (*MongoTables) UpdateTable

func (mts *MongoTables) UpdateTable(db string, table string, args interface{}, pos string, index int) (err error)

type MysqlTables

type MysqlTables struct {
	sync.RWMutex
	*config.MysqlConfig

	*MysqlTablesMeta

	FilePath string
	// contains filtered or unexported fields
}

func (*MysqlTables) AddTable

func (mts *MysqlTables) AddTable(db string, table string) (*Table, error)

func (*MysqlTables) AddTableForMsg

func (mts *MysqlTables) AddTableForMsg(msg *msg.Msg) error

func (*MysqlTables) Close

func (mts *MysqlTables) Close()

func (*MysqlTables) Configure

func (mts *MysqlTables) Configure(pipelineName string, configInput map[string]interface{}) error

func (*MysqlTables) DelTable

func (mts *MysqlTables) DelTable(db string, table string) (err error)

func (*MysqlTables) ExecuteSQL

func (mts *MysqlTables) ExecuteSQL(cmd string, args ...interface{}) (rr *mysql.Result, err error)

func (*MysqlTables) ExecuteSQLForMemDB

func (mts *MysqlTables) ExecuteSQLForMemDB(cmd string, args ...interface{}) (rr *mysql.Result, err error)

func (*MysqlTables) ExecuteSQLForMetaDB

func (mts *MysqlTables) ExecuteSQLForMetaDB(cmd string, args ...interface{}) (rr *mysql.Result, err error)

func (*MysqlTables) GetColumnTypeFromRawType

func (mts *MysqlTables) GetColumnTypeFromRawType(rawType string) int

func (*MysqlTables) GetTable

func (mts *MysqlTables) GetTable(db string, table string) (*Table, error)

func (*MysqlTables) GetTableCreateDDL

func (mts *MysqlTables) GetTableCreateDDL(db string, table string) (string, error)

func (*MysqlTables) LoadMetaFromDB

func (mts *MysqlTables) LoadMetaFromDB(rulesMap map[string]interface{}) map[string]interface{}

func (*MysqlTables) LoadSyncTableMetaFromDB

func (mts *MysqlTables) LoadSyncTableMetaFromDB() map[string]interface{}

func (*MysqlTables) NewSchemaTables

func (mts *MysqlTables) NewSchemaTables(conf *config.BaseConfig, pluginConfig map[string]interface{}, startPos string, rulesMap map[string]interface{})

func (*MysqlTables) RefreshTable

func (mts *MysqlTables) RefreshTable(db string, table string)

func (*MysqlTables) SaveMeta

func (mts *MysqlTables) SaveMeta(tablesMeta string) error

func (*MysqlTables) StartTimerSaveMeta

func (mts *MysqlTables) StartTimerSaveMeta()

func (*MysqlTables) UpdateTable

func (mts *MysqlTables) UpdateTable(db string, table string, ddl interface{}, pos string, index int) (err error)

type MysqlTablesMeta

type MysqlTablesMeta struct {
	// contains filtered or unexported fields
}

type Table

type Table struct {
	Schema  string        `toml:"schema" json:"schema"`
	Name    string        `toml:"name" json:"name"`
	Columns []TableColumn `toml:"columns" json:"columns"`
}

func (*Table) DelColumn

func (t *Table) DelColumn(name string) error

func (*Table) FindColumn

func (t *Table) FindColumn(name string) int

func (*Table) GetTableColumnsName

func (t *Table) GetTableColumnsName() []string

type TableColumn

type TableColumn struct {
	Name    string     `toml:"name" json:"name"`
	Type    ColumnType `toml:"type" json:"type"`
	RawType string     `toml:"raw_type" json:"raw_type"`
}