support data sampling and some clean up

Author: sholokhov

checks.yaml

@@ -1,6 +1,6 @@
 version: "1"
 validations:
-  - dataset: ch-local@[nyc_taxi.trips_small]
+  - dataset: ch@[nyc_taxi.trips_small]
     where: "pickup_datetime > '2014-01-01'"
     checks:
       - id: row_count > 0

cmd/check.go

@@ -50,7 +50,7 @@ By automating these checks, you can proactively identify and address data qualit
 						}
 
 						log.Printf("  [%d/%d] '%s': %s", cIdx+1, len(rule.Checks), check.ID, getCheckResultLabel(pass))
-						if !pass && internal.IdOrDefault(string(check.OnFail), internal.OnFailActionError) == "error" {
+						if !pass && internal.IdOrDefault(string(check.OnFail), internal.OnFailActionError) == internal.OnFailActionError {
 							exitCode = 1
 						}
 					}

cmd/import.go

@@ -58,7 +58,7 @@ 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-checks", "u", false, "Update checks config file in place")
+	cmd.Flags().BoolVarP(&updateCfg, "update-config", "u", false, "Update dbq config file in place")
 
 	return cmd
 }

cmd/profile.go

@@ -10,6 +10,7 @@ import (
 func NewProfileCommand(app internal.DbqApp) *cobra.Command {
 	var dataSource string
 	var dataSet string
+	var sample bool
 
 	cmd := &cobra.Command{
 		Use:   "profile",
@@ -39,7 +40,7 @@ and helps in making better decisions about data processing and analysis.
 			}
 
 			for _, curDataSet := range dataSetsToProfile {
-				metrics, err := app.ProfileDataset(dataSource, curDataSet)
+				metrics, err := app.ProfileDataset(dataSource, curDataSet, sample)
 				if err != nil {
 					log.Printf("Failed to profile %s: %s\n", curDataSet, err)
 				} else {
@@ -62,6 +63,7 @@ and helps in making better decisions about data processing and analysis.
 	_ = 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")
 
 	return cmd
 }

cmd/root.go

@@ -2,11 +2,14 @@ package cmd
 
 import (
 	"dbq/internal"
+	"log/slog"
 	"os"
 
 	"github.com/spf13/cobra"
 )
 
+var verbose bool
+
 var rootCmd = &cobra.Command{
 	Use:   "dbq",
 	Short: "dbq is a CLI tool for profiling data and running quality checks across various data sources",
@@ -25,10 +28,15 @@ func AddCommands(app internal.DbqApp) {
 	rootCmd.AddCommand(NewCheckCommand(app))
 	rootCmd.AddCommand(NewProfileCommand(app))
 	rootCmd.AddCommand(NewVersionCommand())
+
+	if verbose {
+		app.SetLogLevel(slog.LevelInfo)
+	}
 }
 
 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")
 }

dbq.yaml

@@ -1,6 +1,6 @@
 version: "1"
 datasources:
-    - id: ch-local
+    - id: ch
       type: clickhouse
       configuration:
         host: 0.0.0.0:19000
@@ -9,14 +9,4 @@ datasources:
         password: changeme
         database: default
       datasets:
-        - nyc_taxi.trips_big
         - nyc_taxi.trips_small
-    - id: pgsql
-      type: postgres
-      configuration:
-        host: 127.0.0.1
-        port: 9004
-        username: simple
-        password: simple_pass
-      datasets:
-        - public.table_1

internal/app.go

@@ -5,30 +5,34 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"gopkg.in/yaml.v3"
+	"log/slog"
 	"os"
 	"strings"
 )
 
 type DbqApp interface {
 	PingDataSource(srcId string) (string, error)
 	ImportDatasets(srcId string, filter string) ([]string, error)
-	ProfileDataset(srcId string, dataset string) (*TableMetrics, error)
+	ProfileDataset(srcId string, dataset string, sample bool) (*TableMetrics, error)
 	GetDbqConfig() *DbqConfig
 	SaveDbqConfig() error
 	FindDataSourceById(srcId string) *DataSource
 	RunCheck(check *Check, dataSource *DataSource, dataset string, defaultWhere string) (bool, string, error)
+	SetLogLevel(level slog.Level)
 }
 
 type DbqAppImpl struct {
 	dbqConfigPath string
 	dbqConfig     *DbqConfig
+	logLevel      slog.Level
 }
 
 func NewDbqApp(dbqConfigPath string) DbqApp {
 	dbqConfig, dbqConfigUsedPath := initConfig(dbqConfigPath)
 	return &DbqAppImpl{
 		dbqConfigPath: dbqConfigUsedPath,
 		dbqConfig:     dbqConfig,
+		logLevel:      slog.LevelError,
 	}
 }
 
@@ -58,13 +62,13 @@ func (app *DbqAppImpl) ImportDatasets(srcId string, filter string) ([]string, er
 	return cnn.ImportDatasets(filter)
 }
 
-func (app *DbqAppImpl) ProfileDataset(srcId string, dataset string) (*TableMetrics, error) {
+func (app *DbqAppImpl) ProfileDataset(srcId string, dataset string, sample bool) (*TableMetrics, error) {
 	var dataSource = app.FindDataSourceById(srcId)
 	cnn, err := getDbqConnector(*dataSource)
 	if err != nil {
 		return nil, err
 	}
-	return cnn.ProfileDataset(dataset)
+	return cnn.ProfileDataset(dataset, sample)
 }
 
 func (app *DbqAppImpl) GetDbqConfig() *DbqConfig {
@@ -102,6 +106,10 @@ func (app *DbqAppImpl) RunCheck(check *Check, dataSource *DataSource, dataset st
 	return cnn.RunCheck(check, dataset, defaultWhere)
 }
 
+func (app *DbqAppImpl) SetLogLevel(logLevel slog.Level) {
+	app.logLevel = logLevel
+}
+
 func initConfig(dbqConfigPath string) (*DbqConfig, string) {
 	v := viper.New()
 

internal/clickhouse.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"log"
 	"log/slog"
+	"reflect"
 	"regexp"
 	"strings"
 	"time"
@@ -84,7 +85,7 @@ func (c *ClickhouseDbqConnector) ImportDatasets(filter string) ([]string, error)
 	return datasets, nil
 }
 
-func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics, error) {
+func (c *ClickhouseDbqConnector) ProfileDataset(dataset string, sample bool) (*TableMetrics, error) {
 	startTime := time.Now()
 	ctx := context.Background()
 
@@ -112,6 +113,46 @@ func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics,
 	}
 	slog.Debug("Total rows: %d", metrics.TotalRows)
 
+	// sample data if enabled
+	if sample {
+		sampleQuery := fmt.Sprintf("select * from %s.%s order by rand() limit 100", databaseName, tableName)
+
+		toCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		rows, err := c.cnn.Query(toCtx, sampleQuery)
+		if err != nil {
+			log.Printf("Warning: Failed to sample data %s: %v", err)
+		}
+		defer rows.Close()
+
+		var allRows []map[string]interface{}
+		for rows.Next() {
+			scanArgs := make([]interface{}, len(rows.Columns()))
+			for i, colType := range rows.ColumnTypes() {
+				scanType := colType.ScanType()
+				valuePtr := reflect.New(scanType).Interface()
+				scanArgs[i] = valuePtr
+			}
+
+			err = rows.Scan(scanArgs...)
+			if err != nil {
+				log.Printf("Warning: Failed to scan row: %v", err)
+				continue
+			}
+
+			rowData := make(map[string]interface{})
+			for i, colName := range rows.Columns() {
+				scannedValue := reflect.ValueOf(scanArgs[i]).Elem().Interface()
+				rowData[colName] = scannedValue
+			}
+
+			allRows = append(allRows, rowData)
+		}
+
+		metrics.RowsSample = allRows
+	}
+
 	// Get Column Information (Name and Type)
 	columnsToProcess, err := fetchColumns(c.cnn, ctx, databaseName, tableName)
 	if err != nil {
@@ -131,9 +172,10 @@ func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics,
 		colStartTime := time.Now()
 		log.Printf("Processing column: %s (Type: %s)", col.Name, col.Type)
 		colMetrics := &ColumnMetrics{
-			ColumnName:    col.Name,
-			DataType:      col.Type,
-			ColumnComment: col.Comment,
+			ColumnName:     col.Name,
+			DataType:       col.Type,
+			ColumnComment:  col.Comment,
+			ColumnPosition: col.Position,
 		}
 
 		// Null Count (all types)
@@ -259,7 +301,7 @@ func (c *ClickhouseDbqConnector) RunCheck(check *Check, dataset string, defaultW
 
 func fetchColumns(cnn driver.Conn, ctx context.Context, databaseName string, tableName string) ([]ColumnInfo, error) {
 	columnQuery := `
-        SELECT name, type, comment
+        SELECT name, type, comment, position
         FROM system.columns
         WHERE database = ? AND table = ?
         ORDER BY position`
@@ -273,10 +315,11 @@ func fetchColumns(cnn driver.Conn, ctx context.Context, databaseName string, tab
 	var cols []ColumnInfo
 	for rows.Next() {
 		var colName, colType, comment string
-		if err := rows.Scan(&colName, &colType, &comment); err != nil {
+		var pos uint64
+		if err := rows.Scan(&colName, &colType, &comment, &pos); err != nil {
 			return nil, fmt.Errorf("failed to scan column info: %w", err)
 		}
-		cols = append(cols, ColumnInfo{Name: colName, Type: colType, Comment: comment})
+		cols = append(cols, ColumnInfo{Name: colName, Type: colType, Comment: comment, Position: uint(pos)})
 	}
 
 	if err = rows.Err(); err != nil {

internal/dbq_connector.go

@@ -3,7 +3,7 @@ package internal
 type DbqConnector interface {
 	Ping() (string, error)
 	ImportDatasets(filter string) ([]string, error)
-	ProfileDataset(dataset string) (*TableMetrics, error)
+	ProfileDataset(dataset string, sample bool) (*TableMetrics, error)
 	RunCheck(check *Check, dataset string, defaultWhere string) (bool, string, error)
 }
 
@@ -12,8 +12,9 @@ const (
 )
 
 type ColumnMetrics struct {
-	ColumnName          string   `json:"column_name"`
-	ColumnComment       string   `json:"column_comment"`
+	ColumnName          string   `json:"col_name"`
+	ColumnComment       string   `json:"col_comment"`
+	ColumnPosition      uint     `json:"col_position"`
 	DataType            string   `json:"data_type"`
 	NullCount           uint64   `json:"null_count"`
 	BlankCount          *int64   `json:"blank_count,omitempty"`         // string only
@@ -31,13 +32,15 @@ type TableMetrics struct {
 	DatabaseName        string                    `json:"database_name"`
 	TotalRows           uint64                    `json:"total_rows"`
 	ColumnsMetrics      map[string]*ColumnMetrics `json:"columns_metrics"`
+	RowsSample          []map[string]interface{}  `json:"rows_sample"`
 	ProfilingDurationMs int64                     `json:"profiling_duration_ms"`
 }
 
 type ColumnInfo struct {
-	Name    string
-	Type    string
-	Comment string
+	Name     string
+	Type     string
+	Comment  string
+	Position uint
 }
 
 type ProfileResultOutput struct {

readme.md

@@ -54,6 +54,11 @@ datasources:
 
 ### Checks example
 
+```yaml
+# checks.yaml
+
+```
+
 ### Commands
 
 ```bash
@@ -76,9 +81,9 @@ Available Commands:
 Flags:
       --config string   config file (default is $HOME/.dbq.yaml or ./dbq.yaml)
   -h, --help            help for dbq
+  -v, --verbose         Enables verbose logging
 
 Use "dbq [command] --help" for more information about a command.
-
 ```
 
 ### Quick start