reader

Documentation

Index

• type Batch
  • func NewBatch(mgr *BatchMgr, bufferSize int, clientReq chan base.ClientRequest, ...) *Batch
  • func (b *Batch) Add(data base.Data)
  • func (b *Batch) Done()
  • func (b *Batch) IsFull() bool
  • func (b *Batch) SendErrorData(d base.Data, err error)
• type BatchMgr
  • func NewBatchMgr(schema *config.Schema, batchSize int, ...) *BatchMgr
  • func (bm *BatchMgr) Add(data base.Data, runnerLogger *logger.RunnerLogger) error
  • func (bm *BatchMgr) Done()
  • func (bm *BatchMgr) GeneratePropsString(props []*config.Prop) string
  • func (bm *BatchMgr) InitSchema(header base.Record, runnerLogger *logger.RunnerLogger) (err error)
  • func (m *BatchMgr) MakeEdgeStmt(batch []base.Data) (string, error)
  • func (m *BatchMgr) MakeVertexStmt(batch []base.Data) (string, error)
• type DataFileReader
• type FileReader
  • func New(fileIdx int, file *config.File, cleanup bool, ...) (*FileReader, error)
  • func (r *FileReader) Read() (numErrorLines int64, err error)
  • func (r *FileReader) Stop()

Types

type Batch

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

func NewBatch

func NewBatch(mgr *BatchMgr, bufferSize int, clientReq chan base.ClientRequest, errCh chan<- base.ErrData) *Batch

func (*Batch) Add

func (b *Batch) Add(data base.Data)

func (*Batch) Done

func (b *Batch) Done()

func (*Batch) IsFull

func (b *Batch) IsFull() bool

func (*Batch) SendErrorData

func (b *Batch) SendErrorData(d base.Data, err error)

type BatchMgr

type BatchMgr struct {
	Schema           *config.Schema
	Batches          []*Batch
	InsertStmtPrefix string
	// contains filtered or unexported fields
}

func NewBatchMgr

func NewBatchMgr(schema *config.Schema, batchSize int, clientRequestChs []chan base.ClientRequest, errCh chan<- base.ErrData) *BatchMgr

func (*BatchMgr) Add

func (bm *BatchMgr) Add(data base.Data, runnerLogger *logger.RunnerLogger) error

func (*BatchMgr) Done

func (bm *BatchMgr) Done()

func (*BatchMgr) GeneratePropsString

func (bm *BatchMgr) GeneratePropsString(props []*config.Prop) string

func (*BatchMgr) InitSchema

func (bm *BatchMgr) InitSchema(header base.Record, runnerLogger *logger.RunnerLogger) (err error)

func (*BatchMgr) MakeEdgeStmt

func (m *BatchMgr) MakeEdgeStmt(batch []base.Data) (string, error)

func (*BatchMgr) MakeVertexStmt

func (m *BatchMgr) MakeVertexStmt(batch []base.Data) (string, error)

type DataFileReader

type DataFileReader interface {
	InitReader(*os.File, *logger.RunnerLogger)
	ReadLine() (base.Data, error)
	TotalBytes() (int64, error)
}

type FileReader

type FileReader struct {
	FileIdx int
	File    *config.File

	WithHeader  bool
	DataReader  DataFileReader
	Concurrency int
	BatchMgr    *BatchMgr
	StopFlag    bool
	// contains filtered or unexported fields
}

FIXME: private fields

func New

func New(fileIdx int, file *config.File, cleanup bool, clientRequestChs []chan base.ClientRequest,
	errCh chan<- base.ErrData, runnerLogger *logger.RunnerLogger) (*FileReader, error)

func (*FileReader) Read

func (r *FileReader) Read() (numErrorLines int64, err error)

func (*FileReader) Stop

func (r *FileReader) Stop()