src

package
v1.8.4-release Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2022 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

普通模式同步 update 转成 insert on update insert 转成 replace into delete 转成 delete 只要是同一条数据，只要有遍历过，后面遍历出来的数据，则不再进行操作

Index

Constants

View Source 
const BIFROST_VERION = "v1.8.3"
View Source 
const VERSION = "v1.8.4"

Variables

This section is empty.

Functions

func AllTypeToInt64 

func AllTypeToInt64(s interface{}) (int64, error)

func AllTypeToUInt64 

func AllTypeToUInt64(s interface{}) (uint64, error)

func CkDataTypeTransfer 

func CkDataTypeTransfer(data interface{}, fieldName string, toDataType string, NullNotTransferDefault bool) (v interface{}, e error)

func NewConn 

func NewConn() pluginDriver.Driver

func ReplaceBr 

func ReplaceBr(str string) string

func ReplaceTwoReplace 

func ReplaceTwoReplace(sql string) string

去除连续的两个空格

func TransferComma2Other 

func TransferComma2Other(sql string) string

将sql 里 (,) 和 单引号，双引号 里包括的 逗号 先替换成 #@% 感谢 @zeroone2005 正则表达式提供支持

func TransferNotes2Space 

func TransferNotes2Space(sql string) string

func TransferOther2Comma 

func TransferOther2Comma(str string) string

将 #@% 再替换回原来的 逗号

Types

type AlterColumnInfo 

type AlterColumnInfo struct {
	AfterName string
	Default   *string
	Nullable  bool
	Comment   string
	// contains filtered or unexported fields
}

type AlterSQL 

type AlterSQL struct {
	DefaultSchemaName string
	Sql               string
	// contains filtered or unexported fields
}

func NewAlterSQL 

func NewAlterSQL(DefaultSchemaName, sql string, c *Conn) *AlterSQL

func (*AlterSQL) AddColumn 

func (This *AlterSQL) AddColumn(sql string) (destAlterSql string)

mysql : ADD COLUMN `f1` VARCHAR(200) NULL AFTER `number`, ck : add column column_name [type] [default_expr] [after name_after]

func (*AlterSQL) ChangeColumn 

func (This *AlterSQL) ChangeColumn(sql string) (destAlterSql string)

mysql : CHANGE `number` `number` BIGINT(20) NOT NULL COMMENT '馆藏数量', ck : modify column column_name [type] [default_expr]

func (*AlterSQL) DropColumn 

func (This *AlterSQL) DropColumn(sql string) (destAlterSql string)

func (*AlterSQL) GetColumnInfo 

func (This *AlterSQL) GetColumnInfo(pArr []string) *AlterColumnInfo

func (*AlterSQL) GetTransferCkType 

func (This *AlterSQL) GetTransferCkType(mysqlColumnType string) (ckType string)

func (*AlterSQL) ModifyColumn 

func (This *AlterSQL) ModifyColumn(sql string) (destAlterSql string)

mysql : MODIFY column `number` BIGINT(20) NOT NULL COMMENT '馆藏数量', ck : modify column column_name [type] [default_expr]

func (*AlterSQL) Transfer2CkSQL 

func (This *AlterSQL) Transfer2CkSQL(c *Conn) (SchemaName, TableName, destAlterSql, destLocalAlterSql, destDisAlterSql, destViewAlterSql string)

type ClickhouseDB 

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

func NewClickHouseDBConn 

func NewClickHouseDBConn(uri string) *ClickhouseDB

func (*ClickhouseDB) Close 

func (This *ClickhouseDB) Close() bool

func (*ClickhouseDB) Exec 

func (This *ClickhouseDB) Exec(sql string, value []driver.Value) error

func (*ClickhouseDB) GetConn 

func (This *ClickhouseDB) GetConn() clickhouse.Clickhouse

func (*ClickhouseDB) GetSchemaList 

func (This *ClickhouseDB) GetSchemaList() (data []string)

func (*ClickhouseDB) GetSchemaTableList 

func (This *ClickhouseDB) GetSchemaTableList(schema string) (data []string)

func (*ClickhouseDB) GetTableDataList 

func (This *ClickhouseDB) GetTableDataList(schema string, table string, where string) (data []map[string]driver.Value)

func (*ClickhouseDB) GetTableFields 

func (This *ClickhouseDB) GetTableFields(SchemaName, TableName string) (data []ckFieldStruct)

func (*ClickhouseDB) GetVersion 

func (This *ClickhouseDB) GetVersion() (Version string)

