lint

Author: serprex

flow/connectors/postgres/postgres.go

@@ -150,20 +150,20 @@ func (c *PostgresConnector) fetchCustomTypeMapping(ctx context.Context, version
 		c.customTypeMapping = customTypeMapping
 
 		if version >= shared.InternalVersion_CompositeTypeAsTuple {
-		var compositeTypeNames []string
-		for _, typeData := range customTypeMapping {
-			if typeData.Type == 'c' && typeData.Delim == 0 { // Only composite types
-				compositeTypeNames = append(compositeTypeNames, typeData.Name)
+			var compositeTypeNames []string
+			for _, typeData := range customTypeMapping {
+				if typeData.Type == 'c' && typeData.Delim == 0 { // Only composite types
+					compositeTypeNames = append(compositeTypeNames, typeData.Name)
+				}
 			}
+			types, err := c.conn.LoadTypes(ctx, compositeTypeNames)
+			if err != nil {
+				c.logger.Error("failed to load composite types",
+					slog.Any("error", err), slog.Any("composite_type_names", compositeTypeNames))
+				return nil, fmt.Errorf("failed to load composite types: %w", err)
+			}
+			c.typeMap.RegisterTypes(types)
 		}
-		types, err := c.conn.LoadTypes(ctx, compositeTypeNames)
-		if err != nil {
-			c.logger.Error("failed to load composite types",
-				slog.Any("error", err), slog.Any("composite_type_names", compositeTypeNames))
-			return nil, fmt.Errorf("failed to load composite types: %w", err)
-		}
-		c.typeMap.RegisterTypes(types)
-	}
 	}
 	return c.customTypeMapping, nil
 }
