sql: Fix incorrect handling of types (elastic#41607)

CHANGELOG.next.asciidoc

@@ -64,6 +64,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Added back `elasticsearch.node.stats.jvm.mem.pools.*` to the `node_stats` metricset {pull}40571[40571]
 - Add GCP organization and project details to ECS cloud fields. {pull}40461[40461]
 - Add support for specifying a custom endpoint for GCP service clients. {issue}40848[40848] {pull}40918[40918]
+- Fix incorrect handling of types in SQL module. {issue}40090[40090] {pull}41607[41607]
 
 *Osquerybeat*
 

metricbeat/helper/sql/sql.go

@@ -21,11 +21,11 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-
 	"strconv"
-	"strings"
 	"time"
 
+	"strings"
+
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
@@ -114,7 +114,7 @@ func (d *DbClient) fetchTableMode(rows sqlRow) ([]mapstr.M, error) {
 	return rr, nil
 }
 
-// fetchTableMode scan the rows and publishes the event for querys that return the response in a table format.
+// FetchVariableMode executes the provided SQL query and returns the results in a key/value format.
 func (d *DbClient) FetchVariableMode(ctx context.Context, q string) (mapstr.M, error) {
 	rows, err := d.QueryContext(ctx, q)
 	if err != nil {
@@ -123,7 +123,7 @@ func (d *DbClient) FetchVariableMode(ctx context.Context, q string) (mapstr.M, e
 	return d.fetchVariableMode(rows)
 }
 
-// fetchVariableMode scan the rows and publishes the event for querys that return the response in a key/value format.
+// fetchVariableMode scans the provided SQL rows and returns the results in a key/value format.
 func (d *DbClient) fetchVariableMode(rows sqlRow) (mapstr.M, error) {
 	data := mapstr.M{}
 
@@ -167,24 +167,25 @@ func ReplaceUnderscores(ms mapstr.M) mapstr.M {
 }
 
 func getValue(pval *interface{}) interface{} {
+	if pval == nil {
+		return nil
+	}
+
 	switch v := (*pval).(type) {
-	case nil, bool:
+	case nil, bool, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64, string, []interface{}:
 		return v
 	case []byte:
-		s := string(v)
-		num, err := strconv.ParseFloat(s, 64)
-		if err == nil {
-			return num
-		}
-		return s
+		return string(v)
 	case time.Time:
 		return v.Format(time.RFC3339Nano)
-	case []interface{}:
-		return v
 	default:
+		// For any other types, convert to string and try to parse as number
 		s := fmt.Sprint(v)
-		num, err := strconv.ParseFloat(s, 64)
-		if err == nil {
+		if len(s) > 1 && s[0] == '0' && s[1] != '.' {
+			// Preserve string with leading zeros i.e., 00100 stays 00100
+			return s
+		}
+		if num, err := strconv.ParseFloat(s, 64); err == nil {
 			return num
 		}
 		return s

metricbeat/helper/sql/sql_test.go

@@ -22,7 +22,7 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"fmt"
-	"math"
+	"strconv"
 	"testing"
 	"time"
 
@@ -43,10 +43,10 @@ type mockVariableMode struct {
 }
 
 func (m *mockVariableMode) Scan(dest ...interface{}) error {
-	d1 := dest[0].(*string)
+	d1 := dest[0].(*string) //nolint:errcheck // false positive
 	*d1 = m.results[m.index].k
 
-	d2 := dest[1].(*interface{})
+	d2 := dest[1].(*interface{}) //nolint:errcheck // false positive
 	*d2 = m.results[m.index].v
 
 	m.index++
@@ -73,7 +73,7 @@ type mockTableMode struct {
 
 func (m *mockTableMode) Scan(dest ...interface{}) error {
 	for i, d := range dest {
-		d1 := d.(*interface{})
+		d1 := d.(*interface{}) //nolint:errcheck // false positive
 		*d1 = m.results[i].v
 	}
 
@@ -87,14 +87,20 @@ func (m *mockTableMode) Next() bool {
 }
 
 func (m *mockTableMode) Columns() ([]string, error) {
-	return []string{"hello", "integer", "signed_integer", "unsigned_integer", "float64", "float32", "null", "boolean", "array", "byte_array", "time"}, nil
+	cols := make([]string, len(m.results))
+	for i, r := range m.results {
+		cols[i] = r.k
+	}
+	return cols, nil
 }
 
 func (m mockTableMode) Err() error {
 	return nil
 }
 
 var results = []kv{
+	{k: "string", v: "000400"},
+	{k: "varchar", v: "00100"},
 	{k: "hello", v: "world"},
 	{k: "integer", v: int(10)},
 	{k: "signed_integer", v: int(-10)},
@@ -137,54 +143,65 @@ func TestFetchTableMode(t *testing.T) {
 }
 
 func checkValue(t *testing.T, res kv, ms mapstr.M) {
+	t.Helper()
+
+	actual := ms[res.k]
 	switch v := res.v.(type) {
-	case string, bool:
-		if ms[res.k] != v {
-			t.Fail()
+	case string, bool, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64:
+		if actual != v {
+			t.Errorf("key %q: expected %v (%T), got %v (%T)", res.k, v, v, actual, actual)
 		}
 	case nil:
-		if ms[res.k] != nil {
-			t.Fail()
-		}
-	case int:
-		if ms[res.k] != float64(v) {
-			t.Fail()
-		}
-	case uint:
-		if ms[res.k] != float64(v) {
-			t.Fail()
-		}
-	case float32:
-		if math.Abs(float64(ms[res.k].(float64)-float64(v))) > 1 {
-			t.Fail()
-		}
-	case float64:
-		if ms[res.k] != v {
-			t.Fail()
+		if actual != nil {
+			t.Errorf("key %q: expected nil, got %v (%T)", res.k, actual, actual)
 		}
 	case []interface{}:
+		actualSlice := actual.([]interface{})
+		if len(v) != len(actualSlice) {
+			t.Errorf("key %q: slice length mismatch: expected %d, got %d", res.k, len(v), len(actualSlice))
+			return
+		}
 		for i, val := range v {
-			if ms[res.k].([]interface{})[i] != val {
-				t.Fail()
+			if actualSlice[i] != val {
+				t.Errorf("key %q: slice mismatch at index %d: expected %v, got %v", res.k, i, val, actualSlice[i])
 			}
 		}
 	case []byte:
-		ar := ms[res.k].(string)
-		if ar != string(v) {
-			t.Fail()
+		actualStr := actual.(string)
+		if actualStr != string(v) {
+			t.Errorf("key %q: expected %q (string), got %q", res.k, string(v), actualStr)
 		}
 	case time.Time:
-		ar := ms[res.k].(string)
-		if v.Format(time.RFC3339Nano) != ar {
-			t.Fail()
+		actualStr := actual.(string)
+		expectedStr := v.Format(time.RFC3339Nano)
+		if expectedStr != actualStr {
+			t.Errorf("key %q: expected time %q, got %q", res.k, expectedStr, actualStr)
+		}
+	case CustomType:
+		// Handle custom types that should be converted to string
+		expectedStr := fmt.Sprint(v)
+		if num, err := strconv.ParseFloat(expectedStr, 64); err == nil {
+			if actual != num {
+				t.Errorf("key %q: expected %v (float64), got %v (%T)", res.k, num, actual, actual)
+			}
+		} else {
+			actualStr := actual.(string)
+			if actualStr != expectedStr {
+				t.Errorf("key %q: expected %q (string), got %q", res.k, expectedStr, actualStr)
+			}
 		}
 	default:
-		if ms[res.k] != res.v {
-			t.Fail()
+		if actual != res.v {
+			t.Errorf("key %q: expected %v (%T), got %v (%T)", res.k, res.v, res.v, actual, actual)
 		}
 	}
 }
 
+// CustomType for testing custom type handling
+type CustomType struct {
+	value string //nolint:unused // unused checker is buggy
+}
+
 func TestToDotKeys(t *testing.T) {
 	ms := mapstr.M{"key_value": "value"}
 	ms = ReplaceUnderscores(ms)

x-pack/metricbeat/module/sql/query/query.go

@@ -345,19 +345,28 @@ func inferTypeFromMetrics(ms mapstr.M) mapstr.M {
 
 	for k, v := range ms {
 		switch v.(type) {
-		case float64:
+		case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64:
 			numericMetrics[k] = v
 		case string:
 			stringMetrics[k] = v
 		case bool:
 			boolMetrics[k] = v
 		case nil:
-		// Ignore because a nil has no data type and thus cannot be indexed
+			// Ignore nil values as they cannot be indexed
+
+		// TODO: Handle []interface{} properly; for now it is going to "string" field.
+		// Keeping the behaviour as it is for now.
+		//
+		// case []interface{}:
+
 		default:
 			stringMetrics[k] = v
 		}
 	}
 
+	// TODO: Ideally the field keys should have in sync with ES types like s/bool/boolean, etc.
+	// But changing the field keys will be a breaking change. So, we are leaving it as it is.
+
 	if len(numericMetrics) > 0 {
 		ret["numeric"] = numericMetrics
 	}