func (*ClickhouseDB) Open 

func (This *ClickhouseDB) Open() bool

type Conn 

type Conn struct {
	pluginDriver.PluginDriverInterface
	// contains filtered or unexported fields
}

func (*Conn) AutoCommit 

func (This *Conn) AutoCommit() (LastSuccessCommitData *pluginDriver.PluginDataType, ErrData *pluginDriver.PluginDataType, e error)

合并数据，提交到 ck里

func (*Conn) AutoCreateTableCommit 

func (This *Conn) AutoCreateTableCommit(list []*pluginDriver.PluginDataType, n int) (errData *pluginDriver.PluginDataType)

自动创建表的提交

func (*Conn) CheckDataSkip 

func (This *Conn) CheckDataSkip(data *pluginDriver.PluginDataType) bool

func (*Conn) CheckTableExsit 

func (This *Conn) CheckTableExsit(SchemaName, TableName string) (bool, error)

func (*Conn) CheckUri 

func (This *Conn) CheckUri() error

func (*Conn) Close 

func (This *Conn) Close() bool

func (*Conn) Commit 

func (This *Conn) Commit(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) CommitLogMod_Append added in v1.2.2 

func (This *Conn) CommitLogMod_Append(list []*pluginDriver.PluginDataType, n int) (errData *pluginDriver.PluginDataType)

func (*Conn) CommitNormal added in v1.2.2 

func (This *Conn) CommitNormal(list []*pluginDriver.PluginDataType, n int) (errData *pluginDriver.PluginDataType)

func (*Conn) Connect 

func (This *Conn) Connect() bool

func (*Conn) CreateCkDatabase 

func (This *Conn) CreateCkDatabase(SchemaName string) (err error)

通过数据自动创建 ck 库

func (*Conn) CreateCkTable 

func (This *Conn) CreateCkTable(data *pluginDriver.PluginDataType) (ckField []fieldStruct, err error)

通过数据自动创建 ck 表

func (*Conn) Del 

func (This *Conn) Del(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) GetFieldName 

func (This *Conn) GetFieldName(Name string) string

func (*Conn) GetParam 

func (This *Conn) GetParam(p interface{}) (*PluginParam, error)

func (*Conn) GetSchemaName 

func (This *Conn) GetSchemaName(Name string) (SchemaName string)

func (*Conn) GetUriExample 

func (This *Conn) GetUriExample() string

func (*Conn) InitVersion 

func (This *Conn) InitVersion()

func (*Conn) InitVersion0 

func (This *Conn) InitVersion0(versionStr string) int

func (*Conn) Insert 

func (This *Conn) Insert(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) NotCreateTableCommit 

func (This *Conn) NotCreateTableCommit(list []*pluginDriver.PluginDataType, n int) (errData *pluginDriver.PluginDataType)

非自动创建表的提交

func (*Conn) Open 

func (This *Conn) Open() error

func (*Conn) Query 

func (This *Conn) Query(data *pluginDriver.PluginDataType, retry bool) (LastSuccessCommitData *pluginDriver.PluginDataType, ErrData *pluginDriver.PluginDataType, err error)

func (*Conn) ReConnect 

func (This *Conn) ReConnect() bool

func (*Conn) SetOption 

func (This *Conn) SetOption(uri *string, param map[string]interface{})

func (*Conn) SetParam 

func (This *Conn) SetParam(p interface{}) (interface{}, error)

func (*Conn) Skip 

func (This *Conn) Skip(SkipData *pluginDriver.PluginDataType) error

设置跳过的位点

func (*Conn) TimeOutCommit 

