2020-08-06 20:58:47 +03:00
|
|
|
package cli
|
|
|
|
|
|
|
|
import (
|
2021-02-22 10:37:00 +03:00
|
|
|
"context"
|
2020-08-06 20:58:47 +03:00
|
|
|
|
2023-05-19 17:24:18 +03:00
|
|
|
"github.com/neilotoole/sq/cli/run"
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
"github.com/neilotoole/sq/cli/flag"
|
2020-08-23 13:42:15 +03:00
|
|
|
"github.com/neilotoole/sq/libsq/core/errz"
|
2023-04-30 17:18:56 +03:00
|
|
|
"github.com/neilotoole/sq/libsq/core/lg"
|
2023-04-26 18:16:42 +03:00
|
|
|
"github.com/neilotoole/sq/libsq/core/lg/lga"
|
2020-08-23 13:42:15 +03:00
|
|
|
"github.com/neilotoole/sq/libsq/core/options"
|
2020-08-06 20:58:47 +03:00
|
|
|
"github.com/neilotoole/sq/libsq/driver"
|
|
|
|
"github.com/neilotoole/sq/libsq/source"
|
2023-04-26 18:16:42 +03:00
|
|
|
|
|
|
|
"github.com/spf13/cobra"
|
2020-08-06 20:58:47 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
// determineSources figures out what the active source is
|
|
|
|
// from any combination of stdin, flags or cfg. It will
|
2023-05-19 17:24:18 +03:00
|
|
|
// mutate ru.Config.Collection as necessary. If no error
|
2020-08-06 20:58:47 +03:00
|
|
|
// is returned, it is guaranteed that there's an active
|
2023-04-19 08:28:09 +03:00
|
|
|
// source on the collection.
|
2023-05-19 17:24:18 +03:00
|
|
|
func determineSources(ctx context.Context, ru *run.Run) error {
|
|
|
|
cmd, coll := ru.Cmd, ru.Config.Collection
|
2023-04-19 08:28:09 +03:00
|
|
|
activeSrc, err := activeSrcFromFlagsOrConfig(cmd, coll)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
// Note: ^ activeSrc could still be nil
|
|
|
|
|
|
|
|
// check if there's input on stdin
|
2023-05-19 17:24:18 +03:00
|
|
|
stdinSrc, err := checkStdinSource(ctx, ru)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if stdinSrc != nil {
|
|
|
|
// We have a valid source on stdin.
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
// Add the stdin source to coll.
|
|
|
|
err = coll.Add(stdinSrc)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
if !cmdFlagChanged(cmd, flag.ActiveSrc) {
|
2020-08-06 20:58:47 +03:00
|
|
|
// If the user has not explicitly set an active
|
|
|
|
// source via flag, then we set the stdin pipe data
|
|
|
|
// source as the active source.
|
|
|
|
// We do this because the @stdin src is commonly the
|
|
|
|
// only data source the user cares about in a pipe
|
|
|
|
// situation.
|
2023-04-19 08:28:09 +03:00
|
|
|
_, err = coll.SetActive(stdinSrc.Handle, false)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
activeSrc = stdinSrc
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if activeSrc == nil {
|
|
|
|
return errz.New(msgNoActiveSrc)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// activeSrcFromFlagsOrConfig gets the active source, either
|
|
|
|
// from flagActiveSrc or from srcs.Active. An error is returned
|
|
|
|
// if the flag src is not found: if the flag src is found,
|
2023-04-19 08:28:09 +03:00
|
|
|
// it is set as the active src on coll. If the flag was not
|
|
|
|
// set and there is no active src in coll, (nil, nil) is
|
2020-08-06 20:58:47 +03:00
|
|
|
// returned.
|
2023-04-19 08:28:09 +03:00
|
|
|
func activeSrcFromFlagsOrConfig(cmd *cobra.Command, coll *source.Collection) (*source.Source, error) {
|
2020-08-06 20:58:47 +03:00
|
|
|
var activeSrc *source.Source
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
if cmdFlagChanged(cmd, flag.ActiveSrc) {
|
2020-08-06 20:58:47 +03:00
|
|
|
// The user explicitly wants to set an active source
|
|
|
|
// just for this query.
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
handle, _ := cmd.Flags().GetString(flag.ActiveSrc)
|
|
|
|
s, err := coll.Get(handle)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
2023-04-19 08:28:09 +03:00
|
|
|
return nil, errz.Wrapf(err, "flag --%s", flag.ActiveSrc)
|
2020-08-06 20:58:47 +03:00
|
|
|
}
|
|
|
|
|
2023-04-19 08:28:09 +03:00
|
|
|
activeSrc, err = coll.SetActive(s.Handle, false)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
} else {
|
2023-04-19 08:28:09 +03:00
|
|
|
activeSrc = coll.Active()
|
2020-08-06 20:58:47 +03:00
|
|
|
}
|
|
|
|
return activeSrc, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// checkStdinSource checks if there's stdin data (on pipe/redirect).
|
|
|
|
// If there is, that pipe is inspected, and if it has recognizable
|
|
|
|
// input, a new source instance with handle @stdin is constructed
|
2022-12-18 03:51:33 +03:00
|
|
|
// and returned. If the pipe has no data (size is zero),
|
|
|
|
// then (nil,nil) is returned.
|
2023-05-19 17:24:18 +03:00
|
|
|
func checkStdinSource(ctx context.Context, ru *run.Run) (*source.Source, error) {
|
|
|
|
cmd := ru.Cmd
|
2020-08-06 20:58:47 +03:00
|
|
|
|
2023-05-19 17:24:18 +03:00
|
|
|
f := ru.Stdin
|
2020-08-06 20:58:47 +03:00
|
|
|
info, err := f.Stat()
|
|
|
|
if err != nil {
|
|
|
|
return nil, errz.Wrap(err, "failed to get stat on stdin")
|
|
|
|
}
|
|
|
|
|
|
|
|
if info.Size() <= 0 {
|
|
|
|
// Doesn't make sense to have zero-data pipe? just ignore.
|
2022-12-18 03:51:33 +03:00
|
|
|
return nil, nil //nolint:nilnil
|
2020-08-06 20:58:47 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// If we got this far, we have pipe input
|
|
|
|
|
|
|
|
typ := source.TypeNone
|
2023-05-03 15:36:10 +03:00
|
|
|
if cmd.Flags().Changed(flag.IngestDriver) {
|
|
|
|
val, _ := cmd.Flags().GetString(flag.IngestDriver)
|
2023-04-22 06:36:32 +03:00
|
|
|
typ = source.DriverType(val)
|
2023-05-19 17:24:18 +03:00
|
|
|
if ru.DriverRegistry.ProviderFor(typ) == nil {
|
2020-08-06 20:58:47 +03:00
|
|
|
return nil, errz.Errorf("unknown driver type: %s", typ)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-05-19 17:24:18 +03:00
|
|
|
err = ru.Files.AddStdin(f)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
if typ == source.TypeNone {
|
2023-05-19 17:24:18 +03:00
|
|
|
typ, err = ru.Files.TypeStdin(ctx)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if typ == source.TypeNone {
|
|
|
|
return nil, errz.New("unable to detect type of stdin: use flag --driver")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-05-03 15:36:10 +03:00
|
|
|
return newSource(
|
|
|
|
ctx,
|
2023-05-19 17:24:18 +03:00
|
|
|
ru.DriverRegistry,
|
2023-05-03 15:36:10 +03:00
|
|
|
typ,
|
|
|
|
source.StdinHandle,
|
|
|
|
source.StdinHandle,
|
|
|
|
options.Options{},
|
|
|
|
)
|
2020-08-06 20:58:47 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// newSource creates a new Source instance where the
|
|
|
|
// driver type is known. Opts may be nil.
|
2023-04-30 17:18:56 +03:00
|
|
|
func newSource(ctx context.Context, dp driver.Provider, typ source.DriverType, handle, loc string,
|
2022-12-18 11:35:59 +03:00
|
|
|
opts options.Options,
|
|
|
|
) (*source.Source, error) {
|
2023-05-03 15:36:10 +03:00
|
|
|
log := lg.FromContext(ctx)
|
2023-04-30 17:18:56 +03:00
|
|
|
|
2020-08-06 20:58:47 +03:00
|
|
|
if opts == nil {
|
2023-04-02 22:49:45 +03:00
|
|
|
log.Debug("Create new data source",
|
|
|
|
lga.Handle, handle,
|
|
|
|
lga.Driver, typ,
|
|
|
|
lga.Loc, source.RedactLocation(loc),
|
|
|
|
)
|
2020-08-06 20:58:47 +03:00
|
|
|
} else {
|
2023-04-02 22:49:45 +03:00
|
|
|
log.Debug("Create new data source with opts",
|
|
|
|
lga.Handle, handle,
|
|
|
|
lga.Driver, typ,
|
|
|
|
lga.Loc, source.RedactLocation(loc),
|
2023-04-26 18:16:42 +03:00
|
|
|
// lga.Opts, opts.Encode(), // FIXME: encode opts for debugging
|
2023-04-02 22:49:45 +03:00
|
|
|
)
|
2020-08-06 20:58:47 +03:00
|
|
|
}
|
|
|
|
|
2023-04-16 01:28:51 +03:00
|
|
|
err := source.ValidHandle(handle)
|
2020-08-06 20:58:47 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
drvr, err := dp.DriverFor(typ)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2022-12-25 07:04:18 +03:00
|
|
|
src := &source.Source{Handle: handle, Location: loc, Type: typ, Options: opts}
|
2020-08-06 20:58:47 +03:00
|
|
|
|
2023-04-02 22:49:45 +03:00
|
|
|
log.Debug("Validating provisional new data source", lga.Src, src)
|
2020-08-06 20:58:47 +03:00
|
|
|
canonicalSrc, err := drvr.ValidateSource(src)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return canonicalSrc, nil
|
|
|
|
}
|