buildSQL builds the DuckDB COPY FROM statement.
()
| 191 | _ = pipe.Close() |
| 192 | loader.ready <- copyErr |
| 193 | go func() { |
| 194 | loader.Wait() |
| 195 | loader.releaseOwnedLease() |
| 196 | }() |
| 197 | return |
| 198 | } |
| 199 | // Abort may race the writer opening the FIFO. Do not publish a pipe |
| 200 | // to a loader that has already been canceled. |
| 201 | if loader.aborted.Load() { |
| 202 | _ = pipe.Close() |
| 203 | loader.ready <- ErrCopyAborted |
| 204 | go func() { |
| 205 | loader.Wait() |
| 206 | loader.releaseOwnedLease() |
| 207 | }() |
| 208 | return |
| 209 | } |
| 210 | |
| 211 | loader.pipe.Store(pipe) |
| 212 | if loader.aborted.Load() { |
| 213 | if loader.pipe.CompareAndSwap(pipe, nil) { |
| 214 | _ = pipe.Close() |
| 215 | } |
| 216 | loader.ready <- ErrCopyAborted |
| 217 | go func() { |
| 218 | loader.Wait() |
| 219 | loader.releaseOwnedLease() |
| 220 | }() |
| 221 | } |
| 222 | }() |
| 223 | }) |
| 224 | return loader.ready |
| 225 | } |
| 226 | |
| 227 | func (loader *PipeDataLoader) LoadChunk(ctx *sql.Context, data []byte) error { |
| 228 | if ctx == nil { |
| 229 | ctx = loader.ctx |
| 230 | } |
| 231 | if err := loader.validateBinding(ctx); err != nil { |
| 232 | return err |
| 233 | } |
| 234 | if copyErr := loader.currentError(); copyErr != nil { |
| 235 | return fmt.Errorf("COPY operation has been aborted: %w", copyErr) |
| 236 | } |
| 237 | pipe := loader.pipe.Load() |
| 238 | if pipe == nil { |
| 239 | return errors.New("COPY data loader is not ready") |
| 240 | } |
| 241 | if loader.logger != nil { |
| 242 | loader.logger.Tracef("Copying %d bytes to pipe %s", len(data), loader.pipePath) |
| 243 | } |
| 244 | // Write the data to the FIFO pipe. |
| 245 | _, err := pipe.Write(data) |
| 246 | if err != nil { |
| 247 | if loader.logger != nil { |
| 248 | loader.logger.Error("Copying data to pipe failed:", err) |
| 249 | } |
| 250 | loader.Abort(ctx) |
no test coverage detected