source

package
v0.1.8 Latest Latest
Warning

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

 Go to latest Published: Sep 19, 2024 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GenerateJSONFile 

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

func SlimCondition added in v0.1.8 

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

func SplitCondition added in v0.1.8 

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

func SplitConditionAccordingMaxGoRoutine added in v0.1.8 

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

func SplitConditionAccordingToTimeSplitKey added in v0.1.8 

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

func SplitTimeConditionsByMaxThread added in v0.1.8 

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

Types

type DatabendIngesterStatsData added in v0.1.3 

type DatabendIngesterStatsData struct {
	BytesPerSecond float64
	RowsPerSecondd float64
}

type DatabendSourceStatsRecorder added in v0.1.3 

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

func NewDatabendIntesterStatsRecorder added in v0.1.3 

func NewDatabendIntesterStatsRecorder() *DatabendSourceStatsRecorder

func (*DatabendSourceStatsRecorder) RecordMetric added in v0.1.3 

func (stats *DatabendSourceStatsRecorder) RecordMetric(rows int)

func (*DatabendSourceStatsRecorder) Stats added in v0.1.3 

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

type MysqlSource added in v0.1.8 

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

func NewMysqlSource added in v0.1.8 

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

func (*MysqlSource) AdjustBatchSizeAccordingToSourceDbTable added in v0.1.8 

func (s *MysqlSource) AdjustBatchSizeAccordingToSourceDbTable() int64

AdjustBatchSizeAccordingToSourceDbTable has a concept called s, s = (maxKey - minKey) / sourceTableRowCount if s == 1 it means the data is uniform in the table, if s is much bigger than 1, it means the data is not uniform in the table

func (*MysqlSource) DeleteAfterSync added in v0.1.8 

func (s *MysqlSource) DeleteAfterSync() error

func (*MysqlSource) GetAllSourceReadRowsCount added in v0.1.8 

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

func (*MysqlSource) GetDatabasesAccordingToSourceDbRegex added in v0.1.8 

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

func (*MysqlSource) GetDbTablesAccordingToSourceDbTables added in v0.1.8 

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

func (*MysqlSource) GetMinMaxSplitKey added in v0.1.8 

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

func (*MysqlSource) GetMinMaxTimeSplitKey added in v0.1.8 

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

func (*MysqlSource) GetSourceReadRowsCount added in v0.1.8 

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

func (*MysqlSource) GetTablesAccordingToSourceTableRegex added in v0.1.8 

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

func (*MysqlSource) QueryTableData added in v0.1.8 

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

type PostgresSource added in v0.1.8 

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

func NewPostgresSource added in v0.1.8 

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

func (PostgresSource) AdjustBatchSizeAccordingToSourceDbTable added in v0.1.8 

func (p PostgresSource) AdjustBatchSizeAccordingToSourceDbTable() int64

func (PostgresSource) DeleteAfterSync added in v0.1.8 

func (p PostgresSource) DeleteAfterSync() error

func (PostgresSource) GetAllSourceReadRowsCount added in v0.1.8 

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

func (PostgresSource) GetDatabasesAccordingToSourceDbRegex added in v0.1.8 

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

func (PostgresSource) GetDbTablesAccordingToSourceDbTables added in v0.1.8 

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

func (PostgresSource) GetMinMaxSplitKey added in v0.1.8 

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

func (PostgresSource) GetMinMaxTimeSplitKey added in v0.1.8 

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

func (PostgresSource) GetSourceReadRowsCount added in v0.1.8 

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

func (PostgresSource) GetTablesAccordingToSourceTableRegex added in v0.1.8 

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

func (PostgresSource) QueryTableData added in v0.1.8 

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

func (*PostgresSource) SwitchDatabase added in v0.1.8 

func (p *PostgresSource) SwitchDatabase() error

type Sourcer 

type Sourcer interface {
	AdjustBatchSizeAccordingToSourceDbTable() int64
	GetSourceReadRowsCount() (int, error)
	GetMinMaxSplitKey() (int64, int64, error)
	GetMinMaxTimeSplitKey() (string, string, error)
	DeleteAfterSync() error
	QueryTableData(threadNum int, conditionSql string) ([][]interface{}, []string, error)
	GetDatabasesAccordingToSourceDbRegex(sourceDatabasePattern string) ([]string, error)
	GetTablesAccordingToSourceTableRegex(sourceTablePattern string, databases []string) (map[string][]string, error)
	GetAllSourceReadRowsCount() (int, error)
	GetDbTablesAccordingToSourceDbTables() (map[string][]string, error)
}

func NewSource 

func NewSource(cfg *config.Config) (Sourcer, error)

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.