output format improvements

Author: sholokhov

checks.yaml

@@ -1,6 +1,6 @@
 version: "1"
 validations:
-  - dataset: ch-local@[nyc_taxi.trips_small, nyc_taxi.trips_big]
+  - dataset: ch-local@[nyc_taxi.trips_small]
     where: "pickup_datetime > '2014-01-01'"
     checks:
       - id: row_count > 0
@@ -27,6 +27,10 @@ validations:
         description: "sum of value"
         severity: error
 
+      - id: countIf(trip_id == 1) == 1
+        description: "check trip id"
+        severity: warn
+
       - id: raw_query
         description: "some raw query description here"
         severity: error

cmd/check.go

@@ -4,6 +4,7 @@ import (
 	"dbq/internal"
 	"fmt"
 	"log"
+	"os"
 	"strings"
 
 	"github.com/spf13/cobra"
@@ -28,9 +29,8 @@ By automating these checks, you can proactively identify and address data qualit
 				return fmt.Errorf("error while loading checks configuration file: %w", err)
 			}
 
-			for i, rule := range checksCfg.Validations {
-				log.Printf("Running check for %s [%d/%d]", rule.Dataset, i+1, len(checksCfg.Validations))
-
+			shouldFail := false
+			for _, rule := range checksCfg.Validations {
 				dataSourceId, datasets, err := parseDatasetString(rule.Dataset)
 				if err != nil {
 					return fmt.Errorf("error while parsing dataset property: %w", err)
@@ -42,18 +42,26 @@ By automating these checks, you can proactively identify and address data qualit
 				}
 
 				for dsIdx, dataset := range datasets {
-					log.Printf("  [%d/%d] Running checks for: %s", dsIdx+1, len(datasets), dataset)
-					for _, check := range rule.Checks {
-						_, err := app.RunCheck(&check, dataSource, dataset, rule.Where)
+					log.Printf("[%d/%d] Running quality checks for: %s", dsIdx+1, len(datasets), dataset)
+					for cIdx, check := range rule.Checks {
+						pass, _, err := app.RunCheck(&check, dataSource, dataset, rule.Where)
 						if err != nil {
 							log.Printf("Failed to run check: %s", err.Error())
 						}
-						// todo: act on check result
-						// if check.Severity {...}
+
+						log.Printf("  [%d/%d] '%s': %s", cIdx+1, len(rule.Checks), check.ID, getCheckResultLabel(pass))
+						if !pass && check.Severity == "error" {
+							shouldFail = true
+						}
 					}
 				}
 			}
 
+			if shouldFail {
+				log.Printf("One or more checks with 'error' severity have failed, exiting...")
+				os.Exit(1)
+			}
+
 			return nil
 		},
 	}
@@ -98,3 +106,11 @@ func parseDatasetString(input string) (datasource string, datasets []string, err
 
 	return datasource, datasets, nil
 }
+
+func getCheckResultLabel(passed bool) string {
+	if passed {
+		return "passed"
+	} else {
+		return "failed"
+	}
+}

internal/app.go

