source

Sep 5, 2025

Changes in this version

+ type NullUint64 struct

+ Uint64 uint64

+ Valid bool

+ func (n *NullUint64) Scan(value interface{}) error

+ func (n NullUint64) Value() (driver.Value, error)

Jan 23, 2025

Changes in this version

+ type SQLServerSource struct

+ func NewSqlServerSource(cfg *config.Config) (*SQLServerSource, error)

+ func (s *SQLServerSource) AdjustBatchSizeAccordingToSourceDbTable() int64

+ func (s *SQLServerSource) DeleteAfterSync() error

+ func (s *SQLServerSource) GetAllSourceReadRowsCount() (int, error)

+ func (s *SQLServerSource) GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)

+ func (s *SQLServerSource) GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)

+ func (s *SQLServerSource) GetMinMaxSplitKey() (int64, int64, error)

+ func (s *SQLServerSource) GetMinMaxTimeSplitKey() (string, string, error)

+ func (s *SQLServerSource) GetSourceReadRowsCount() (int, error)

+ func (s *SQLServerSource) GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)

+ func (s *SQLServerSource) QueryTableData(threadNum int, conditionSql string) ([][]interface{}, []string, error)

Oct 18, 2024

Changes in this version

+ type OracleSource struct

+ func NewOracleSource(cfg *config.Config) (*OracleSource, error)

+ func (p *OracleSource) AdjustBatchSizeAccordingToSourceDbTable() int64

+ func (p *OracleSource) DeleteAfterSync() error

+ func (p *OracleSource) GetAllSourceReadRowsCount() (int, error)

+ func (p *OracleSource) GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)

+ func (p *OracleSource) GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)

+ func (p *OracleSource) GetMinMaxSplitKey() (int64, int64, error)

+ func (p *OracleSource) GetMinMaxTimeSplitKey() (string, string, error)

+ func (p *OracleSource) GetSourceReadRowsCount() (int, error)

+ func (p *OracleSource) GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)

+ func (p *OracleSource) QueryTableData(threadNum int, conditionSql string) ([][]interface{}, []string, error)

+ func (p *OracleSource) SwitchDatabase() error

Sep 19, 2024

Changes in this version

+ func SlimCondition(maxThread int, minSplitKey, maxSplitKey int64) [][]int64

+ func SplitCondition(sourceSplitKey string, batchSize, minSplitKey, maxSplitKey int64) []string

+ func SplitConditionAccordingMaxGoRoutine(sourceSplitKey string, batchSize, minSplitKey, maxSplitKey, allMax int64) <-chan string

+ func SplitConditionAccordingToTimeSplitKey(cfg *config.Config, minTimeSplitKey, maxTimeSplitKey string) ([]string, error)

+ func SplitTimeConditionsByMaxThread(conditions []string, maxThread int) [][]string

+ type MysqlSource struct

+ func NewMysqlSource(cfg *config.Config) (*MysqlSource, error)

+ func (s *MysqlSource) AdjustBatchSizeAccordingToSourceDbTable() int64

+ func (s *MysqlSource) DeleteAfterSync() error

+ func (s *MysqlSource) GetAllSourceReadRowsCount() (int, error)

+ func (s *MysqlSource) GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)

+ func (s *MysqlSource) GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)

+ func (s *MysqlSource) GetMinMaxSplitKey() (int64, int64, error)

+ func (s *MysqlSource) GetMinMaxTimeSplitKey() (string, string, error)

+ func (s *MysqlSource) GetSourceReadRowsCount() (int, error)

+ func (s *MysqlSource) GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)

+ func (s *MysqlSource) QueryTableData(threadNum int, conditionSql string) ([][]interface{}, []string, error)

+ type PostgresSource struct

+ func NewPostgresSource(cfg *config.Config) (*PostgresSource, error)

+ func (p *PostgresSource) SwitchDatabase() error

+ func (p PostgresSource) AdjustBatchSizeAccordingToSourceDbTable() int64

+ func (p PostgresSource) DeleteAfterSync() error

