diff --git a/arrow.go b/arrow.go index 3e268d08f..4b416351e 100644 --- a/arrow.go +++ b/arrow.go @@ -158,8 +158,13 @@ func (st *BasicTable) Release() { func (st *BasicTable) Get(column, row int) interface{} { field := st.Schema().Field(column) c, i := st.resolver.Resolve(row) + nullable := field.Nullable chunk := st.Column(column).Data().Chunk(c) + // TODO(twg) 2023/01/26 potential NULL support? + if nullable && chunk.IsNull(i) { + return nil + } switch field.Type.(type) { case *arrow.BooleanType: return chunk.(*array.Boolean).Value(i) diff --git a/cmd/parquet-info.go b/cmd/parquet-info.go new file mode 100644 index 000000000..746b1ec21 --- /dev/null +++ b/cmd/parquet-info.go @@ -0,0 +1,33 @@ +// Copyright 2022 Molecula Corp. (DBA FeatureBase). +// SPDX-License-Identifier: Apache-2.0 +package cmd + +import ( + "fmt" + + "github.com/featurebasedb/featurebase/v3/ctl" + "github.com/featurebasedb/featurebase/v3/logger" + "github.com/spf13/cobra" +) + +func newParquetInfoCommand(logdest logger.Logger) *cobra.Command { + c := ctl.NewParquetInfoCommand(logdest) + cmd := &cobra.Command{ + Use: "parquet-info PATH|URL", + Short: "Inspect Parquet Files.", + Long: ` +Displays schema and sample data from the specified file +`, + Args: func(cmd *cobra.Command, args []string) error { + if len(args) == 0 { + return fmt.Errorf("data directory path required") + } else if len(args) > 1 { + return fmt.Errorf("too many command line arguments") + } + c.Path = args[0] + return nil + }, + RunE: usageErrorWrapper(c), + } + return cmd +} diff --git a/cmd/root.go b/cmd/root.go index c29f50f05..22f562867 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -108,6 +108,7 @@ at https://docs.featurebase.com/. rc.AddCommand(newDAXCommand(stderr)) rc.AddCommand(newDataframeCsvLoaderCommand(logdest)) rc.AddCommand(newPreSortCommand(logdest)) + rc.AddCommand(newParquetInfoCommand(logdest)) rc.SetOutput(stderr) return rc diff --git a/ctl/parquet-info.go b/ctl/parquet-info.go new file mode 100644 index 000000000..6e9125920 --- /dev/null +++ b/ctl/parquet-info.go @@ -0,0 +1,119 @@ +// Copyright 2022 Molecula Corp. (DBA FeatureBase). +// SPDX-License-Identifier: Apache-2.0 +package ctl + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + + "github.com/apache/arrow/go/v10/arrow/memory" + "github.com/apache/arrow/go/v10/parquet/file" + "github.com/apache/arrow/go/v10/parquet/pqarrow" + pilosa "github.com/featurebasedb/featurebase/v3" + "github.com/featurebasedb/featurebase/v3/logger" +) + +// ParquetInfoCommand represents a command for displaying info about a parquet file +type ParquetInfoCommand struct { + // Filepath or URL to the parquet file. + Path string + + // Standard input/output + stdout io.Writer + logDest logger.Logger +} + +// NewParquetInfoCommand returns a new instance of ParquetInfoCommand. +func NewParquetInfoCommand(logdest logger.Logger) *ParquetInfoCommand { + return &ParquetInfoCommand{ + stdout: os.Stdout, + logDest: logdest, + } +} + +// Run displays schema and samples data from a parquet file +func (cmd *ParquetInfoCommand) Run(ctx context.Context) error { + // Open database. + var f *os.File + _, err := url.ParseRequestURI(cmd.Path) + if err == nil { // treat as a URL + response, err := http.Get(cmd.Path) + if err != nil { + return err + } + if response.StatusCode != 200 { + return errors.New(fmt.Sprintf("unexpected response %d", response.StatusCode)) + } + defer response.Body.Close() + // download to temp file first + f, err = os.CreateTemp("", "BulkParquetFile.parquet") + if err != nil { + return errors.New(fmt.Sprintf("error creating tempfile %v", err)) + } + _, err = io.Copy(f, response.Body) + if err != nil { + return errors.New(fmt.Sprintf("error downloading url %v %v", cmd.Path, err)) + } + defer os.Remove(f.Name()) + + _, err = f.Seek(0, io.SeekStart) + if err != nil { + return errors.New(fmt.Sprintf("error reseting file for reading %v ", err)) + } + } else { + f, err = os.Open(cmd.Path) + if err != nil { + return err + } + } + + pf, err := file.NewParquetReader(f) + if err != nil { + return err + } + mem := memory.NewGoAllocator() + reader, err := pqarrow.NewFileReader(pf, pqarrow.ArrowReadProperties{}, mem) + if err != nil { + return err + } + table, err := reader.ReadTable(ctx) + if err != nil { + return err + } + // print file name + fmt.Printf("\n\nName:%v\n", cmd.Path) + // print schema + schema := table.Schema() + fields := schema.Fields() + for i, field := range fields { + fmt.Printf("%v. Name: %v\n", i, field.Name) + fmt.Printf("%v. Type: %v\n", i, field.Type) + fmt.Printf("%v. Nullable: %v\n\n", i, field.Nullable) + } + bt := pilosa.BasicTableFromArrow(table, mem) + // print num rows + numRows := int(bt.NumRows()) + fmt.Printf("Number of rows:%v\n", numRows) + if numRows > 10 { + numRows = 10 + } + fmt.Println("Sample:") + // print at most 10 sample rows in table format + for _, field := range fields { + fmt.Printf("%v\t", field.Name) + } + fmt.Println("") + for i := 0; i < numRows; i++ { + for j := 0; j < len(fields); j++ { + fmt.Printf("%v\t", bt.Get(j, i)) + } + fmt.Println("") + } + + return nil +}