@@ -16,7 +16,7 @@ type DbqApp interface {
 	GetDbqConfig() *DbqConfig
 	SaveDbqConfig() error
 	FindDataSourceById(srcId string) *DataSource
-	RunCheck(check *Check, dataSource *DataSource, dataset string, defaultWhere string) (string, error)
+	RunCheck(check *Check, dataSource *DataSource, dataset string, defaultWhere string) (bool, string, error)
 }
 
 type DbqAppImpl struct {
@@ -94,10 +94,10 @@ func (app *DbqAppImpl) FindDataSourceById(srcId string) *DataSource {
 	return nil
 }
 
-func (app *DbqAppImpl) RunCheck(check *Check, dataSource *DataSource, dataset string, defaultWhere string) (string, error) {
+func (app *DbqAppImpl) RunCheck(check *Check, dataSource *DataSource, dataset string, defaultWhere string) (bool, string, error) {
 	cnn, err := getDbqConnector(*dataSource)
 	if err != nil {
-		return "", err
+		return false, "", err
 	}
 	return cnn.RunCheck(check, dataset, defaultWhere)
 }

internal/checks_config.go

@@ -2,7 +2,6 @@ package internal
 
 import (
 	"gopkg.in/yaml.v3"
-	"log"
 	"os"
 )
 
@@ -28,14 +27,12 @@ func LoadChecksConfig(fileName string) (*ChecksConfig, error) {
 	file, err := os.Open(fileName)
 	defer file.Close()
 	if err != nil {
-		log.Printf("Error opening file: %v\n", err)
 		return nil, err
 	}
 
 	var cfg ChecksConfig
 	decoder := yaml.NewDecoder(file)
 	if err := decoder.Decode(&cfg); err != nil {
-		log.Printf("Error decoding YAML: %v\n", err)
 		return nil, err
 	}
 

internal/clickhouse.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"log"
+	"log/slog"
 	"regexp"
 	"strings"
 	"time"
@@ -101,15 +102,15 @@ func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics,
 		ColumnsMetrics: make(map[string]*ColumnMetrics),
 	}
 
-	log.Printf("Calculating metrics for table: %s", dataset)
+	slog.Debug("Calculating metrics for table:", dataset)
 
 	// Total Row Count
-	log.Printf("Fetching total row count...")
+	slog.Debug("Fetching total row count...")
 	err := c.cnn.QueryRow(ctx, fmt.Sprintf("SELECT count() FROM %s", dataset)).Scan(&metrics.TotalRows)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get total row count for %s: %w", dataset, err)
 	}
-	log.Printf("Total rows: %d", metrics.TotalRows)
+	slog.Debug("Total rows: %d", metrics.TotalRows)
 
 	// Get Column Information (Name and Type)
 	columnsToProcess, err := fetchColumns(c.cnn, ctx, databaseName, tableName)
@@ -118,7 +119,7 @@ func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics,
 	}
 
 	if len(columnsToProcess) == 0 {
-		log.Printf("Warning: No columns found for table %s. Returning basic info.", dataset)
+		slog.Warn("Warning: No columns found for table %s. Returning basic info.", dataset)
 		metrics.ProfilingDurationMs = time.Since(startTime).Milliseconds()
 		return metrics, nil
 	}
@@ -221,40 +222,39 @@ func (c *ClickhouseDbqConnector) ProfileDataset(dataset string) (*TableMetrics,
 	return metrics, nil
 }
 
