support parallel column metrics computation

Author: sholokhov

.gitignore

@@ -27,3 +27,4 @@ go.work.sum
 # env file
 .env
 
+build

cmd/check.go

@@ -109,7 +109,7 @@ By automating these checks, you can proactively identify and address data qualit
 		},
 	}
 
-	cmd.Flags().StringVarP(&checksFile, "checks", "c", "", "Path to data quality checks file")
+	cmd.Flags().StringVarP(&checksFile, "checks", "c", "", "path to data quality checks file")
 	_ = cmd.MarkFlagRequired("checks")
 
 	return cmd

cmd/import.go

@@ -70,9 +70,9 @@ This command is useful for quickly onboarding data from external systems, allowi
 		},
 	}
 
-	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "Datasource from which datasets will be imported")
-	cmd.Flags().StringVarP(&filter, "filter", "f", "", "Filter applied for dataset selection")
-	cmd.Flags().BoolVarP(&updateCfg, "update-config", "u", false, "Update dbq config file in place")
+	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "datasource from which datasets will be imported")
+	cmd.Flags().StringVarP(&filter, "filter", "f", "", "filter applied for dataset selection")
+	cmd.Flags().BoolVarP(&updateCfg, "update-config", "u", false, "update dbq config file in place")
 
 	return cmd
 }

cmd/ping.go

@@ -50,7 +50,7 @@ This is useful for quickly determining if the data source is online and respondi
 		},
 	}
 
-	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "Datasource to ping")
+	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "datasource to ping")
 
 	return cmd
 }

cmd/profile.go

@@ -20,12 +20,18 @@ import (
 	"github.com/DataBridgeTech/dbqcore"
 	"github.com/DataBridgeTech/dbqctl/internal"
 	"github.com/spf13/cobra"
+	"runtime"
 )
 
+type ProfileResultOutput struct {
+	Profiles map[string]*dbqcore.TableMetrics `json:"profiles"`
+}
+
 func NewProfileCommand(app internal.DbqCliApp) *cobra.Command {
 	var dataSource string
 	var dataSet string
 	var sample bool
+	var maxConcurrent int
 
 	cmd := &cobra.Command{
 		Use:   "profile",
@@ -50,13 +56,13 @@ and helps in making better decisions about data processing and analysis.
 				}
 			}
 
-			profileResults := &dbqcore.ProfileResultOutput{
+			profileResults := &ProfileResultOutput{
 				Profiles: make(map[string]*dbqcore.TableMetrics),
 			}
 
 			for _, curDataSet := range dataSetsToProfile {
-				fmt.Printf("Profiling '%s', this may take some time...\n", curDataSet)
-				metrics, err := app.ProfileDataset(dataSource, curDataSet, sample)
+				fmt.Printf("Profiling '%s' (using %d jobs) , this may take some time...\n", curDataSet, maxConcurrent)
+				metrics, err := app.ProfileDataset(dataSource, curDataSet, sample, maxConcurrent)
 				if err != nil {
 					fmt.Printf("Failed to profile %s: %s\n", curDataSet, err)
 				} else {
@@ -76,11 +82,12 @@ and helps in making better decisions about data processing and analysis.
 		},
 	}
 
-	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "Datasource in which datasets will be profiled")
+	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "datasource in which datasets will be profiled")
 	_ = cmd.MarkFlagRequired("datasource")
 
-	cmd.Flags().StringVarP(&dataSet, "dataset", "s", "", "Dataset within specified data source")
-	cmd.Flags().BoolVarP(&sample, "sample", "m", false, "Include data samples in profiling report")
+	cmd.Flags().StringVarP(&dataSet, "dataset", "s", "", "dataset within specified data source")
+	cmd.Flags().BoolVarP(&sample, "sample", "m", false, "include data samples in profiling report")
+	cmd.Flags().IntVarP(&maxConcurrent, "jobs", "j", runtime.NumCPU(), "set the maximum number of jobs to execute against the datasource during profiling. By default, this is equal to the number of CPUs on the host machine.")
 
 	return cmd
 }

cmd/root.go

@@ -52,5 +52,5 @@ func init() {
 	// workaround for bootstrap config flag & unsupported flag issue
 	var dbqConfigFile string
 	rootCmd.PersistentFlags().StringVar(&dbqConfigFile, "config", "", "config file (default is $HOME/.dbq.yaml or ./dbq.yaml)")
-	rootCmd.PersistentFlags().BoolVarP(&verbose, "verbose", "v", false, "Enables verbose logging")
+	rootCmd.PersistentFlags().BoolVarP(&verbose, "verbose", "v", false, "enables verbose logging")
 }

cmd/version.go

@@ -22,7 +22,7 @@ import (
 )
 
 const (
-	DbqCtlVersion = "v0.0.7"
+	DbqCtlVersion = "v0.1.0"
 )
 
 func NewVersionCommand() *cobra.Command {

go.sum

@@ -4,6 +4,7 @@ github.com/ClickHouse/clickhouse-go/v2 v2.34.0 h1:Y4rqkdrRHgExvC4o/NTbLdY5LFQ3LH
 github.com/ClickHouse/clickhouse-go/v2 v2.34.0/go.mod h1:yioSINoRLVZkLyDzdMXPLRIqhDvel8iLBlwh6Iefso8=
 github.com/DataBridgeTech/dbqcore v0.0.2 h1:fvGqKOlphcvERnQl7cLIbm7IRhprGbY7la29U9a0fR8=
 github.com/DataBridgeTech/dbqcore v0.0.2/go.mod h1:U2++eavpf8oSKxukSEJpKGfPaOtkApA7AUpzwkrXmVM=
+github.com/DataBridgeTech/dbqcore v0.0.3/go.mod h1:U2++eavpf8oSKxukSEJpKGfPaOtkApA7AUpzwkrXmVM=
 github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
 github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=

internal/app.go

@@ -30,7 +30,7 @@ import (
 type DbqCliApp interface {
 	PingDataSource(srcId string) (string, error)
 	ImportDatasets(srcId string, filter string) ([]string, error)
-	ProfileDataset(srcId string, dataset string, sample bool) (*dbqcore.TableMetrics, error)
+	ProfileDataset(srcId string, dataset string, sample bool, maxConcurrent int) (*dbqcore.TableMetrics, error)
 	RunCheck(check *dbqcore.Check, dataSource *dbqcore.DataSource, dataset string, defaultWhere string) (bool, string, error)
 	GetDbqConfig() *dbqcore.DbqConfig
 	SaveDbqConfig() error
@@ -80,15 +80,15 @@ func (app *DbqAppImpl) ImportDatasets(srcId string, filter string) ([]string, er
 	return cnn.ImportDatasets(filter)
 }
 
-func (app *DbqAppImpl) ProfileDataset(srcId string, dataset string, sample bool) (*dbqcore.TableMetrics, error) {
+func (app *DbqAppImpl) ProfileDataset(srcId string, dataset string, sample bool, maxConcurrent int) (*dbqcore.TableMetrics, error) {
 	var dataSource = app.FindDataSourceById(srcId)
 
 	cnn, err := getDbqConnector(*dataSource, app.logLevel)
 	if err != nil {
 		return nil, err
 	}
 
-	return cnn.ProfileDataset(dataset, sample)
+	return cnn.ProfileDataset(dataset, sample, maxConcurrent)
 }
 
 func (app *DbqAppImpl) GetDbqConfig() *dbqcore.DbqConfig {