+ func (p PostgresSource) GetAllSourceReadRowsCount() (int, error)

+ func (p PostgresSource) GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)

+ func (p PostgresSource) GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)

+ func (p PostgresSource) GetMinMaxSplitKey() (int64, int64, error)

+ func (p PostgresSource) GetMinMaxTimeSplitKey() (string, string, error)

+ func (p PostgresSource) GetSourceReadRowsCount() (int, error)

+ func (p PostgresSource) GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)

+ func (p PostgresSource) QueryTableData(threadNum int, conditionSql string) ([][]interface{}, []string, error)

type Sourcer

+ AdjustBatchSizeAccordingToSourceDbTable func() int64

+ DeleteAfterSync func() error

+ GetAllSourceReadRowsCount func() (int, error)

+ GetDatabasesAccordingToSourceDbRegex func(sourceDatabasePattern string) ([]string, error)

+ GetDbTablesAccordingToSourceDbTables func() (map[string][]string, error)

+ GetMinMaxSplitKey func() (int64, int64, error)

+ GetMinMaxTimeSplitKey func() (string, string, error)

+ GetSourceReadRowsCount func() (int, error)

+ GetTablesAccordingToSourceTableRegex func(sourceTablePattern string, databases []string) (map[string][]string, error)

Sep 4, 2024

Changes in this version

type Source

+ func (s *Source) AdjustBatchSizeAccordingToSourceDbTable() int64

Aug 30, 2024

Changes in this version

type Source

+ func (s *Source) GetAllSourceReadRowsCount() (int, error)

+ func (s *Source) GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)

+ func (s *Source) GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)

+ func (s *Source) GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)

Aug 9, 2024

Changes in this version

+ type DatabendIngesterStatsData struct

+ BytesPerSecond float64

+ RowsPerSecondd float64

+ type DatabendSourceStatsRecorder struct

+ func NewDatabendIntesterStatsRecorder() *DatabendSourceStatsRecorder

+ func (stats *DatabendSourceStatsRecorder) RecordMetric(rows int)

+ func (stats *DatabendSourceStatsRecorder) Stats(statsWindow time.Duration) DatabendIngesterStatsData

Jul 4, 2024

Changes in this version

type Source

+ func (s *Source) GetMinMaxSplitKey() (int, int, error)

+ func (s *Source) GetMinMaxTimeSplitKey() (string, string, error)

+ func (s *Source) GetRowsCountByConditionSql(conditionSql string) (int, error)

+ func (s *Source) SplitConditionAccordingToTimeSplitKey(minTimeSplitKey, maxTimeSplitKey string) ([]string, error)

+ func (s *Source) SplitTimeConditionsByMaxThread(conditions []string, maxThread int) [][]string

Jun 28, 2024

Changes in this version

type Source

+ func (s *Source) SlimCondition(minSplitKey, maxSplitKey int) [][]int

+ func (s *Source) SplitConditionAccordingMaxGoRoutine(minSplitKey, maxSplitKey, allMax int) []string

Jun 26, 2024

Changes in this version

+ func GenerateJSONFile(columns []string, data [][]interface{}) (string, int, error)

+ type Source struct

+ func NewSource(cfg *config.Config) (*Source, error)

+ func (s *Source) DeleteAfterSync() error

+ func (s *Source) GerMinMaxSplitKey() (int, int, error)

+ func (s *Source) GetBatchNum(rows *sql.Rows) (int, int)

+ func (s *Source) GetRows(conditionSql string) (*sql.Rows, error)

+ func (s *Source) GetSourceReadRowsCount() (int, error)

+ func (s *Source) QueryTableData(rows *sql.Rows) ([][]interface{}, []string, error)

+ func (s *Source) SplitCondition(minSplitKey, maxSplitKey int) []string

+ type Sourcer interface

+ GetBatchNum func(rows *sql.Rows) (int, int)

+ GetRows func(conditionSql string) (*sql.Rows, error)

+ QueryTableData func(rows *sql.Rows) ([][]interface{}, []string, error)