From 003e4ebb51f1c78df074437832b6220da0201568 Mon Sep 17 00:00:00 2001 From: Marcus Breese Date: Tue, 28 Jul 2026 00:16:57 -0400 Subject: [PATCH] Add Parquet support to view, less, and a new parquet2tab Adds a parquet2tab subcommand (mirroring csv2tab) and teaches view and less to read Parquet files directly. Parquet is detected from its PAR1 magic bytes, in the same spirit as the existing gzip sniffing, so no flag is needed for a .parquet file. An explicit --parquet flag is also available, mirroring --csv. To make this work without duplicating the formatting stack, the readers now sit behind a RecordReader interface. The viewer, pager, and CSV exporter only ever needed ReadLine/Close/header access, so they consume the interface instead of *DelimitedTextFile. The exporter and sorter reach further into the text internals and were left on the concrete type. DelimitedTextFile keeps its exported Header field, so those two are untouched. Type mapping is the substance of the change. parquet.Value.String() renders the physical type only -- it would print a DATE as a day count and a DECIMAL(10,2) as its raw unscaled integer -- so ParquetFile builds a per-column formatter from the logical type instead: decimals get their scale applied, dates/times/timestamps are decoded from their integer representations, unsigned ints don't wrap negative, and UUID and the legacy INT96 timestamps are decoded properly. Schema handling: - nested structs flatten to dotted column names (addr.city) - lists and maps render as JSON in a single cell - NULL cells are empty by default, configurable with --na=STRING Parquet keeps its schema in a footer at the end of the file, so reading one requires random access. It cannot be read from a pipe; "-" is rejected with an explicit message rather than failing obscurely. Flags that are meaningless for Parquet (--no-header, --header-comment, --show-comments, --csv) are rejected when the user actually sets them. Note that parquet-go pulls mattn/go-runewidth from v0.0.2 to v0.0.15, which underpins termbox's width calculations. Pager output was diffed before and after, including CJK text, and is byte-identical. Co-Authored-By: Claude Opus 5 --- README.md | 20 ++ cmd/less.go | 13 +- cmd/parquet2tab.go | 52 +++ cmd/root.go | 69 ++++ cmd/view.go | 13 +- examples/iris.parquet | Bin 0 -> 7865 bytes go.mod | 10 +- go.sum | 25 +- textfile/csv_to_tab.go | 14 +- textfile/pager.go | 44 ++- textfile/parquet.go | 669 +++++++++++++++++++++++++++++++++++++++ textfile/parquet_test.go | 212 +++++++++++++ textfile/reader.go | 27 ++ textfile/textfile.go | 29 +- textfile/viewer.go | 30 +- 15 files changed, 1163 insertions(+), 64 deletions(-) create mode 100644 cmd/parquet2tab.go create mode 100644 examples/iris.parquet create mode 100644 textfile/parquet.go create mode 100644 textfile/parquet_test.go create mode 100644 textfile/reader.go diff --git a/README.md b/README.md index 76f1464..84ac5a2 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,26 @@ For more information, see here: [https://compgen.io/tabl](https://compgen.io/tab Note: The `tabl less` pager forces 256-color output to keep formatting/colors consistent on terminals like tmux/screen (especially on RHEL8). +## Parquet + +`view`, `less`, and `parquet2tab` can read Parquet files. Parquet is detected +automatically from the file's magic bytes, so no flag is needed: + + tabl view data.parquet + tabl less data.parquet + tabl parquet2tab data.parquet > data.txt + +Use `--parquet` to force it for a file that isn't named/detected as one. + +The parquet schema supplies the header. Nested structs are flattened into dotted +column names (`addr.city`), while lists and maps are rendered as JSON in a single +cell. NULL cells are written as an empty value; use `--na=STRING` to change that +(e.g. `--na=NA`). + +Because a parquet file's schema lives in a footer at the end of the file, +reading one requires random access -- parquet cannot be read from a pipe, so a +real file path is required. + ## Examples ![Demo](https://github.com/compgen-io/tabl-docs/raw/master/assets/img/tabl-demo-2.gif) diff --git a/cmd/less.go b/cmd/less.go index 4f67547..437d209 100644 --- a/cmd/less.go +++ b/cmd/less.go @@ -13,6 +13,8 @@ func init() { lessCmd.Flags().BoolVar(&NoHeader, "no-header", false, "File has no header") lessCmd.Flags().BoolVar(&HeaderComment, "header-comment", false, "The header is the last commented line") lessCmd.Flags().BoolVar(&IsCSV, "csv", false, "The file is a CSV file") + lessCmd.Flags().BoolVar(&IsParquet, "parquet", false, "The file is a Parquet file") + lessCmd.Flags().StringVar(&NAString, "na", "", "Value shown for NULL cells (Parquet)") lessCmd.Flags().IntVar(&MinWidth, "min", 0, "Minimum column width") lessCmd.Flags().IntVar(&MaxWidth, "max", 0, "Maximum column width") rootCmd.AddCommand(lessCmd) @@ -34,16 +36,11 @@ var lessCmd = &cobra.Command{ if len(args) == 0 { args = []string{"-"} } - var txt *textfile.DelimitedTextFile - if !IsCSV { - txt = textfile.NewTabFile(args[0]) - } else { - txt = textfile.NewCSVFile(args[0]) + txt, err := openReader(cmd, args[0]) + if err != nil { + er(err) } - txt = txt.WithNoHeader(NoHeader). - WithHeaderComment(HeaderComment) - textfile.NewTextPager(txt). WithShowLineNum(ShowLineNum). WithMaxWidth(MaxWidth). diff --git a/cmd/parquet2tab.go b/cmd/parquet2tab.go new file mode 100644 index 0000000..aa5d489 --- /dev/null +++ b/cmd/parquet2tab.go @@ -0,0 +1,52 @@ +package cmd + +import ( + "fmt" + "os" + + "github.com/compgen-io/tabl/textfile" + "github.com/spf13/cobra" +) + +func init() { + parquet2TabCmd.Flags().StringVar(&NAString, "na", "", "Value to write for NULL cells") + rootCmd.AddCommand(parquet2TabCmd) +} + +var parquet2TabCmd = &cobra.Command{ + Use: "parquet2tab [file]", + Short: "Convert a Parquet file to tab-delimited format", + Long: `Convert a Parquet file to tab-delimited format. + +The parquet schema becomes the header. Nested structs are flattened into +dotted column names (addr.city), and lists and maps are written as JSON. + +NULL cells are written as an empty value unless --na is given. + +Note that parquet files can't be read from a pipe -- the schema lives in a +footer at the end of the file, so a real file path is required. +`, + Args: func(cmd *cobra.Command, args []string) error { + if len(args) == 0 { + return fmt.Errorf("Missing [file]") + } + if args[0] == "-" { + return textfile.ErrParquetStdin + } + _, err := os.Stat(args[0]) + if os.IsNotExist(err) { + return fmt.Errorf("Missing file: %s", args[0]) + } + return nil + }, + Run: func(cmd *cobra.Command, args []string) { + pq := textfile.NewParquetFile(args[0]).WithNAString(NAString) + if err := pq.Open(); err != nil { + er(err) + } + + if err := textfile.NewCSVExporter(pq).WriteFile(os.Stdout); err != nil { + er(err) + } + }, +} diff --git a/cmd/root.go b/cmd/root.go index 6cfc4bd..0d6e780 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -1,15 +1,24 @@ package cmd import ( + "bytes" "fmt" + "io" "os" + "github.com/compgen-io/tabl/textfile" "github.com/spf13/cobra" ) // IsCSV -- the file is a CSV file var IsCSV bool +// IsParquet -- the file is a Parquet file +var IsParquet bool + +// NAString -- the value written for NULL cells (Parquet only) +var NAString string + // NoHeader -- the file has no header var NoHeader bool @@ -45,3 +54,63 @@ func er(msg interface{}) { fmt.Println("Error:", msg) os.Exit(1) } + +// parquetMagic is the marker written at the start (and end) of a parquet file +var parquetMagic = []byte("PAR1") + +// looksLikeParquet sniffs the leading magic bytes, in the same spirit as the +// gzip detection the delimited reader already does. +func looksLikeParquet(fname string) bool { + if fname == "-" { + return false + } + + f, err := os.Open(fname) + if err != nil { + return false + } + defer f.Close() + + buf := make([]byte, len(parquetMagic)) + if _, err := io.ReadFull(f, buf); err != nil { + return false + } + return bytes.Equal(buf, parquetMagic) +} + +// openReader builds the appropriate reader for a file. --parquet and --csv +// force the format; otherwise we sniff for parquet and fall back to text. +func openReader(cmd *cobra.Command, fname string) (textfile.RecordReader, error) { + if IsParquet || (!IsCSV && looksLikeParquet(fname)) { + if err := rejectTextOnlyFlags(cmd); err != nil { + return nil, err + } + pq := textfile.NewParquetFile(fname).WithNAString(NAString) + // open now so a bad path or corrupt file is reported before we start + // writing output + if err := pq.Open(); err != nil { + return nil, err + } + return pq, nil + } + + var txt *textfile.DelimitedTextFile + if IsCSV { + txt = textfile.NewCSVFile(fname) + } else { + txt = textfile.NewTabFile(fname) + } + + return txt.WithNoHeader(NoHeader).WithHeaderComment(HeaderComment), nil +} + +// rejectTextOnlyFlags errors on flags that mean nothing for parquet input. We +// check Changed so that only flags the user actually typed complain. +func rejectTextOnlyFlags(cmd *cobra.Command) error { + for _, name := range []string{"csv", "no-header", "header-comment", "show-comments"} { + if f := cmd.Flags().Lookup(name); f != nil && f.Changed { + return fmt.Errorf("--%s cannot be used with parquet input", name) + } + } + return nil +} diff --git a/cmd/view.go b/cmd/view.go index 12c7a9e..eb8316b 100644 --- a/cmd/view.go +++ b/cmd/view.go @@ -12,6 +12,8 @@ func init() { viewCmd.Flags().BoolVarP(&ShowComments, "show-comments", "H", false, "Show comments") viewCmd.Flags().BoolVarP(&ShowLineNum, "show-linenum", "L", false, "Show line number") viewCmd.Flags().BoolVar(&IsCSV, "csv", false, "The file is a CSV file") + viewCmd.Flags().BoolVar(&IsParquet, "parquet", false, "The file is a Parquet file") + viewCmd.Flags().StringVar(&NAString, "na", "", "Value shown for NULL cells (Parquet)") viewCmd.Flags().BoolVar(&HeaderComment, "header-comment", false, "The header is the last commented line") viewCmd.Flags().BoolVar(&NoHeader, "no-header", false, "File has no header") viewCmd.Flags().IntVar(&MinWidth, "min", 0, "Minimum column width") @@ -35,16 +37,11 @@ var viewCmd = &cobra.Command{ if len(args) == 0 { args = []string{"-"} } - var txt *textfile.DelimitedTextFile - if !IsCSV { - txt = textfile.NewTabFile(args[0]) - } else { - txt = textfile.NewCSVFile(args[0]) + txt, err := openReader(cmd, args[0]) + if err != nil { + er(err) } - txt = txt.WithNoHeader(NoHeader). - WithHeaderComment(HeaderComment) - textfile.NewTextViewer(txt). WithShowComments(ShowComments). WithShowLineNum(ShowLineNum). diff --git a/examples/iris.parquet b/examples/iris.parquet new file mode 100644 index 0000000000000000000000000000000000000000..808e179a88e5eea8a9e6d260af0d87eb5c79cf98 GIT binary patch literal 7865 zcmeI1L2OiI5XZmWF0gEu((d-X?rWE_2O)X@QB1rL`#d2D2jYnk5>rYko0PV83tqW$ zGlnR7;-HBK5>FbMc)=S{2njL5jbOsjlt8cqQhmEKzxnpPZQ-oxX6fuVGv9yyGxPO( zQ})=0AHNf94L)zQ-7nK&iGX{pCQv{d9o%C;UAd4#H%Y#q;gE{p~_LKlkCT@s=Tm! ze!_c1_t(@987khW>cdQ*ydOBizoveX;fbB~sM^O2U-GLNe_4~4kBgt+Q6Kh!OB}x+ zyfi=Z9=80X@8pdnPtFT`BCChs)-7*|N8QG?U!410mbb*K$M{nx=^ynW-iY-df7w5- zc;kvwv-2Um$Yb32$v)0QUH6kG`ysIT=E(o8i-s@jL5;)x zwo~n8J#PahCBJ_4-(h^DuBDzh_)8pkv)|{1pWqBCF4ETl9I3a@gZjz?4 z7cDQT({{^?eA0T)7krUh6$eQj`m`R~6nBfpbrxUnbLJ;``2N9D{I)1gPWKlqKX{0L z!T5j&PTH?%aiyLGUBkQ2coQES>P;W02X=UXi=H+9>*H=SUf{r6WXX7eo6QqF`_uM< zom+`d-{}wM%GVJ;aD~T!;?Rd&Pe0&Io@xHutxsv(t#NiuJn)dz6aQ>q(9@@^gcte2 zk9{HwF>CNy+kN)Sjmv+Od+#&Bdp3cT_)jNf|ElQ)_l2#qcGgJY``r9x{dhufo)|vy zh)*1FvTM<=C=PtqCWwb*Ke*_T*pc|Lh9~^cZ?r~_B#z*&W$NJT0xt1oKXWnj9`oX$ zo=0!TU-p;GF8uv-Li`3}iO0QN$^4!%S2CaD&s@v%%=3{v{ai0`nb&z<;m^DXfB*av z9G-(aHIJ-6e5ng_GCcirQtCk*>Q>4;x2VIQ-e=^`a|c}J8U9WgQGR}&l6a$9U+R^e z>pYIsiMf;clKF91`FOkZ&(As1M?XiV`+y&P0iSugs`nuCd^Qi}AwU00e((!5fARqz z9Ps(O3V-(dI!a#5mGqmu@B?Sq&u0&B-Tt-f|E2kRQ|fS2?}OLL`+ASu-N1kOHTtam zzWRGy>GRzU;^2Q%@52Wh#Pfcy&JVxk3{G}UU3&E!@fH1_J^8(w&rQ4coNE%*(pS^Hvwhk2?7}L=_j+`n$dgi)sV%bqD_g9JnZSuH#pfE)T^~ttbiZj=RkSG4s$m z=9LLEkEP6lxZE}iPyus2ECzQP2@4n1!+VY3YE?h49`v8D9QL2DpU}@iA)5WD&`iGc z3ra31?{T#)Q?rw;)*g4Xp1=jA;9@RtwQ_5EVRH79+3C5NrL(5(7wu=w(_PB)!t~NQ zuA&v!@LE2qM<@5XdQgs%x5f-yVhEEw?lkrm4;{~xIn{(D&tz~e2YKhhs|!IG-a8i@ zb8WyT&?cC*En?CT2r%r37+fV{j?C&!tdvT54!NWGwp_3J!o)4F>OhH zDLV87?sgmQQ5r5)qi(z_hYfGc@RCch3-3`I@7wt_y{_tZ!(Bg-z;RYMlFKQ$wKm+t zHe7h;T)Q#lu$X*uQMva1Ompe%`BVF+=Fh#oFuC~Y`RS#7GxJ>^8=bHG=I0s*_P_hy d{sV8GI^Uc<-FPcHnH-(&w>y*c=~(im^}jjJ==1;p literal 0 HcmV?d00001 diff --git a/go.mod b/go.mod index 62defbf..de4d211 100644 --- a/go.mod +++ b/go.mod @@ -5,12 +5,20 @@ go 1.22 require ( github.com/gizak/termui/v3 v3.1.0 github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d + github.com/parquet-go/parquet-go v0.24.0 github.com/spf13/cobra v1.0.0 ) require ( + github.com/andybalholm/brotli v1.1.0 // indirect + github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.0.0 // indirect - github.com/mattn/go-runewidth v0.0.2 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/mattn/go-runewidth v0.0.15 // indirect github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 // indirect + github.com/olekukonko/tablewriter v0.0.5 // indirect + github.com/pierrec/lz4/v4 v4.1.21 // indirect + github.com/rivo/uniseg v0.4.7 // indirect github.com/spf13/pflag v1.0.3 // indirect + golang.org/x/sys v0.21.0 // indirect ) diff --git a/go.sum b/go.sum index e014b22..d42e067 100644 --- a/go.sum +++ b/go.sum @@ -3,6 +3,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M= +github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY= github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= @@ -34,25 +36,33 @@ github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk= github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= +github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM= +github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ= -github.com/mattn/go-runewidth v0.0.2 h1:UnlwIPBGaTZfPQ6T1IGzPI0EkYAQmT9fAEJ/poFC63o= github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= +github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI= +github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U= +github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 h1:DpOJ2HYzCv8LZP15IdmG+YdwD2luVPHITV96TkirNBM= @@ -62,7 +72,13 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d h1:x3S6kxmy49zXVVyhcnrFqxvNVCBPb2KZ9hV2RBdS840= github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d/go.mod h1:IuKpRQcYE1Tfu+oAQqaLisqDeXgjyyltCfsaoYN18NQ= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= +github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec= +github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY= +github.com/parquet-go/parquet-go v0.24.0 h1:VrsifmLPDnas8zpoHmYiWDZ1YHzLmc7NmNwPGkI2JM4= +github.com/parquet-go/parquet-go v0.24.0/go.mod h1:OqBBRGBl7+llplCvDMql8dEKaDqjaFA/VAPw+OJiNiw= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= +github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= +github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= @@ -74,6 +90,9 @@ github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU= +github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= +github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= +github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= @@ -116,6 +135,8 @@ golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -125,6 +146,8 @@ google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9Ywl google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/textfile/csv_to_tab.go b/textfile/csv_to_tab.go index c1b669c..3629954 100644 --- a/textfile/csv_to_tab.go +++ b/textfile/csv_to_tab.go @@ -9,12 +9,12 @@ import ( // CSVExporter is used to export specific columns from a tab delimited file type CSVExporter struct { - txt *DelimitedTextFile + txt RecordReader showComments bool } // NewCSVExporter - create a new text exporter -func NewCSVExporter(f *DelimitedTextFile) *CSVExporter { +func NewCSVExporter(f RecordReader) *CSVExporter { return &CSVExporter{ txt: f, showComments: false, @@ -36,6 +36,10 @@ func (tex *CSVExporter) WriteFile(out io.Writer) error { for err == nil { line, err = tex.txt.ReadLine() if err != nil { + if err != io.EOF { + tex.txt.Close() + return err + } break } @@ -48,7 +52,7 @@ func (tex *CSVExporter) WriteFile(out io.Writer) error { } if !wroteHeader { - if !tex.txt.noHeader { + if !tex.txt.NoHeader() { err := tex.writeHeader(out) if err != nil { return err @@ -68,10 +72,10 @@ func (tex *CSVExporter) WriteFile(out io.Writer) error { } func (tex *CSVExporter) writeHeader(out io.Writer) error { - if tex.txt.noHeader { + if tex.txt.NoHeader() { return nil } - for i, v := range tex.txt.Header { + for i, v := range tex.txt.GetHeader() { if i > 0 { fmt.Fprint(out, "\t") } diff --git a/textfile/pager.go b/textfile/pager.go index 6928b4d..9549bc9 100644 --- a/textfile/pager.go +++ b/textfile/pager.go @@ -16,7 +16,7 @@ const maxLines = 20000 // TextPager is a viewer for tab-delimited data, it handles formatting and showing the data on a stream type TextPager struct { - txt *DelimitedTextFile + txt RecordReader showComments bool showLineNum bool minWidth int @@ -36,7 +36,7 @@ type TextPager struct { } // NewTextPager - create a new text viewer -func NewTextPager(f *DelimitedTextFile) *TextPager { +func NewTextPager(f RecordReader) *TextPager { return &TextPager{ txt: f, showComments: false, @@ -98,14 +98,14 @@ func (tv *TextPager) load() { // first, let's add the header (if missing) if tv.colNames == nil { - tv.colNames = make([]string, len(tv.txt.Header)) - copy(tv.colNames, tv.txt.Header) + tv.colNames = make([]string, len(tv.txt.GetHeader())) + copy(tv.colNames, tv.txt.GetHeader()) - tv.colWidth = make([]int, len(tv.txt.Header)) - tv.colSticky = make([]bool, len(tv.txt.Header)) + tv.colWidth = make([]int, len(tv.txt.GetHeader())) + tv.colSticky = make([]bool, len(tv.txt.GetHeader())) - for j := 0; j < len(tv.txt.Header); j++ { - r := []rune(tv.txt.Header[j] + " ") + for j := 0; j < len(tv.txt.GetHeader()); j++ { + r := []rune(tv.txt.GetHeader()[j] + " ") tv.colWidth[j] = support.MaxInt(tv.minWidth, tv.colWidth[j], len(r)) if tv.maxWidth > 0 { tv.colWidth[j] = support.MinInt(tv.colWidth[j], tv.maxWidth) @@ -113,20 +113,20 @@ func (tv *TextPager) load() { } } - if len(tv.colNames) < len(tv.txt.Header) { - tv.colNames = make([]string, len(tv.txt.Header)) - copy(tv.colNames, tv.txt.Header) + if len(tv.colNames) < len(tv.txt.GetHeader()) { + tv.colNames = make([]string, len(tv.txt.GetHeader())) + copy(tv.colNames, tv.txt.GetHeader()) - newWidths := make([]int, len(tv.txt.Header)) + newWidths := make([]int, len(tv.txt.GetHeader())) copy(newWidths, tv.colWidth) tv.colWidth = newWidths - newSticky := make([]bool, len(tv.txt.Header)) + newSticky := make([]bool, len(tv.txt.GetHeader())) copy(newSticky, tv.colSticky) tv.colSticky = newSticky - for j := 0; j < len(tv.txt.Header); j++ { - r := []rune(tv.txt.Header[j] + " ") + for j := 0; j < len(tv.txt.GetHeader()); j++ { + r := []rune(tv.txt.GetHeader()[j] + " ") tv.colWidth[j] = support.MaxInt(tv.minWidth, tv.colWidth[j], len(r)) if tv.maxWidth > 0 { tv.colWidth[j] = support.MinInt(tv.colWidth[j], tv.maxWidth) @@ -321,7 +321,7 @@ ESC to hide help text break } } - if !found && el.Next() == nil && !tv.txt.isEOF { + if !found && el.Next() == nil && !tv.txt.IsEOF() { l, err := tv.txt.ReadLine() if err != nil { break @@ -587,7 +587,7 @@ ESC to hide help text e := tv.topRow i := 0 for i = 0; e.Next() != nil && i < tv.visibleRows-3; i++ { - if e.Next() == nil && !tv.txt.isEOF { + if e.Next() == nil && !tv.txt.IsEOF() { // need to load more lines! // qfmt.Fprintln(os.Stderr, "Loading more lines") l, err := tv.txt.ReadLine() @@ -754,7 +754,7 @@ ESC to hide help text break } } - if !found && el.Next() == nil && !tv.txt.isEOF { + if !found && el.Next() == nil && !tv.txt.IsEOF() { l, err := tv.txt.ReadLine() if err != nil { break @@ -943,7 +943,7 @@ func (tv *TextPager) updateTable(tbl *widgets.Table) { tbl.RowStyles[i] = defaultStyle } - if e.Next() == nil && !tv.txt.isEOF { + if e.Next() == nil && !tv.txt.IsEOF() { // need to load more lines! // qfmt.Fprintln(os.Stderr, "Loading more lines") l, err := tv.txt.ReadLine() @@ -978,11 +978,7 @@ func (tv *TextPager) saveToFile(fname string) error { return err } - if tv.txt.headerComment { - f.WriteString(tv.txt.lastComment) - } else if tv.txt.rawHeaderLine != "" { - f.WriteString(tv.txt.rawHeaderLine) - } + f.WriteString(tv.txt.HeaderLine()) e := tv.topRow for i := 0; e.Next() != nil; i++ { diff --git a/textfile/parquet.go b/textfile/parquet.go new file mode 100644 index 0000000..6448a59 --- /dev/null +++ b/textfile/parquet.go @@ -0,0 +1,669 @@ +package textfile + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "math/big" + "os" + "strconv" + "strings" + "time" + + "github.com/parquet-go/parquet-go" + "github.com/parquet-go/parquet-go/format" +) + +// how many rows we pull from the file at a time +const parquetBatchSize int = 256 + +// julianEpoch is the Julian day number for 1970-01-01, used to decode the +// legacy INT96 timestamps written by older Hive/Impala versions. +const julianEpoch int64 = 2440588 + +// ErrParquetStdin is returned when we're asked to read parquet from a pipe. +// The file footer (which holds the schema) lives at the *end* of the file, so +// reading one requires random access -- there's nothing we can do with a stream. +var ErrParquetStdin = errors.New("parquet input cannot be read from a pipe (the file footer requires random access); please supply a file path") + +// colShape describes how a display column is assembled from the underlying +// parquet leaf columns. +type colShape int + +const ( + shapeScalar colShape = iota // one leaf, one value + shapeList // repeated values, rendered as a JSON array + shapeMap // key/value leaves, rendered as a JSON object +) + +// parquetLeaf is a single physical column in the parquet file, along with the +// knowledge of how to turn its values into text. +type parquetLeaf struct { + name string // name relative to its display column (for structs in lists) + format func(parquet.Value) string + numeric bool // formats to a bare number, so it can go into JSON unquoted + boolean bool +} + +// jsonValue renders a value for embedding in a JSON cell. +func (l *parquetLeaf) jsonValue(v parquet.Value) interface{} { + if v.IsNull() { + return nil + } + s := l.format(v) + if l.numeric { + return json.Number(s) + } + if l.boolean { + return v.Boolean() + } + return s +} + +// parquetColumn is a column as the user sees it. A scalar maps 1:1 onto a leaf; +// lists and maps gather several values (and possibly several leaves) into one +// JSON-encoded cell. +type parquetColumn struct { + name string + shape colShape + first int // index of its first leaf column + leaves []*parquetLeaf +} + +// ParquetFile reads a parquet file and presents it as tabular text records. +// It implements RecordReader, so the viewer, pager, and exporters can consume +// it exactly like a delimited text file. +type ParquetFile struct { + Filename string + naString string + + f *os.File + pf *parquet.File + + rgs []parquet.RowGroup + rgIdx int + rows parquet.Rows + + buf []parquet.Row + bufPos int + bufLen int + + header []string + cols []*parquetColumn + byLeaf [][]parquet.Value // scratch: values of the current row, per leaf + + curLineNum int + curDataLineNum int + isEOF bool + openErr error + opened bool +} + +// NewParquetFile returns a reader for a parquet file +func NewParquetFile(fname string) *ParquetFile { + return &ParquetFile{ + Filename: fname, + } +} + +// WithNAString - the value to write for NULL cells (default is empty) +func (pq *ParquetFile) WithNAString(s string) *ParquetFile { + pq.naString = s + return pq +} + +// Open eagerly opens the file and reads its schema. This is optional -- +// ReadLine will open on demand -- but calling it up front lets the caller +// report a bad path or a corrupt file before any output is written. +func (pq *ParquetFile) Open() error { + return pq.open() +} + +func (pq *ParquetFile) open() error { + if pq.opened { + return pq.openErr + } + pq.opened = true + + if pq.Filename == "-" { + pq.openErr = ErrParquetStdin + return pq.openErr + } + + f, err := os.Open(pq.Filename) + if err != nil { + pq.openErr = err + return err + } + + st, err := f.Stat() + if err != nil { + f.Close() + pq.openErr = err + return err + } + + pf, err := parquet.OpenFile(f, st.Size()) + if err != nil { + f.Close() + pq.openErr = fmt.Errorf("%s: %v", pq.Filename, err) + return pq.openErr + } + + pq.f = f + pq.pf = pf + pq.rgs = pf.RowGroups() + pq.buf = make([]parquet.Row, parquetBatchSize) + + leafIdx := 0 + pq.buildColumns(pf.Schema(), "", &leafIdx) + pq.byLeaf = make([][]parquet.Value, leafIdx) + + pq.header = make([]string, len(pq.cols)) + for i, c := range pq.cols { + pq.header[i] = c.name + } + + return nil +} + +// buildColumns walks the schema depth-first, turning it into a flat list of +// display columns. Plain groups (structs) are flattened into dotted names; +// lists and maps become a single JSON column rather than being descended into. +func (pq *ParquetFile) buildColumns(node parquet.Node, prefix string, leafIdx *int) { + for _, f := range node.Fields() { + name := f.Name() + if prefix != "" { + name = prefix + "." + name + } + + lt := logicalTypeOf(f) + isList := lt != nil && lt.List != nil + isMap := lt != nil && lt.Map != nil + + switch { + case f.Leaf() && !f.Repeated(): + col := &parquetColumn{name: name, shape: shapeScalar, first: *leafIdx} + col.leaves = []*parquetLeaf{newLeaf("", f)} + *leafIdx++ + pq.cols = append(pq.cols, col) + + case f.Leaf(): + // a repeated primitive with no LIST annotation + col := &parquetColumn{name: name, shape: shapeList, first: *leafIdx} + col.leaves = []*parquetLeaf{newLeaf("", f)} + *leafIdx++ + pq.cols = append(pq.cols, col) + + case isMap: + col := &parquetColumn{name: name, shape: shapeMap, first: *leafIdx} + collectLeaves(f, "", &col.leaves, leafIdx) + pq.cols = append(pq.cols, col) + + case isList || f.Repeated(): + col := &parquetColumn{name: name, shape: shapeList, first: *leafIdx} + collectLeaves(f, "", &col.leaves, leafIdx) + pq.cols = append(pq.cols, col) + + default: + // a plain struct -- flatten it into dotted column names + pq.buildColumns(f, name, leafIdx) + } + } +} + +// collectLeaves gathers every leaf under a list/map group in schema order, +// naming them relative to that group. +func collectLeaves(node parquet.Node, prefix string, out *[]*parquetLeaf, leafIdx *int) { + for _, f := range node.Fields() { + name := joinRelative(prefix, f.Name()) + if f.Leaf() { + *out = append(*out, newLeaf(name, f)) + *leafIdx++ + } else { + collectLeaves(f, name, out, leafIdx) + } + } +} + +// joinRelative builds a relative field name, dropping the structural wrapper +// names that the LIST and MAP annotations require ("list", "element", +// "key_value"). Those carry no information for a text rendering. +func joinRelative(prefix, name string) string { + switch name { + case "list", "element", "key_value", "array", "item": + return prefix + } + if prefix == "" { + return name + } + return prefix + "." + name +} + +// logicalTypeOf safely fetches a node's logical type. Node.Type() is documented +// as panicking on some non-leaf nodes, so guard against that. +func logicalTypeOf(n parquet.Node) (lt *format.LogicalType) { + defer func() { + recover() + }() + if t := n.Type(); t != nil { + lt = t.LogicalType() + } + return +} + +// ReadLine reads the next row from the file +func (pq *ParquetFile) ReadLine() (*TextRecord, error) { + if err := pq.open(); err != nil { + return nil, err + } + + row, err := pq.nextRow() + if err != nil { + return nil, err + } + + values := pq.formatRow(row) + + pq.curLineNum++ + pq.curDataLineNum++ + + // RawString is what gets written back out when saving from the pager, so + // it needs the same escaping the tab exporter applies. + var sb strings.Builder + for i, v := range values { + if i > 0 { + sb.WriteString("\t") + } + sb.WriteString(quoteTab(v)) + } + sb.WriteString("\n") + raw := sb.String() + + return &TextRecord{ + Values: values, + LineNum: pq.curLineNum, + DataLineNum: pq.curDataLineNum, + RawString: raw, + Flag: false, + ByteSize: len(raw), + parent: pq, + }, nil +} + +// nextRow pulls one row, refilling the batch buffer and advancing across row +// groups as needed. +func (pq *ParquetFile) nextRow() (parquet.Row, error) { + for { + if pq.bufPos < pq.bufLen { + row := pq.buf[pq.bufPos] + pq.bufPos++ + return row, nil + } + + if pq.rows == nil { + if pq.rgIdx >= len(pq.rgs) { + pq.isEOF = true + return nil, io.EOF + } + pq.rows = pq.rgs[pq.rgIdx].Rows() + pq.rgIdx++ + } + + n, err := pq.rows.ReadRows(pq.buf) + pq.bufPos = 0 + pq.bufLen = n + + // Either an error or an empty read means this row group is done. We + // still serve whatever came back in the same call before moving on. + if err != nil || n == 0 { + pq.rows.Close() + pq.rows = nil + if err != nil && err != io.EOF { + pq.isEOF = true + return nil, err + } + } + } +} + +// formatRow buckets the row's values by leaf column, then renders each display +// column. +func (pq *ParquetFile) formatRow(row parquet.Row) []string { + for i := range pq.byLeaf { + pq.byLeaf[i] = pq.byLeaf[i][:0] + } + for _, v := range row { + c := v.Column() + if c >= 0 && c < len(pq.byLeaf) { + pq.byLeaf[c] = append(pq.byLeaf[c], v) + } + } + + out := make([]string, len(pq.cols)) + for i, col := range pq.cols { + out[i] = pq.renderColumn(col) + } + return out +} + +func (pq *ParquetFile) renderColumn(col *parquetColumn) string { + switch col.shape { + case shapeScalar: + vals := pq.byLeaf[col.first] + if len(vals) == 0 || vals[0].IsNull() { + return pq.naString + } + return col.leaves[0].format(vals[0]) + + case shapeMap: + if len(col.leaves) < 2 { + return pq.naString + } + keys := pq.byLeaf[col.first] + vals := pq.byLeaf[col.first+1] + if len(keys) == 0 || (len(keys) == 1 && keys[0].IsNull()) { + return pq.naString + } + // A JSON object needs string keys, so the key is always formatted as + // text even when it's numeric. + obj := make(map[string]interface{}, len(keys)) + for i, k := range keys { + if k.IsNull() { + continue + } + key := col.leaves[0].format(k) + if i < len(vals) { + obj[key] = col.leaves[1].jsonValue(vals[i]) + } else { + obj[key] = nil + } + } + return marshalCell(obj, pq.naString) + + default: // shapeList + if len(col.leaves) == 1 { + vals := pq.byLeaf[col.first] + if len(vals) == 0 || (len(vals) == 1 && vals[0].IsNull()) { + return pq.naString + } + arr := make([]interface{}, len(vals)) + for i, v := range vals { + arr[i] = col.leaves[0].jsonValue(v) + } + return marshalCell(arr, pq.naString) + } + + // a list of structs -- zip the leaves together by position + n := len(pq.byLeaf[col.first]) + if n == 0 { + return pq.naString + } + arr := make([]interface{}, 0, n) + for i := 0; i < n; i++ { + obj := make(map[string]interface{}, len(col.leaves)) + for j, leaf := range col.leaves { + vals := pq.byLeaf[col.first+j] + if i < len(vals) { + obj[leaf.name] = leaf.jsonValue(vals[i]) + } else { + obj[leaf.name] = nil + } + } + arr = append(arr, obj) + } + return marshalCell(arr, pq.naString) + } +} + +func marshalCell(v interface{}, na string) string { + b, err := json.Marshal(v) + if err != nil { + return na + } + return string(b) +} + +// Close the file +func (pq *ParquetFile) Close() { + if pq.rows != nil { + pq.rows.Close() + pq.rows = nil + } + if pq.f != nil { + pq.f.Close() + pq.f = nil + } +} + +// GetHeader returns the flattened column names from the parquet schema +func (pq *ParquetFile) GetHeader() []string { + pq.open() + return pq.header +} + +// IsEOF - have we read every row? +func (pq *ParquetFile) IsEOF() bool { + return pq.isEOF +} + +// NoHeader is always false -- the parquet schema is the header +func (pq *ParquetFile) NoHeader() bool { + return false +} + +// HeaderLine returns the column names as a tab-delimited line +func (pq *ParquetFile) HeaderLine() string { + pq.open() + var sb strings.Builder + for i, v := range pq.header { + if i > 0 { + sb.WriteString("\t") + } + sb.WriteString(quoteTab(v)) + } + sb.WriteString("\n") + return sb.String() +} + +// newLeaf builds the text formatter for a single physical column. +// +// Note that parquet.Value.String() is deliberately not used here: it renders +// the *physical* type only, so a DATE would come out as a day count and a +// DECIMAL(10,2) as its raw unscaled integer. +func newLeaf(name string, n parquet.Node) *parquetLeaf { + leaf := &parquetLeaf{name: name} + t := n.Type() + lt := t.LogicalType() + + if lt != nil { + switch { + case lt.UTF8 != nil, lt.Enum != nil, lt.Json != nil: + leaf.format = func(v parquet.Value) string { return string(v.ByteArray()) } + return leaf + + case lt.Bson != nil: + leaf.format = func(v parquet.Value) string { return fmt.Sprintf("%x", v.ByteArray()) } + return leaf + + case lt.Decimal != nil: + scale := lt.Decimal.Scale + leaf.numeric = true + leaf.format = func(v parquet.Value) string { return formatDecimal(v, scale) } + return leaf + + case lt.Date != nil: + leaf.format = func(v parquet.Value) string { + return time.Unix(int64(v.Int32())*86400, 0).UTC().Format("2006-01-02") + } + return leaf + + case lt.Time != nil: + unit := lt.Time.Unit + leaf.format = func(v parquet.Value) string { return formatParquetTime(v, unit) } + return leaf + + case lt.Timestamp != nil: + ts := lt.Timestamp + leaf.format = func(v parquet.Value) string { return formatTimestamp(v, ts) } + return leaf + + case lt.UUID != nil: + leaf.format = func(v parquet.Value) string { return formatUUID(v.ByteArray()) } + return leaf + + case lt.Integer != nil: + it := lt.Integer + leaf.numeric = true + leaf.format = func(v parquet.Value) string { return formatInteger(v, it) } + return leaf + } + } + + // no logical type -- fall back to the physical representation + switch t.Kind() { + case parquet.Boolean: + leaf.boolean = true + leaf.format = func(v parquet.Value) string { return strconv.FormatBool(v.Boolean()) } + case parquet.Int32: + leaf.numeric = true + leaf.format = func(v parquet.Value) string { return strconv.FormatInt(int64(v.Int32()), 10) } + case parquet.Int64: + leaf.numeric = true + leaf.format = func(v parquet.Value) string { return strconv.FormatInt(v.Int64(), 10) } + case parquet.Int96: + leaf.format = formatInt96 + case parquet.Float: + leaf.numeric = true + leaf.format = func(v parquet.Value) string { + return strconv.FormatFloat(float64(v.Float()), 'g', -1, 32) + } + case parquet.Double: + leaf.numeric = true + leaf.format = func(v parquet.Value) string { + return strconv.FormatFloat(v.Double(), 'g', -1, 64) + } + case parquet.ByteArray: + leaf.format = func(v parquet.Value) string { return string(v.ByteArray()) } + default: // FixedLenByteArray and anything else -- show the bytes + leaf.format = func(v parquet.Value) string { return fmt.Sprintf("%x", v.ByteArray()) } + } + + return leaf +} + +// decimalUnscaled pulls the unscaled integer out of a DECIMAL value. It can be +// backed by an int32, an int64, or a big-endian two's complement byte array. +func decimalUnscaled(v parquet.Value) *big.Int { + switch v.Kind() { + case parquet.Int32: + return big.NewInt(int64(v.Int32())) + case parquet.Int64: + return big.NewInt(v.Int64()) + default: + b := v.ByteArray() + i := new(big.Int).SetBytes(b) + if len(b) > 0 && b[0]&0x80 != 0 { + // negative: subtract 2^(8n) to undo the two's complement + i.Sub(i, new(big.Int).Lsh(big.NewInt(1), uint(len(b)*8))) + } + return i + } +} + +func formatDecimal(v parquet.Value, scale int32) string { + i := decimalUnscaled(v) + + if scale == 0 { + return i.String() + } + if scale < 0 { + // a negative scale means the value is scaled *up* + i.Mul(i, new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(-scale)), nil)) + return i.String() + } + + neg := i.Sign() < 0 + digits := new(big.Int).Abs(i).String() + if len(digits) <= int(scale) { + digits = strings.Repeat("0", int(scale)-len(digits)+1) + digits + } + + cut := len(digits) - int(scale) + out := digits[:cut] + "." + digits[cut:] + if neg { + out = "-" + out + } + return out +} + +func formatParquetTime(v parquet.Value, unit format.TimeUnit) string { + var d time.Duration + switch { + case unit.Millis != nil: + d = time.Duration(v.Int32()) * time.Millisecond + case unit.Micros != nil: + d = time.Duration(v.Int64()) * time.Microsecond + default: + d = time.Duration(v.Int64()) + } + return time.Unix(0, 0).UTC().Add(d).Format("15:04:05.999999999") +} + +func formatTimestamp(v parquet.Value, ts *format.TimestampType) string { + n := v.Int64() + + var t time.Time + switch { + case ts.Unit.Millis != nil: + t = time.Unix(n/1e3, (n%1e3)*1e6) + case ts.Unit.Micros != nil: + t = time.Unix(n/1e6, (n%1e6)*1e3) + default: + t = time.Unix(n/1e9, n%1e9) + } + t = t.UTC() + + if ts.IsAdjustedToUTC { + return t.Format(time.RFC3339Nano) + } + // a local (unzoned) timestamp -- don't imply a timezone we don't know + return t.Format("2006-01-02T15:04:05.999999999") +} + +func formatInteger(v parquet.Value, it *format.IntType) string { + if it.IsSigned { + if v.Kind() == parquet.Int32 { + return strconv.FormatInt(int64(v.Int32()), 10) + } + return strconv.FormatInt(v.Int64(), 10) + } + + switch it.BitWidth { + case 8: + return strconv.FormatUint(uint64(uint8(v.Int32())), 10) + case 16: + return strconv.FormatUint(uint64(uint16(v.Int32())), 10) + case 32: + return strconv.FormatUint(uint64(uint32(v.Int32())), 10) + default: + return strconv.FormatUint(uint64(v.Int64()), 10) + } +} + +func formatUUID(b []byte) string { + if len(b) != 16 { + return fmt.Sprintf("%x", b) + } + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) +} + +// formatInt96 decodes the legacy INT96 timestamp: nanoseconds within the day in +// the low 64 bits, Julian day number in the high 32. +func formatInt96(v parquet.Value) string { + i := v.Int96() + nanos := int64(uint64(i[0]) | uint64(i[1])<<32) + days := int64(i[2]) - julianEpoch + return time.Unix(days*86400, nanos).UTC().Format(time.RFC3339Nano) +} diff --git a/textfile/parquet_test.go b/textfile/parquet_test.go new file mode 100644 index 0000000..8fe1044 --- /dev/null +++ b/textfile/parquet_test.go @@ -0,0 +1,212 @@ +package textfile_test + +import ( + "io" + "path/filepath" + "testing" + + "github.com/compgen-io/tabl/textfile" + "github.com/parquet-go/parquet-go" +) + +type testAddr struct { + City string `parquet:"city"` + Zip string `parquet:"zip"` +} + +// testRow covers the type mappings that are easy to get wrong -- decimals need +// their scale applied, dates and timestamps need converting out of their +// integer representations, and unsigned ints must not be printed as signed. +type testRow struct { + ID int64 `parquet:"id"` + Name *string `parquet:"name,optional"` + Score float64 `parquet:"score"` + Flag bool `parquet:"flag"` + Count uint32 `parquet:"count"` + Price int64 `parquet:"price,decimal(2:10)"` + Day int32 `parquet:"day,date"` + When int64 `parquet:"when,timestamp(microsecond)"` + Tags []string `parquet:"tags,list"` + Addr testAddr `parquet:"addr"` + Attrs map[string]string `parquet:"attrs"` +} + +func strptr(s string) *string { return &s } + +func writeTestParquet(t *testing.T) string { + t.Helper() + + name := strptr("abc") + rows := []testRow{ + { + ID: 1, + Name: name, + Score: 3.5, + Flag: true, + Count: 4000000000, // > MaxInt32, so a signed render would be negative + Price: 12345, // decimal(scale=2) -> 123.45 + Day: 19723, // 2024-01-01 + When: 1700000000000000, + Tags: []string{"a", "b"}, + Addr: testAddr{City: "Boston", Zip: "02115"}, + Attrs: map[string]string{"k": "v"}, + }, + { + ID: 2, + Name: nil, // NULL + Score: -0.25, + Flag: false, + Count: 0, + Price: -50, // -0.50 + Day: 0, // 1970-01-01 + When: 0, + Tags: []string{}, + Addr: testAddr{City: "", Zip: ""}, + Attrs: map[string]string{}, + }, + } + + path := filepath.Join(t.TempDir(), "test.parquet") + if err := parquet.WriteFile(path, rows); err != nil { + t.Fatalf("could not write test parquet: %v", err) + } + return path +} + +func readAll(t *testing.T, pq *textfile.ParquetFile) [][]string { + t.Helper() + + var out [][]string + for { + rec, err := pq.ReadLine() + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("ReadLine: %v", err) + } + out = append(out, rec.Values) + } + pq.Close() + return out +} + +func TestParquetHeader(t *testing.T) { + path := writeTestParquet(t) + + pq := textfile.NewParquetFile(path) + defer pq.Close() + + if err := pq.Open(); err != nil { + t.Fatalf("Open: %v", err) + } + + // structs flatten to dotted names; lists and maps stay a single column + want := []string{ + "id", "name", "score", "flag", "count", "price", "day", "when", + "tags", "addr.city", "addr.zip", "attrs", + } + + got := pq.GetHeader() + if len(got) != len(want) { + t.Fatalf("header = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("header[%d] = %q, want %q", i, got[i], want[i]) + } + } +} + +func TestParquetValues(t *testing.T) { + path := writeTestParquet(t) + + rows := readAll(t, textfile.NewParquetFile(path)) + if len(rows) != 2 { + t.Fatalf("read %d rows, want 2", len(rows)) + } + + tests := []struct { + row int + col int + name string + want string + }{ + {0, 0, "id", "1"}, + {0, 1, "name", "abc"}, + {0, 2, "score", "3.5"}, + {0, 3, "flag", "true"}, + {0, 4, "count", "4000000000"}, // unsigned, must not wrap negative + {0, 5, "price", "123.45"}, // decimal scale applied + {0, 6, "day", "2024-01-01"}, // days since epoch decoded + {0, 7, "when", "2023-11-14T22:13:20Z"}, + {0, 8, "tags", `["a","b"]`}, + {0, 9, "addr.city", "Boston"}, + {0, 10, "addr.zip", "02115"}, + {0, 11, "attrs", `{"k":"v"}`}, + + {1, 0, "id", "2"}, + {1, 1, "name", ""}, // NULL -> empty by default + {1, 2, "score", "-0.25"}, + {1, 3, "flag", "false"}, + {1, 5, "price", "-0.50"}, + {1, 6, "day", "1970-01-01"}, + {1, 7, "when", "1970-01-01T00:00:00Z"}, + } + + for _, tt := range tests { + got := rows[tt.row][tt.col] + if got != tt.want { + t.Errorf("row %d %s = %q, want %q", tt.row, tt.name, got, tt.want) + } + } +} + +func TestParquetNAString(t *testing.T) { + path := writeTestParquet(t) + + rows := readAll(t, textfile.NewParquetFile(path).WithNAString("NA")) + if rows[1][1] != "NA" { + t.Errorf("NULL with --na=NA = %q, want %q", rows[1][1], "NA") + } + // a non-null value must be untouched + if rows[0][1] != "abc" { + t.Errorf("non-null value = %q, want %q", rows[0][1], "abc") + } +} + +func TestParquetStdinRejected(t *testing.T) { + pq := textfile.NewParquetFile("-") + if err := pq.Open(); err != textfile.ErrParquetStdin { + t.Errorf("Open(\"-\") = %v, want ErrParquetStdin", err) + } +} + +func TestParquetMissingFile(t *testing.T) { + pq := textfile.NewParquetFile(filepath.Join(t.TempDir(), "nope.parquet")) + if err := pq.Open(); err == nil { + t.Error("Open() on a missing file returned no error") + } +} + +// A parquet file is a valid RecordReader, so it can drive the viewer/pager. +func TestParquetIsRecordReader(t *testing.T) { + path := writeTestParquet(t) + + var rd textfile.RecordReader = textfile.NewParquetFile(path) + defer rd.Close() + + rec, err := rd.ReadLine() + if err != nil { + t.Fatalf("ReadLine: %v", err) + } + if rd.NoHeader() { + t.Error("NoHeader() = true, want false (the schema is the header)") + } + if rd.HeaderLine() == "" { + t.Error("HeaderLine() is empty") + } + if v, err := rec.GetValue("addr.city"); err != nil || v != "Boston" { + t.Errorf("GetValue(addr.city) = %q, %v; want Boston", v, err) + } +} diff --git a/textfile/reader.go b/textfile/reader.go new file mode 100644 index 0000000..7a5efbe --- /dev/null +++ b/textfile/reader.go @@ -0,0 +1,27 @@ +package textfile + +// RecordReader is a source of tabular records. It is implemented by +// DelimitedTextFile (tab/CSV) and ParquetFile, and is what the viewer, pager, +// and CSV exporter consume -- they only ever need to pull records and ask +// about the header, so any format that can produce a *TextRecord will work. +type RecordReader interface { + // ReadLine returns the next record, or io.EOF when there are none left. + ReadLine() (*TextRecord, error) + + // Close releases the underlying file/stream. + Close() + + // GetHeader returns the column names. + GetHeader() []string + + // IsEOF is true once the end of the input has been reached. + IsEOF() bool + + // HeaderLine returns the header formatted as a raw line (including the + // trailing newline), suitable for writing back out to a file. + HeaderLine() string + + // NoHeader is true when the first row should be treated as data and the + // column names are synthesized (col1, col2, ...). + NoHeader() bool +} diff --git a/textfile/textfile.go b/textfile/textfile.go index edf4d48..fdefec4 100644 --- a/textfile/textfile.go +++ b/textfile/textfile.go @@ -47,7 +47,7 @@ type TextRecord struct { RawString string Flag bool ByteSize int - parent *DelimitedTextFile + parent RecordReader } // NewDelimitedFile returns an open delimited text file @@ -103,6 +103,31 @@ func (txt *DelimitedTextFile) WithHeaderComment(val bool) *DelimitedTextFile { return txt } +// GetHeader - the column names (RecordReader) +func (txt *DelimitedTextFile) GetHeader() []string { + return txt.Header +} + +// IsEOF - have we hit the end of the file? (RecordReader) +func (txt *DelimitedTextFile) IsEOF() bool { + return txt.isEOF +} + +// NoHeader - is the first row data rather than a header? (RecordReader) +func (txt *DelimitedTextFile) NoHeader() bool { + return txt.noHeader +} + +// HeaderLine - the header as it appeared in the source file (RecordReader). +// If the header was a comment, that's the line we want; otherwise it's the +// first non-comment line. Returns "" if we haven't read far enough to know. +func (txt *DelimitedTextFile) HeaderLine() string { + if txt.headerComment { + return txt.lastComment + } + return txt.rawHeaderLine +} + func (txt *DelimitedTextFile) nextRune() (rune, error) { if !txt.hasNext { err := txt.populateNext() @@ -500,7 +525,7 @@ func (txt *DelimitedTextFile) splitLine(buf string) []string { // GetValue - Fetch a value from a record by column name func (rec *TextRecord) GetValue(k string) (string, error) { - for i, v := range rec.parent.Header { + for i, v := range rec.parent.GetHeader() { if v == k { return rec.Values[i], nil } diff --git a/textfile/viewer.go b/textfile/viewer.go index f1b922b..fa817bb 100644 --- a/textfile/viewer.go +++ b/textfile/viewer.go @@ -13,7 +13,7 @@ const linesForEstimation int = 10000 // TextViewer is a viewer for tab-delimited data, it handles formatting and showing the data on a stream type TextViewer struct { - txt *DelimitedTextFile + txt RecordReader showComments bool showLineNum bool noHeader bool @@ -25,7 +25,7 @@ type TextViewer struct { } // NewTextViewer - create a new text viewer -func NewTextViewer(f *DelimitedTextFile) *TextViewer { +func NewTextViewer(f RecordReader) *TextViewer { return &TextViewer{ txt: f, showComments: false, @@ -83,12 +83,12 @@ func (tv *TextViewer) WriteFile(out io.Writer) { } if tv.colNames == nil { - tv.colNames = make([]string, len(tv.txt.Header)) - copy(tv.colNames, tv.txt.Header) - tv.colWidth = make([]int, len(tv.txt.Header)) + tv.colNames = make([]string, len(tv.txt.GetHeader())) + copy(tv.colNames, tv.txt.GetHeader()) + tv.colWidth = make([]int, len(tv.txt.GetHeader())) - for j := 0; j < len(tv.txt.Header); j++ { - r := []rune(tv.txt.Header[j] + " ") + for j := 0; j < len(tv.txt.GetHeader()); j++ { + r := []rune(tv.txt.GetHeader()[j] + " ") tv.colWidth[j] = support.MaxInt(tv.minWidth, tv.colWidth[j], len(r)) if tv.maxWidth > 0 { tv.colWidth[j] = support.MinInt(tv.colWidth[j], tv.maxWidth) @@ -96,15 +96,15 @@ func (tv *TextViewer) WriteFile(out io.Writer) { } } - if len(tv.colNames) < len(tv.txt.Header) { - tv.colNames = make([]string, len(tv.txt.Header)) - copy(tv.colNames, tv.txt.Header) - newWidths := make([]int, len(tv.txt.Header)) + if len(tv.colNames) < len(tv.txt.GetHeader()) { + tv.colNames = make([]string, len(tv.txt.GetHeader())) + copy(tv.colNames, tv.txt.GetHeader()) + newWidths := make([]int, len(tv.txt.GetHeader())) copy(newWidths, tv.colWidth) tv.colWidth = newWidths - for j := 0; j < len(tv.txt.Header); j++ { - r := []rune(tv.txt.Header[j] + " ") + for j := 0; j < len(tv.txt.GetHeader()); j++ { + r := []rune(tv.txt.GetHeader()[j] + " ") tv.colWidth[j] = support.MaxInt(tv.minWidth, tv.colWidth[j], len(r)) if tv.maxWidth > 0 { tv.colWidth[j] = support.MinInt(tv.colWidth[j], tv.maxWidth) @@ -155,7 +155,7 @@ func (tv *TextViewer) WriteFile(out io.Writer) { } func (tv *TextViewer) writeHeader(out io.Writer) { - for i, v := range tv.txt.Header { + for i, v := range tv.txt.GetHeader() { if i > 0 { fmt.Fprint(out, "| ") } @@ -172,7 +172,7 @@ func (tv *TextViewer) writeHeader(out io.Writer) { } fmt.Fprint(out, "\n") - for i := 0; i < len(tv.txt.Header); i++ { + for i := 0; i < len(tv.txt.GetHeader()); i++ { if i > 0 { fmt.Fprint(out, "=+=") } else if tv.showLineNum {