@@ -941,11 +941,12 @@ func (c *PostgresConnector) GetSelectedColumns(
 }
 
 func (c *PostgresConnector) getCompositeTypeDetails(ctx context.Context, system protos.TypeSystem, version uint32,
-	customTypeMapping map[uint32]shared.CustomDataType, OID uint32) ([]*protos.FieldDescription, error) {
+	customTypeMapping map[uint32]shared.CustomDataType, oid uint32,
+) ([]*protos.FieldDescription, error) {
 	result := make([]*protos.FieldDescription, 0)
-	subfields, err := shared.GetCompositeDataTypeDetails(ctx, c.conn, OID)
+	subfields, err := shared.GetCompositeDataTypeDetails(ctx, c.conn, oid)
 	if err != nil {
-		return nil, fmt.Errorf("error getting composite data type details for %d: %w", OID, err)
+		return nil, fmt.Errorf("error getting composite data type details for %d: %w", oid, err)
 	}
 	for _, subfield := range subfields {
 		var subColType string

flow/connectors/postgres/qrep_query_executor.go

@@ -101,7 +101,7 @@ func (qe *QRepQueryExecutor) buildQFieldFromOID(name string, oid uint32, typeMod
 			}
 		}
 		qe.logger.Error("[pg_query_executor] type not found for oid or not a composite type",
-			slog.String("type_oid", fmt.Sprintf("%d", oid)),
+			slog.Uint64("type_oid", uint64(oid)),
 			slog.String("type_name", name))
 		return types.QField{
 			Name:     name,
@@ -123,7 +123,7 @@ func (qe *QRepQueryExecutor) buildQFieldFromOID(name string, oid uint32, typeMod
 			}
 		}
 		qe.logger.Error("[pg_query_executor] type not found for oid or not an array composite type",
-			slog.String("type_oid", fmt.Sprintf("%d", oid)),
+			slog.Uint64("type_oid", uint64(oid)),
 			slog.String("type_name", name))
 		return types.QField{
 			Name:     name,

flow/connectors/postgres/qvalue_convert.go

@@ -765,7 +765,12 @@ func (c *PostgresConnector) parseFieldFromPostgresOID(
 	return nil, fmt.Errorf("failed to parse value %v (%T) into QValueKind %v", value, value, qvalueKind)
 }
 
-func (c *PostgresConnector) compositeToQValue(oid uint32, value any, customTypeMapping map[uint32]shared.CustomDataType, version uint32) (types.QValue, error) {
+func (c *PostgresConnector) compositeToQValue(
+	oid uint32,
+	value any,
+	customTypeMapping map[uint32]shared.CustomDataType,
+	version uint32,
+) (types.QValue, error) {
 	typ, ok := c.typeMap.TypeForOID(oid)
 	if !ok {
 		return nil, fmt.Errorf("composite type OID %d not found in typeMap", oid)

flow/model/qvalue/avro_converter.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"log/slog"
 	"math/big"
@@ -115,8 +116,8 @@ func GetAvroSchemaFromQValueKind(
 	case types.QValueKindHStore, types.QValueKindJSON, types.QValueKindJSONB:
 		return avro.NewPrimitiveSchema(avro.String, nil), nil
 	case types.QValueKindComposite:
-		if qField.SubFields == nil || len(qField.SubFields) == 0 {
-			return nil, fmt.Errorf("composite subfields are required for composite type but none provided")
+		if len(qField.SubFields) == 0 {
+			return nil, errors.New("composite subfields are required for composite type but none provided")
 		}
 
 		avroFields := make([]*avro.Field, 0, len(qField.SubFields))
@@ -143,8 +144,8 @@ func GetAvroSchemaFromQValueKind(
 
 		return avro.NewRecordSchema(qField.Name, "", avroFields)
 	case types.QValueKindArrayComposite:
-		if qField.SubFields == nil || len(qField.SubFields) != 1 {
-			return nil, fmt.Errorf("array composite subfields must contain exactly one field definition")
+		if len(qField.SubFields) != 1 {
+			return nil, errors.New("array composite subfields must contain exactly one field definition")
 		}
 		fieldSchema, err := GetAvroSchemaFromQValueKind(
 			ctx, env, *qField.SubFields[0], targetDWH)
@@ -682,7 +683,6 @@ func (c *QValueAvroConverter) processComposite(
 			ctx, subfieldValue, subfield, c.TargetDWH, c.logger,
 			c.UnboundedNumericAsString, c.Stat, binaryFormat,
 		)
-
 		if err != nil {
 			return nil, fmt.Errorf("failed to convert composite field %s: %w", subfield.Name, err)
 		}

flow/model/qvalue/kind.go

@@ -91,7 +91,7 @@ func ToDWHColumnType(
 			colType = "JSON"
 		} else if kind == types.QValueKindComposite {
 			colType = "Tuple("
-			if column.Composite == nil || len(column.Composite) == 0 {
+			if len(column.Composite) == 0 {
 				return "", fmt.Errorf("composite type is nil or empty for column %s", column.Name)
 			}
 			for _, subfield := range column.Composite {
@@ -110,10 +110,11 @@ func ToDWHColumnType(
 			}
 			colType = strings.TrimSuffix(colType, ", ") + ")"
 		} else if kind == types.QValueKindArrayComposite {
-			if column.Composite == nil || len(column.Composite) != 1 {
+			if len(column.Composite) != 1 {
 				return "", fmt.Errorf("composite array type %s must have exactly 1 subfield", column.Name)
 			}
-			elementType, err := ToDWHColumnType(ctx, types.QValueKind(column.Composite[0].Type), env, dwhType, dwhVersion, column.Composite[0], nullableEnabled)
+			elementType, err := ToDWHColumnType(ctx, types.QValueKind(column.Composite[0].Type),
+				env, dwhType, dwhVersion, column.Composite[0], nullableEnabled)
 			if err != nil {
 				return "", fmt.Errorf("failed to get DWH column type for composite field %s: %w", column.Composite[0].Name, err)
 			}

flow/shared/types/qschema.go

@@ -8,10 +8,10 @@ import (
 type QField struct {
 	Name      string
 	Type      QValueKind
+	SubFields []*QField
 	Precision int16
 	Scale     int16
 	Nullable  bool
-	SubFields []*QField // For composite types, this holds the sub-fields
 }
 
 type QRecordSchema struct {

flow/shared/types/qvalue.go

@@ -862,9 +862,11 @@ type QValueArrayComposite struct {
 func (QValueArrayComposite) Kind() QValueKind {
 	return QValueKindArrayComposite
 }
+
 func (v QValueArrayComposite) Value() any {
 	return v.Val
 }
+
 func (v QValueArrayComposite) LValue(ls *lua.LState) lua.LValue {
 	table := ls.NewTable()
 	for i, composite := range v.Val {