parquet-info command to browse parquet files (#2230)

* parquet-info command to browse parquet files

* null support in parquet

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)

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
+}

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

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
+}