various fixes

Author: sholokhov

adapters/clickhouse_adapter.go

@@ -233,7 +233,7 @@ func (a *ClickhouseDbqDataSourceAdapter) InterpretDataQualityCheck(check *dbqcor
 func (a *ClickhouseDbqDataSourceAdapter) ExecuteQuery(ctx context.Context, query string) (interface{}, error) {
 	rows, err := a.cnn.Query(ctx, query)
 	if err != nil {
-		return "", fmt.Errorf("failed to execute query for check: %v", err)
+		return "", fmt.Errorf("failed to execute query for check: %w", err)
 	}
 	defer rows.Close()
 
@@ -255,7 +255,7 @@ func (a *ClickhouseDbqDataSourceAdapter) ExecuteQuery(ctx context.Context, query
 	}
 
 	if err = rows.Err(); err != nil {
-		return "", fmt.Errorf("failed to scan result for check: %v", err)
+		return "", fmt.Errorf("failed to scan result for check: %w", err)
 	}
 
 	return queryResult, nil

adapters/mysql_adapter.go

@@ -232,19 +232,19 @@ func (a *MysqlDbqDataSourceAdapter) InterpretDataQualityCheck(check *dbqcore.Dat
 func (a *MysqlDbqDataSourceAdapter) ExecuteQuery(ctx context.Context, query string) (interface{}, error) {
 	rows, err := a.db.QueryContext(ctx, query)
 	if err != nil {
-		return "", fmt.Errorf("failed to execute query for check: %v", err)
+		return "", fmt.Errorf("failed to execute query for check: %w", err)
 	}
 	defer rows.Close()
 
 	var queryResult interface{}
 	for rows.Next() {
 		if err := rows.Scan(&queryResult); err != nil {
-			return "", fmt.Errorf("failed to scan result for check: %v", err)
+			return "", fmt.Errorf("failed to scan result for check: %w", err)
 		}
 	}
 
 	if err = rows.Err(); err != nil {
-		return "", fmt.Errorf("failed to scan result for check: %v", err)
+		return "", fmt.Errorf("failed to scan result for check: %w", err)
 	}
 
 	return queryResult, nil

adapters/postgresql_adapter.go

@@ -232,19 +232,19 @@ func (a *PostgresqlDbqDataSourceAdapter) InterpretDataQualityCheck(check *dbqcor
 func (a *PostgresqlDbqDataSourceAdapter) ExecuteQuery(ctx context.Context, query string) (interface{}, error) {
 	rows, err := a.db.QueryContext(ctx, query)
 	if err != nil {
-		return "", fmt.Errorf("failed to execute query for check: %v", err)
+		return "", fmt.Errorf("failed to execute query for check: %w", err)
 	}
 	defer rows.Close()
 
 	var queryResult interface{}
 	for rows.Next() {
 		if err := rows.Scan(&queryResult); err != nil {
-			return "", fmt.Errorf("failed to scan result for check: %v", err)
+			return "", fmt.Errorf("failed to scan result for check: %w", err)
 		}
 	}
 
 	if err = rows.Err(); err != nil {
-		return "", fmt.Errorf("failed to scan result for check: %v", err)
+		return "", fmt.Errorf("failed to scan result for check: %w", err)
 	}
 
 	return queryResult, nil

check_parser.go

@@ -43,6 +43,10 @@ type CheckExpression struct {
 }
 
 var (
+	betweenRegex      = regexp.MustCompile(`^(\w+)(?:\((.*?)\))?\s+between\s+(.+)\s+and\s+(.+)$`)
+	operatorRegex     = regexp.MustCompile(`^(\w+)(?:\((.*?)\))?\s*([<>=!]+)\s*(.+)$`)
+	functionOnlyRegex = regexp.MustCompile(`^(\w+)(?:\((.*?)\))?$`)
+
 	tableScopeFunctions = map[string]bool{
 		"row_count": true,
 		"raw_query": true,
@@ -76,10 +80,6 @@ func ParseCheckExpression(expression string) (*CheckExpression, error) {
 		FunctionParameters: []string{},
 	}
 
-	betweenRegex := regexp.MustCompile(`^(\w+)(?:\((.*?)\))?\s+between\s+(.+)\s+and\s+(.+)$`)
-	operatorRegex := regexp.MustCompile(`^(\w+)(?:\((.*?)\))?\s*([<>=!]+)\s*(.+)$`)
-	functionOnlyRegex := regexp.MustCompile(`^(\w+)(?:\((.*?)\))?$`)
-
 	if matches := betweenRegex.FindStringSubmatch(expression); matches != nil {
 		check.FunctionName = matches[1]
 		check.Operator = "between"

checks_cfg.go

@@ -156,10 +156,10 @@ func (c *DataQualityCheck) UnmarshalYAML(node *yaml.Node) error {
 
 func LoadChecksFileConfig(fileName string) (*ChecksFileConfig, error) {
 	file, err := os.Open(fileName)
-	defer file.Close()
 	if err != nil {
 		return nil, err
 	}
+	defer file.Close()
 
 	var cfg ChecksFileConfig
 	decoder := yaml.NewDecoder(file)

cnn/clickhouse_cnn.go

@@ -16,6 +16,7 @@ package cnn
 
 import (
 	"crypto/tls"
+	"fmt"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
@@ -26,7 +27,7 @@ func NewClickhouseConnection(connectionCfg dbqcore.ConnectionConfig, poolSize in
 	proto, tlsCfg := clickhouseProtocol(connectionCfg.Protocol)
 
 	cnn, err := clickhouse.Open(&clickhouse.Options{
-		Addr: []string{connectionCfg.Host},
+		Addr: []string{fmt.Sprintf("%s:%d", connectionCfg.Host, connectionCfg.Port)},
 		Auth: clickhouse.Auth{
 			Database: connectionCfg.Database,
 			Username: connectionCfg.Username,

connectors/clickhouse_connector.go

@@ -34,9 +34,13 @@ func NewClickhouseDbqConnector(cnn driver.Conn, logger *slog.Logger) dbqcore.Dbq
 }
 
 func (c *ClickhouseDbqConnector) Ping(ctx context.Context) (string, error) {
+	if err := c.cnn.Ping(ctx); err != nil {
+		return "", fmt.Errorf("clickhouse ping failed: %w", err)
+	}
+
 	serverVersion, err := c.cnn.ServerVersion()
 	if err != nil {
-		return "", err
+		return "", fmt.Errorf("failed to get server version: %w", err)
 	}
 
 	return serverVersion.String(), nil

dbq_validator.go

@@ -18,10 +18,13 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"math"
 	"strconv"
 	"time"
 )
 
+const floatEqualityEpsilon = 1e-9
+
 // DataQualityCheckType represents the type of data quality check.
 type DataQualityCheckType string
 
@@ -138,8 +141,9 @@ func (d DbqDataValidatorImpl) RunCheck(ctx context.Context, adapter DbqDataSourc
 				result.Pass = true
 			}
 		} else {
-			// Unknown schema check type, default to pass
-			result.Pass = true
+			// Unknown schema check type, fail-safe
+			result.Pass = false
+			result.Error = fmt.Sprintf("unknown schema check type for expression: %s", check.Expression)
 		}
 	} else {
 		// Regular checks use the existing validation logic
@@ -186,9 +190,9 @@ func (d DbqDataValidatorImpl) validateResult(queryResult interface{}, parsedChec
 	case "!=", "<>":
 		return d.validateNotEqual(actualValue, parsedCheck.ThresholdValue)
 	default:
-		d.logger.Warn("Unknown operator, defaulting to true",
+		d.logger.Error("Unknown operator, failing check",
 			"operator", parsedCheck.Operator)
-		return true
+		return false
 	}
 }
 
@@ -297,7 +301,7 @@ func (d DbqDataValidatorImpl) validateLessThanOrEqual(actualValue float64, thres
 	return actualValue <= threshold
 }
 
-// validateEqual checks if actual == threshold
+// validateEqual checks if actual == threshold (with epsilon tolerance for float comparison)
 func (d DbqDataValidatorImpl) validateEqual(actualValue float64, thresholdValue interface{}) bool {
 	threshold, err := d.convertToFloat64(thresholdValue)
 	if err != nil {
@@ -306,10 +310,10 @@ func (d DbqDataValidatorImpl) validateEqual(actualValue float64, thresholdValue
 			"error", err)
 		return false
 	}
-	return actualValue == threshold
+	return math.Abs(actualValue-threshold) < floatEqualityEpsilon
 }
 
-// validateNotEqual checks if actual != threshold
+// validateNotEqual checks if actual != threshold (with epsilon tolerance for float comparison)
 func (d DbqDataValidatorImpl) validateNotEqual(actualValue float64, thresholdValue interface{}) bool {
 	threshold, err := d.convertToFloat64(thresholdValue)
 	if err != nil {
@@ -318,7 +322,7 @@ func (d DbqDataValidatorImpl) validateNotEqual(actualValue float64, thresholdVal
 			"error", err)
 		return false
 	}
-	return actualValue != threshold
+	return math.Abs(actualValue-threshold) >= floatEqualityEpsilon
 }
 
 // convertToFloat64 converts various types to float64

dbq_validator_test.go

@@ -273,7 +273,7 @@ func TestDbqDataValidatorImpl_validateResult(t *testing.T) {
 				Operator:       "~=",
 				ThresholdValue: 100,
 			},
-			expectedPass: true,
+			expectedPass: false,
 		},
 	}