-func (c *ClickhouseDbqConnector) RunCheck(check *Check, dataset string, defaultWhere string) (string, error) {
+func (c *ClickhouseDbqConnector) RunCheck(check *Check, dataset string, defaultWhere string) (bool, string, error) {
 	if c.cnn == nil {
-		return "", fmt.Errorf("database connection is not initialized")
+		return false, "", fmt.Errorf("database connection is not initialized")
 	}
 
 	query, err := generateDataCheckQuery(check, dataset, defaultWhere)
 	if err != nil {
-		return "", fmt.Errorf("failed to generate SQL for check (%s)/(%s): %s", check.ID, dataset, err.Error())
+		return false, "", fmt.Errorf("failed to generate SQL for check (%s)/(%s): %s", check.ID, dataset, err.Error())
 	}
 
-	log.Printf("Executing SQL for '%s': %s", check.ID, query)
+	// todo: debug
+	// log.Printf("Executing SQL for '%s': %s", check.ID, query)
 
-	startTime := time.Now()
+	// startTime := time.Now()
 	rows, err := c.cnn.Query(context.Background(), query)
 	if err != nil {
-		return "", fmt.Errorf("failed to query database: %w", err)
+		return false, "", fmt.Errorf("failed to query database: %w", err)
 	}
 	defer rows.Close()
-	elapsed := time.Since(startTime).Milliseconds()
+	// _ := time.Since(startTime).Milliseconds()
 
+	var checkPassed bool
 	for rows.Next() {
-		var checkPassed bool
 		if err := rows.Scan(&checkPassed); err != nil {
-			return "", fmt.Errorf("failed to scan row: %w", err)
+			return false, "", fmt.Errorf("failed to scan row: %w", err)
 		}
-		log.Printf("Check passed: %t (in %d ms)", checkPassed, elapsed)
-		log.Printf("---")
 	}
 
 	if err = rows.Err(); err != nil {
-		return "", fmt.Errorf("error occurred during row iteration: %w", err)
+		return false, "", fmt.Errorf("error occurred during row iteration: %w", err)
 	}
 
-	return "", nil
+	return checkPassed, "", nil
 }
 
 func fetchColumns(cnn driver.Conn, ctx context.Context, databaseName string, tableName string) ([]ColumnInfo, error) {
@@ -379,7 +379,7 @@ func isStringCHType(dataType string) bool {
 
 func startWithAnyOf(prefixes []string, s string) bool {
 	for _, prefix := range prefixes {
-		if strings.HasPrefix(s, prefix) {
+		if strings.HasPrefix(s, strings.ToLower(prefix)) {
 			return true
 		}
 	}

internal/dbq_config.go

@@ -1,11 +1,5 @@
 package internal
 
-import (
-	"gopkg.in/yaml.v3"
-	"log"
-	"os"
-)
-
 type DbqConfig struct {
 	Version     string       `yaml:"version"`
 	DataSources []DataSource `yaml:"datasources"`
@@ -25,21 +19,3 @@ type ConfigDetails struct {
 	Password string `yaml:"password"`
 	Database string `yaml:"database,omitempty"`
 }
-
-func LoadDbqSetting(fileName string) (*DbqConfig, error) {
-	file, err := os.Open(fileName)
-	defer file.Close()
-	if err != nil {
-		log.Printf("Error opening file: %v\n", err)
-		return nil, err
-	}
-
-	var settings DbqConfig
-	decoder := yaml.NewDecoder(file)
-	if err := decoder.Decode(&settings); err != nil {
-		log.Printf("Error decoding YAML: %v\n", err)
-		return nil, err
-	}
-
-	return &settings, nil
-}

internal/dbq_connector.go

@@ -4,7 +4,7 @@ type DbqConnector interface {
 	Ping() (string, error)
 	ImportDatasets(filter string) ([]string, error)
 	ProfileDataset(dataset string) (*TableMetrics, error)
-	RunCheck(check *Check, dataset string, defaultWhere string) (string, error)
+	RunCheck(check *Check, dataset string, defaultWhere string) (bool, string, error)
 }
 
 const (
@@ -16,12 +16,12 @@ type ColumnMetrics struct {
 	ColumnComment       string   `json:"column_comment"`
 	DataType            string   `json:"data_type"`
 	NullCount           uint64   `json:"null_count"`
-	BlankCount          *int64   `json:"blank_count,omitempty"`         // Applicable only for String types
-	MinValue            *float64 `json:"min_value,omitempty"`           // Numeric only
-	MaxValue            *float64 `json:"max_value,omitempty"`           // Numeric only
-	AvgValue            *float64 `json:"avg_value,omitempty"`           // Numeric only
-	StddevValue         *float64 `json:"stddev_value,omitempty"`        // Numeric only (Population StdDev)
-	MostFrequentValue   *string  `json:"most_frequent_value,omitempty"` // Using NullString to handle NULL as most frequent
+	BlankCount          *int64   `json:"blank_count,omitempty"`         // string only
+	MinValue            *float64 `json:"min_value,omitempty"`           // numeric only
+	MaxValue            *float64 `json:"max_value,omitempty"`           // numeric only
+	AvgValue            *float64 `json:"avg_value,omitempty"`           // numeric only
+	StddevValue         *float64 `json:"stddev_value,omitempty"`        // numeric only (Population StdDev)
+	MostFrequentValue   *string  `json:"most_frequent_value,omitempty"` // pointer to handle NULL as most frequent
 	ProfilingDurationMs int64    `json:"profiling_duration_ms"`
 }
 

readme.md

@@ -28,11 +28,11 @@ It is designed to be easy to use and integrate into your existing workflow.
 - [x] implement aliases for common checks based on raw sql check
 - [x] fix cmd descriptions
 - [x] review todos
+- [x] improve output
 - [ ] basic cross validation (dataset is defined)
 - [ ] review logs
 - [ ] review crashes (wrong arguments)
 - [ ] default values (e.g. severity)
-- [ ] improve output
 - [ ] quiet/verbose mode for logs
 - [ ] docs