Fix float32 handling and add tests for real type (#1221)

Fixes avro conversion for float32 to bigquery, adds tests for real data type which gets translated to float32
PeerDB-io · Feb 7, 2024 · 9d7e4b6 · 9d7e4b6
1 parent 14dc0be
commit 9d7e4b6
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 2 deletions.
diff --git a/flow/e2e/test_utils.go b/flow/e2e/test_utils.go
@@ -265,6 +265,9 @@ func CreateTableForQRep(conn *pgx.Conn, suffix string, tableName string) error {
 		"geography_linestring geography(linestring)",
 		"geometry_polygon geometry(polygon)",
 		"geography_polygon geography(polygon)",
+		"myreal REAL",
+		"myreal2 REAL",
+		"myreal3 REAL",
 	}
 	tblFieldStr := strings.Join(tblFields, ",")
 	var pgErr *pgconn.PgError
@@ -322,7 +325,10 @@ func PopulateSourceTable(conn *pgx.Conn, suffix string, tableName string, rowCou
 							CURRENT_DATE, CURRENT_TIME,'happy', '"a"=>"b"','POINT(1 2)','POINT(40.7128 -74.0060)',
 						'LINESTRING(0 0, 1 1, 2 2)',
 						'LINESTRING(-74.0060 40.7128, -73.9352 40.7306, -73.9123 40.7831)',
-						'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))','POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))'
+						'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))','POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))',
+						3.14159,
+						1,
+						1.0
 					)`,
 			id, uuid.New().String(), uuid.New().String(),
 			uuid.New().String(), uuid.New().String(), uuid.New().String(), uuid.New().String())
@@ -339,7 +345,10 @@ func PopulateSourceTable(conn *pgx.Conn, suffix string, tableName string, rowCou
 					asset_id, status, transaction_id, settled_at, reference_id,
 					settle_at, settlement_delay_reason, f1, f2, f3, f4, f5, f6, f7, f8, f9, f10, f11, f12, f13, my_date,
 					my_time, my_mood, myh,
-					"geometryPoint", geography_point,geometry_linestring, geography_linestring,geometry_polygon, geography_polygon
+					"geometryPoint", geography_point,geometry_linestring, geography_linestring,geometry_polygon, geography_polygon,
+					myreal,
+					myreal2,
+					myreal3
 			) VALUES %s;
 	`, suffix, tableName, strings.Join(rows, ",")))
 	if err != nil {
@@ -476,6 +485,9 @@ func GetOwnersSchema() *model.QRecordSchema {
 			{Name: "geography_point", Type: qvalue.QValueKindGeography, Nullable: true},
 			{Name: "geography_linestring", Type: qvalue.QValueKindGeography, Nullable: true},
 			{Name: "geography_polygon", Type: qvalue.QValueKindGeography, Nullable: true},
+			{Name: "myreal", Type: qvalue.QValueKindFloat32, Nullable: true},
+			{Name: "myreal2", Type: qvalue.QValueKindFloat32, Nullable: true},
+			{Name: "myreal3", Type: qvalue.QValueKindFloat32, Nullable: true},
 		},
 	}
 }

diff --git a/flow/model/qvalue/avro_converter.go b/flow/model/qvalue/avro_converter.go
@@ -280,6 +280,9 @@ func (c *QValueAvroConverter) ToAvroValue() (interface{}, error) {
 		}
 		return c.processNullableUnion("string", c.Value.Value)
 	case QValueKindFloat32:
+		if c.TargetDWH == QDWHTypeBigQuery {
+			return c.processNullableUnion("double", c.Value.Value)
+		}
 		return c.processNullableUnion("float", c.Value.Value)
 	case QValueKindFloat64:
 		if c.TargetDWH == QDWHTypeSnowflake || c.TargetDWH == QDWHTypeBigQuery {

diff --git a/flow/model/qvalue/qvalue.go b/flow/model/qvalue/qvalue.go
@@ -129,6 +129,9 @@ func compareInt64(value1, value2 interface{}) bool {
 }
 
 func compareFloat32(value1, value2 interface{}) bool {
+	if value1 == nil && value2 == nil {
+		return true
+	}
 	float1, ok1 := getFloat32(value1)
 	float2, ok2 := getFloat32(value2)
 	return ok1 && ok2 && float1 == float2