func (This *Conn) TimeOutCommit() (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

func (*Conn) TranferQuerySql 

func (This *Conn) TranferQuerySql(data *pluginDriver.PluginDataType) (SchemaName, TableName, newSql, newLocalSql, newDisSql, newViewSql string)

func (*Conn) TransferToCkTypeByColumnData 

func (This *Conn) TransferToCkTypeByColumnData(v interface{}, nullable bool) (toType string)

func (*Conn) TransferToCkTypeByColumnType 

func (This *Conn) TransferToCkTypeByColumnType(columnType string, nullable bool) (toType string)

func (*Conn) TransferToCreateDatabaseSql 

func (This *Conn) TransferToCreateDatabaseSql(SchemaName string) (sql string)

func (*Conn) TransferToCreateTableSql 

func (This *Conn) TransferToCreateTableSql(data *pluginDriver.PluginDataType) (sql string, distributeSql, viewSql string, ckField []fieldStruct)

func (*Conn) Update 

func (This *Conn) Update(data *pluginDriver.PluginDataType, retry bool) (*pluginDriver.PluginDataType, *pluginDriver.PluginDataType, error)

type DDLSupportType 

type DDLSupportType struct {
	ColumnAdd      bool
	ColumnModify   bool
	ColumnDrop     bool
	TableRename    bool
	DropDbAndTable bool
	Rruncate       bool
}

type DropDBTableOrTableSQL 

type DropDBTableOrTableSQL struct {
	DefaultSchemaName string
	Sql               string
	// contains filtered or unexported fields
}

func NewDropDBOrTableSQL 

func NewDropDBOrTableSQL(DefaultSchemaName, sql string, c *Conn) *DropDBTableOrTableSQL

func (*DropDBTableOrTableSQL) Transfer2CkSQL 

func (This *DropDBTableOrTableSQL) Transfer2CkSQL(c *Conn) (SchemaName, TableName, newSql, newLocalSql, newDisSql, newViewSql string)

type PluginClickHouseController 

type PluginClickHouseController struct {
	controller.CommonController
}

func (*PluginClickHouseController) GetClickHouseSchemaList 

func (c *PluginClickHouseController) GetClickHouseSchemaList()

func (*PluginClickHouseController) GetClickHouseSchemaTableList 

func (c *PluginClickHouseController) GetClickHouseSchemaTableList()

func (*PluginClickHouseController) GetClickHouseTableFields 

func (c *PluginClickHouseController) GetClickHouseTableFields()

type PluginParam 

type PluginParam struct {
	Field                  []fieldStruct
	BatchSize              int
	CkSchema               string
	CkTable                string
	PriKey                 []fieldStruct
	SyncType               SyncType
	AutoCreateTable        bool
	NullNotTransferDefault bool //是否将null值强制转成相对应类型的默认值 , false 将 null 转成相对就的 0 或者 "" , true 不进行转换，为了兼容老版本，才反过来的
	BifrostMustBeSuccess   bool // bifrost server 保留,数据是否能丢
	LowerCaseTableNames    int8 // 0 源字段怎么样，就怎么样，1 转成小写，2 全部转成大写; 只对自动建表的功能有效
	//ModifDDLMap            map[string]bool //ddl同步程度选择
	ModifDDLType  *DDLSupportType //ddl同步程度选择
	CkEngine      int
	CkClusterName string

	Data *TableDataStruct

	SkipBinlogData *pluginDriver.PluginDataType // 在执行 skip 的时候 ，进行传入进来的时候需要要过滤的 位点，在每次commit之后，这个数据会被清空
	// contains filtered or unexported fields
}

type PluginParam0 

type PluginParam0 struct {
	Field            []fieldStruct
	CkSchema         string
	CkTable          string
	CkSchemaAndTable string
}

type ReNameSQL 

type ReNameSQL struct {
	DefaultSchemaName string
	Sql               string
	// contains filtered or unexported fields
}

func NewReNameSQL 

func NewReNameSQL(DefaultSchemaName, sql string, c *Conn) *ReNameSQL

func (*ReNameSQL) Transfer2CkSQL 

func (This *ReNameSQL) Transfer2CkSQL(c *Conn) (SchemaName, TableName, destAlterSql, destAlterViewSql, destAlterDisSql string)

type SyncType 

type SyncType string
const (
	SYNCMODE_NORMAL     SyncType = "Normal"
	SYNCMODE_LOG_UPDATE SyncType = "LogUpdate"
	SYNCMODE_LOG_APPEND SyncType = "insertAll"
)

type TableDataStruct 

type TableDataStruct struct {
	Data       []*pluginDriver.PluginDataType
	CommitData []*pluginDriver.PluginDataType // commit 提交的数据列表，Data 每 BatchSize 数据量划分为一个最后提交的commit
}

func NewTableData 

func NewTableData() *TableDataStruct

type TruncateSQL 

type TruncateSQL struct {
	DefaultSchemaName string
	Sql               string
	// contains filtered or unexported fields
}

func NewTruncateSQL 

func NewTruncateSQL(DefaultSchemaName, sql string, c *Conn) *TruncateSQL

func (*TruncateSQL) Transfer2CkSQL 

func (This *TruncateSQL) Transfer2CkSQL(c *Conn) (SchemaName, TableName, newSql, newLocalSql, newDisSql, newViewSql string)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.