feat: influx inspect export parquet

cmd/influx_inspect/README.md

@@ -49,17 +49,26 @@ WAL storage path.
 `default` = "$HOME/.influxdb/wal"
 
 #### `-out` string
-Destination file to export to
+Destination file to export to.
+In case of export to Parquet, destination should be existing directory.
 
 `default` = "$HOME/.influxdb/export"
 
-#### `-database` string (optional)
+#### `-database` string
 Database to export.
+Mandatory for export to Parquet, optional otherwise (default).
 
 `default` = ""
 
-#### `-retention` string (optional)
+#### `-retention` string
 Retention policy to export.
+Mandatory for export to Parquet, optional otherwise (default).
+
+`default` = ""
+
+#### `-measurement` string
+Name of the measurement to export.
+Mandatory for export to Parquet, optional otherwise (default).
 
 `default` = ""
 
@@ -74,6 +83,16 @@ Compress the output.
 
 `default` = false
 
+#### `-parquet` bool (optional)
+Export to Parquet.
+
+`default` = false
+
+#### `-chunk-size` int (optional)
+Size to partition Parquet files, in bytes.
+
+`default` = 100000000
+
 #### Sample Commands
 
 Export entire database and compress output:
@@ -86,6 +105,11 @@ Export specific retention policy:
 influx_inspect export --database mydb --retention autogen
 ```
 
+Export specific measurement to Parquet:
+```
+influx_inspect export --database mydb --retention autogen --measurement cpu --parquet
+```
+
 ##### Sample Data
 This is a sample of what the output will look like.
 

cmd/influx_inspect/export/export.go

@@ -33,14 +33,20 @@ type Command struct {
 	out             string
 	database        string
 	retentionPolicy string
+	measurement     string
 	startTime       int64
 	endTime         int64
 	compress        bool
 	lponly          bool
+	parquet         bool
+	pqChunkSize     int
 
 	manifest map[string]struct{}
 	tsmFiles map[string][]string
 	walFiles map[string][]string
+
+	writeValues func(io.Writer, []byte, string, []tsm1.Value) error
+	exportDone  func(string) error
 }
 
 const stdoutMark = "-"
@@ -68,17 +74,20 @@ func (cmd *Command) Run(args ...string) error {
 	fs := flag.NewFlagSet("export", flag.ExitOnError)
 	fs.StringVar(&cmd.dataDir, "datadir", os.Getenv("HOME")+"/.influxdb/data", "Data storage path")
 	fs.StringVar(&cmd.walDir, "waldir", os.Getenv("HOME")+"/.influxdb/wal", "WAL storage path")
-	fs.StringVar(&cmd.out, "out", os.Getenv("HOME")+"/.influxdb/export", "'-' for standard out or the destination file to export to")
+	fs.StringVar(&cmd.out, "out", os.Getenv("HOME")+"/.influxdb/export", "'-' for standard out or the destination file to export to (line protocol) | directory to write Parquet files")
 	fs.StringVar(&cmd.database, "database", "", "Optional: the database to export")
 	fs.StringVar(&cmd.retentionPolicy, "retention", "", "Optional: the retention policy to export (requires -database)")
+	fs.StringVar(&cmd.measurement, "measurement", "", "Name of measurement to export")
 	fs.StringVar(&start, "start", "", "Optional: the start time to export (RFC3339 format)")
 	fs.StringVar(&end, "end", "", "Optional: the end time to export (RFC3339 format)")
 	fs.BoolVar(&cmd.lponly, "lponly", false, "Only export line protocol")
 	fs.BoolVar(&cmd.compress, "compress", false, "Compress the output")
+	fs.BoolVar(&cmd.parquet, "parquet", false, "Export to Parquet format (requires -database -retention -measurement)")
+	fs.IntVar(&cmd.pqChunkSize, "chunk-size", 100_000_000, "Size to partition Parquet files (in bytes)")
 
 	fs.SetOutput(cmd.Stdout)
 	fs.Usage = func() {
-		fmt.Fprintf(cmd.Stdout, "Exports TSM files into InfluxDB line protocol format.\n\n")
+		fmt.Fprintf(cmd.Stdout, "Exports TSM files into InfluxDB line protocol or Parquet format.\n\n")
 		fmt.Fprintf(cmd.Stdout, "Usage: %s export [flags]\n\n", filepath.Base(os.Args[0]))
 		fs.PrintDefaults()
 	}
@@ -112,6 +121,14 @@ func (cmd *Command) Run(args ...string) error {
 		return err
 	}
 
+	if cmd.parquet {
+		cmd.writeValues = cmd.writeValuesParquet
+		cmd.exportDone = cmd.exportDoneParquet
+	} else {
+		cmd.writeValues = cmd.writeValuesLp
+		cmd.exportDone = func(_ string) error { return nil }
+	}
+
 	return cmd.export()
 }
 
@@ -122,6 +139,17 @@ func (cmd *Command) validate() error {
 	if cmd.startTime != 0 && cmd.endTime != 0 && cmd.endTime < cmd.startTime {
 		return fmt.Errorf("end time before start time")
 	}
+	if cmd.parquet {
+		if cmd.database == "" || cmd.retentionPolicy == "" || cmd.measurement == "" {
+			return fmt.Errorf("must specify database, retention and measurement when exporting to Parquet")
+		}
+		if cmd.out == "-" {
+			return fmt.Errorf("-out must point to a folder for Parquet files")
+		}
+		if cmd.pqChunkSize < 1_000_000 {
+			return fmt.Errorf("minimum Parquet partition size is 1000000 bytes")
+		}
+	}
 	return nil
 }
 
@@ -133,6 +161,9 @@ func (cmd *Command) export() error {
 		return err
 	}
 
+	if cmd.parquet {
+		return cmd.writeDML(io.Discard, io.Discard)
+	}
 	return cmd.write()
 }
 
@@ -331,6 +362,9 @@ func (cmd *Command) writeTsmFiles(mw io.Writer, w io.Writer, files []string) err
 		if err := cmd.exportTSMFile(f, w); err != nil {
 			return err
 		}
+		if err := cmd.exportDone(f); err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -368,11 +402,16 @@ func (cmd *Command) exportTSMFile(tsmFilePath string, w io.Writer) error {
 		measurement, field := tsm1.SeriesAndFieldFromCompositeKey(key)
 		field = escape.Bytes(field)
 
+		if cmd.measurement != "" && cmd.measurement != strings.Split(string(measurement), ",")[0] {
+			continue
+		}
+
 		if err := cmd.writeValues(w, measurement, string(field), values); err != nil {
 			// An error from writeValues indicates an IO error, which should be returned.
 			return err
 		}
 	}
+
 	return nil
 }
 
@@ -436,19 +475,24 @@ func (cmd *Command) exportWALFile(walFilePath string, w io.Writer, warnDelete fu
 				// measurements are stored escaped, field names are not
 				field = escape.Bytes(field)
 
+				if cmd.measurement != "" && cmd.measurement != strings.Split(string(measurement), ",")[0] {
+					continue
+				}
+
 				if err := cmd.writeValues(w, measurement, string(field), values); err != nil {
 					// An error from writeValues indicates an IO error, which should be returned.
 					return err
 				}
 			}
 		}
 	}
+
 	return nil
 }
 
 // writeValues writes every value in values to w, using the given series key and field name.
 // If any call to w.Write fails, that error is returned.
-func (cmd *Command) writeValues(w io.Writer, seriesKey []byte, field string, values []tsm1.Value) error {
+func (cmd *Command) writeValuesLp(w io.Writer, seriesKey []byte, field string, values []tsm1.Value) error {
 	buf := []byte(string(seriesKey) + " " + field + "=")
 	prefixLen := len(buf)