Function NewCsvDataLoader

pgserver/dataloader.go:156–190

(
	ctx *sql.Context, handler *DuckHandler,
	schema string, table sql.InsertableTable, columns tree.NameList, options *tree.CopyOptions,
	rawOptions string, // For non-PG-parsable COPY FROM, unused for now
)

var _ DataLoader = (*CsvDataLoader)(nil)

func NewCsvDataLoader(
	ctx *sql.Context, handler *DuckHandler,
	schema string, table sql.InsertableTable, columns tree.NameList, options *tree.CopyOptions,
	rawOptions string, // For non-PG-parsable COPY FROM, unused for now
) (DataLoader, error) {
	// Create the FIFO pipe
	duckBuilder := handler.e.Analyzer.ExecBuilder.PriorityBuilder.(*backend.DuckBuilder)
	pipePath, err := duckBuilder.CreatePipe(ctx, "pg-copy-from")
	if err != nil {
		return nil, err
	}

	// Create a child without mutating the parent context into its own ancestor.
	ctx, cancel := newCopyContext(ctx)

	loader := &CsvDataLoader{
		PipeDataLoader: PipeDataLoader{
			ctx:      ctx,
			cancel:   cancel,
			schema:   schema,
			table:    table,
			columns:  columns,
			pipePath: pipePath,
			rowCount: make(chan int64, 1),
			logger:   ctx.GetLogger(),
		},
		options: options,
	}
	loader.read = func() {
		loader.executeCopy(loader.buildSQL(), pipePath)
	}

	return loader, nil
}

func newCopyContext(ctx *sql.Context) (*sql.Context, context.CancelFunc) {
	return ctx.NewSubContext()
}