Since the data is sent to the server in the form of a byte stream, we use a Unix pipe to stream the data to DuckDB.
(ctx *sql.Context, insert *plan.InsertInto, dst sql.InsertableTable, load *plan.LoadData)
| 65 | // Since the data is sent to the server in the form of a byte stream, |
| 66 | // we use a Unix pipe to stream the data to DuckDB. |
| 67 | func (db *DuckBuilder) buildClientSideLoadData(ctx *sql.Context, insert *plan.InsertInto, dst sql.InsertableTable, load *plan.LoadData) (sql.RowIter, error) { |
| 68 | _, localInfile, ok := sql.SystemVariables.GetGlobal("local_infile") |
| 69 | if !ok { |
| 70 | return nil, fmt.Errorf("error: local_infile variable was not found") |
| 71 | } |
| 72 | |
| 73 | if localInfile.(int8) == 0 { |
| 74 | return nil, fmt.Errorf("local_infile needs to be set to 1 to use LOCAL") |
| 75 | } |
| 76 | |
| 77 | reader, err := ctx.LoadInfile(load.File) |
| 78 | if err != nil { |
| 79 | return nil, err |
| 80 | } |
| 81 | defer reader.Close() |
| 82 | |
| 83 | pipePath, err := db.CreatePipe(ctx, "load-data") |
| 84 | if err != nil { |
| 85 | return nil, err |
| 86 | } |
| 87 | defer os.Remove(pipePath) |
| 88 | |
| 89 | // Write the data to the FIFO pipe. |
| 90 | go func() { |
| 91 | pipe, err := os.OpenFile(pipePath, os.O_WRONLY, os.ModeNamedPipe) |
| 92 | if err != nil { |
| 93 | return |
| 94 | } |
| 95 | defer pipe.Close() |
| 96 | io.Copy(pipe, reader) |
| 97 | }() |
| 98 | |
| 99 | return db.executeLoadData(ctx, insert, dst, load, pipePath) |
| 100 | } |
| 101 | |
| 102 | // In the non-local case, we can directly use the file path to read the data. |
| 103 | func (db *DuckBuilder) buildServerSideLoadData(ctx *sql.Context, insert *plan.InsertInto, dst sql.InsertableTable, load *plan.LoadData) (sql.RowIter, error) { |
no test coverage detected