basic checks support

Author: sholokhov

checks.yaml

@@ -1,37 +1,84 @@
 version: "1"
 validations:
-  - dataset: clickhouse-prod-euw@[ads.my_table_1]
-    where: "timestamp > '2025-01-01'"
+  # todo: support arrays of data sets [ ... ]
+  - dataset: ch-local@nyc_taxi.trips_small
+    where: "pickup_datetime > '2014-01-01'"
     checks:
-      - id: row count between 0 and 100
+      - id: row_count > 0
+        description: "data is present" # optional
+        severity: error # optional (error, warn, info), default "error"
+
+      - id: null_count(pickup_ntaname) == 0
+        description: "no nulls in column" # optional
         severity: error
-        type: row_count
-        params:
-            # filter: "timestamp > '2023-01-01'"
-            min: 0
-            max: 100
-
-      - id: "no null values"
-        description: "Check if table_1 has the correct number of columns" # optional
+
+      - id: min(pickup_datetime) < now() - interval 3 day
+        description: "min check"
         severity: error
-        type: custom
-        params:
-          query: SELECT COUNT(*) FROM ${table} WHERE column_name IS NULL
-          expected: 0
 
-      - id: "no duplicates"
-        description: "Check if table_1 has no duplicates"
+      # todo: support "between X and Y"
+      - id: stddevPop(trip_distance) < 100_000
+        description: "check stddev value"
         severity: error
-        type: no_duplicates
-        params:
-          columns: ["column_1", "column_2"]
-          # filter: "timestamp > '2023-01-01'"
 
-  - dataset: pgsql-staging@[public.table_1, public.table_2]
-    checks:
-      - id: "row count between 0 and 100"
-        severity: warn
-        type: row_count
-        params:
-          min: 0
-          max: 1000
+      - id: sum(fare_amount) <= 10_000_000
+        description: "sum of value"
+        severity: error
+
+      - id: raw_query
+        description: "some raw query description here"
+        severity: error
+        query: |
+          select countIf(trip_distance == 0) > 0 from {{table}}
+
+#  - dataset: pgsql-staging@[public.table_1, public.table_2]
+#    checks:
+#      - id: "row count between 0 and 100"
+#        severity: warn
+#        type: row_count
+#        params:
+#          min: 0
+#          max: 1000
+
+# v1 supported functions:
+# ---
+# row_count > 10
+# null_count(col) == 0
+# avg(col) <= 24.2
+# max(col) < 1000
+# min(col) == 0
+# sum(col) > 0
+# stddev(col) between 1 and 100_000_000
+# custom
+
+# AI anomaly detection
+
+# empties / blanks
+# invalid count with filter
+# duplicates
+#  avg_length
+#  cross
+#  distribution
+#  duplicate_count
+#  duplicate_percent
+#  failed rows
+#  freshness
+#  group by
+#  group evolution
+#  invalid_count
+#  invalid_percent
+#  max_length
+#  min_length
+#  missing_count
+#  missing_percent
+#  percentile
+#  reconciliation
+#  reference
+#  schema
+#  schema evolution
+#  stddev_pop
+#  stddev_samp
+#  user-defined
+#  variance
+#  var_pop
+#  var_samp

cmd/check.go

@@ -2,14 +2,14 @@ package cmd
 
 import (
 	"dbq/internal"
-	"fmt"
+	"log"
+	"strings"
 
 	"github.com/spf13/cobra"
 )
 
 func NewCheckCommand(app internal.DbqApp) *cobra.Command {
 	var checksFile string
-	var dataSource string
 
 	cmd := &cobra.Command{
 		Use:   "check",
@@ -20,15 +20,36 @@ which outlines the rules and constraints that the data within the dataset should
 By automating these checks, you can proactively identify and address data quality issues, ensuring that your datasets meet the required standards for analysis and decision-making.
 `,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			fmt.Println("Reading checks from " + checksFile)
-			if dataSource != "" {
-				fmt.Println("Data source is not empty: " + dataSource)
+			log.Printf("Reading checks configuration file: %s \n", checksFile)
+
+			checksCfg, err := internal.LoadChecksConfig(checksFile)
+			if err != nil {
+				log.Printf("Failed to read checks configuration: %s", err.Error())
 			}
+
+			for i, ruleSet := range checksCfg.Validations {
+				log.Printf("Running check for %s [%d/%d]", ruleSet.Dataset, i+1, len(checksCfg.Validations))
+
+				// todo: validation
+				parts := strings.Split(ruleSet.Dataset, "@")
+				dataSourceId := parts[0]
+				dataSet := parts[1] // todo: parse list
+
+				dataSource := app.FindDataSourceById(dataSourceId)
+
+				for _, check := range ruleSet.Checks {
+					_, err := app.RunCheck(&check, dataSource, dataSet, ruleSet.Where)
+					if err != nil {
+						log.Printf("Failed to run check: %s", err.Error())
+					}
+					// todo: act on check result
+				}
+			}
+
 			return nil
 		},
 	}
 
-	cmd.Flags().StringVarP(&dataSource, "datasource", "d", "", "Datasource")
 	cmd.Flags().StringVarP(&checksFile, "checks", "c", "", "Validation checks")
 	_ = cmd.MarkFlagRequired("checks")
 

dbq.yaml

@@ -9,9 +9,7 @@ datasources:
         password: changeme
         database: default
       datasets:
-        - default.my_first_table
-        - default.trips_small
-        - uk.uk_price_paid
+        - nyc_taxi.trips_small
     - id: pgsql
       type: postgres
       configuration:

internal/app.go

@@ -17,6 +17,7 @@ type DbqApp interface {
 	GetDbqConfig() *DbqConfig
 	SaveDbqConfig() error
 	FindDataSourceById(srcId string) *DataSource
+	RunCheck(check *Check, dataSource *DataSource, dataSet string, defaultWhere string) (string, error)
 }
 
 type DbqAppImpl struct {
@@ -94,6 +95,14 @@ func (app *DbqAppImpl) FindDataSourceById(srcId string) *DataSource {
 	return nil
 }
 
+func (app *DbqAppImpl) RunCheck(check *Check, dataSource *DataSource, dataSet string, defaultWhere string) (string, error) {
+	cnn, err := getDbqConnector(*dataSource)
+	if err != nil {
+		return "", err
+	}
+	return cnn.RunCheck(check, dataSet, defaultWhere)
+}
+
 func initConfig(dbqConfigPath string) *DbqConfig {
 	v := viper.New()
 

internal/checks_config.go

@@ -11,25 +11,17 @@ type ChecksConfig struct {
 	Validations []Validation `yaml:"validations"`
 }
 
-type ConfigDetails struct {
-	Host     string `yaml:"host"`
-	Port     int    `yaml:"port"`
-	Username string `yaml:"username"`
-	Password string `yaml:"password"`
-	Database string `yaml:"database,omitempty"`
-}
-
 type Validation struct {
 	Dataset string  `yaml:"dataset"`
+	Where   string  `yaml:"where,omitempty"` // Optional where clause
 	Checks  []Check `yaml:"checks"`
 }
 
 type Check struct {
-	ID          string                 `yaml:"id"`
-	Description string                 `yaml:"description,omitempty"`
-	Severity    string                 `yaml:"severity"`
-	Type        string                 `yaml:"type"`
-	Params      map[string]interface{} `yaml:"params"`
+	ID          string `yaml:"id"`
+	Description string `yaml:"description,omitempty"` // Optional
+	Severity    string `yaml:"severity,omitempty"`    // Optional (error, warn, info)
+	Query       string `yaml:"query,omitempty"`       // Optional raw query
 }
 
 func LoadChecksConfig(fileName string) (*ChecksConfig, error) {

internal/clickhouse.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"log"
+	"regexp"
 	"strings"
 	"time"
 )
@@ -243,6 +244,116 @@ func (c *ClickhouseDbqConnector) ProfileDataSet(dataSet string) (*TableMetrics,
 	return metrics, nil
 }
 
+func (c *ClickhouseDbqConnector) RunCheck(check *Check, dataSet string, defaultWhere string) (string, error) {
+	if c.cnn == nil {
+		return "", 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())
+	}
+
+	log.Printf("Executing SQL for (%s): %s", check.ID, query)
+
+	startTime := time.Now()
+	rows, err := c.cnn.Query(context.Background(), query)
+	if err != nil {
+		return "", fmt.Errorf("failed to query database: %w", err)
+	}
+	defer rows.Close()
+	elapsed := time.Since(startTime).Milliseconds()
+
+	for rows.Next() {
+		var checkPassed bool
+		if err := rows.Scan(&checkPassed); err != nil {
+			return "", fmt.Errorf("failed to scan row: %w", err)
+		}
+
+		log.Printf("Result is: %t (%d ms)", checkPassed, elapsed)
+		log.Printf("---")
+	}
+
+	if err = rows.Err(); err != nil {
+		return "", fmt.Errorf("error occurred during row iteration: %w", err)
+	}
+
+	return "", nil
+}
+
+func generateDataCheckQuery(check *Check, dataSet string, whereClause string) (string, error) {
+	var sqlQuery string
+
+	// handle raw_query first
+	if check.ID == CheckTypeRawQuery {
+		if check.Query == "" {
+			return "", fmt.Errorf("check with id 'raw_query' requires a 'query' field")
+		}
+		sqlQuery = strings.ReplaceAll(check.Query, "{{table}}", dataSet)
+
+		if whereClause != "" {
+			// todo: more sophisticated check might be needed
+			if strings.Contains(strings.ToLower(sqlQuery), " where ") {
+				sqlQuery = fmt.Sprintf("%s and (%s)", sqlQuery, whereClause)
+			} else {
+				sqlQuery = fmt.Sprintf("%s where %s", sqlQuery, whereClause)
+			}
+		}
+
+		return sqlQuery, nil
+	}
+
+	isAggFunction := startWithAnyOf([]string{
+		"min", "max", "avg", "stddevPop", "sum",
+	}, check.ID)
+
+	var checkExpression string
+	switch {
+	case strings.HasPrefix(check.ID, "row_count"):
+		// format "row_count <operator> <value>"
+		parts := strings.Fields(check.ID)
+		if len(parts) != 3 {
+			return "", fmt.Errorf("invalid format for row_count check: %s", check.ID)
+		}
+		checkExpression = fmt.Sprintf("count() %s %s", parts[1], parts[2])
+
+	case strings.HasPrefix(check.ID, "null_count"):
+		// format "null_count(<column_name>) <operator> <value>"
+		re := regexp.MustCompile(`null_count\((.*?)\)\s*(==|!=|>|<|>=|<=)\s*(\d+)`)
+		matches := re.FindStringSubmatch(check.ID)
+		if len(matches) != 4 {
+			return "", fmt.Errorf("invalid format for null_count check: %s", check.ID)
+		}
+
+		column := matches[1]
+		operator := matches[2]
+		value := matches[3]
+		checkExpression = fmt.Sprintf("countIf(%s IS NULL) %s %s", column, operator, value)
+
+	case isAggFunction:
+		// format: <func>(<column_name>) <operator> <value>
+		re := regexp.MustCompile(`^(min|max|avg|stddevPop|sum)\(([^)]+)\)\s+(==|>=|<=|>|<)\s+(.*)$`)
+		matches := re.FindStringSubmatch(check.ID)
+		if len(matches) < 4 {
+			return "", fmt.Errorf("invalid format for aggregation function check: %s", check.ID)
+		}
+		checkExpression = fmt.Sprintf("%s", matches[0])
+
+	default:
+		// Assume the ID itself is a valid boolean expression if no specific pattern matches
+		// This is less robust but covers simple cases.
+		log.Printf("Warning: Check ID '%s' did not match known patterns. Assuming it's a direct SQL boolean expression.", check.ID)
+		checkExpression = check.ID
+	}
+
+	sqlQuery = fmt.Sprintf("select %s from %s", checkExpression, dataSet)
+	if whereClause != "" {
+		sqlQuery = fmt.Sprintf("%s where %s", sqlQuery, whereClause)
+	}
+
+	return sqlQuery, nil
+}
+
 // isNumericCHType checks if a ClickHouse data type string represents a numeric type
 // that supports standard aggregate functions like min, max, avg, stddev
 func isNumericCHType(dataType string) bool {
@@ -260,3 +371,12 @@ func isStringCHType(dataType string) bool {
 	return strings.HasPrefix(dataType, "string") ||
 		strings.HasPrefix(dataType, "fixedstring")
 }
+
+func startWithAnyOf(prefixes []string, s string) bool {
+	for _, prefix := range prefixes {
+		if strings.HasPrefix(s, prefix) {
+			return true
+		}
+	}
+	return false
+}

internal/dbq_config.go

@@ -18,6 +18,14 @@ type DataSource struct {
 	Datasets      []string      `yaml:"datasets"`
 }
 
+type ConfigDetails struct {
+	Host     string `yaml:"host"`
+	Port     int    `yaml:"port"`
+	Username string `yaml:"username"`
+	Password string `yaml:"password"`
+	Database string `yaml:"database,omitempty"`
+}
+
 func LoadDbqSetting(fileName string) (*DbqConfig, error) {
 	file, err := os.Open(fileName)
 	defer file.Close()