fix(clickhouse): exact microsecond->Time64 conversion on snapshot path (#4378)

Author: ilidemi

flow/connectors/clickhouse/table_function.go

@@ -76,8 +76,8 @@ func timeFieldExpressionConverter(
 	}
 
 	// QValueTime is stored as time-micro logical type in Avro, toTime64 accepts a
-	// fractional second, so conversion is necessary
-	return fmt.Sprintf("toTime64(%s / 1000000.0, 6)", sourceFieldIdentifier), nil
+	// fractional second, so conversion is necessary. Trip through Decimal64 to preserve precision.
+	return fmt.Sprintf("toTime64(toDecimal64(%s, 6) / 1000000, 6)", sourceFieldIdentifier), nil
 }
 
 var defaultFieldExpressionConverters = []fieldExpressionConverter{

flow/e2e/clickhouse.go

@@ -175,15 +175,15 @@ func (s ClickHouseSuite) GetRows(table string, cols string) (*model.QRecordBatch
 
 	batch := &model.QRecordBatch{}
 	colTypes := rows.ColumnTypes()
-	row := make([]any, 0, len(colTypes))
+	scanTypes := make([]reflect.Type, len(colTypes))
 	tableSchema, err := connclickhouse.GetTableSchemaForTable(&protos.TableMapping{SourceTableIdentifier: table}, colTypes)
 	if err != nil {
 		return nil, err
 	}
 
 	for idx, ty := range colTypes {
 		fieldDesc := tableSchema.Columns[idx]
-		row = append(row, reflect.New(ty.ScanType()).Interface())
+		scanTypes[idx] = ty.ScanType()
 		batch.Schema.Fields = append(batch.Schema.Fields, types.QField{
 			Name:      ty.Name(),
 			Type:      types.QValueKind(fieldDesc.Type),
@@ -194,6 +194,13 @@ func (s ClickHouseSuite) GetRows(table string, cols string) (*model.QRecordBatch
 	}
 
 	for rows.Next() {
+		// Allocate fresh scan targets per row: clickhouse-go does not reset a
+		// nullable destination to nil when scanning a NULL, so a reused buffer
+		// would make a NULL row inherit the previous row's value.
+		row := make([]any, len(scanTypes))
+		for idx, st := range scanTypes {
+			row[idx] = reflect.New(st).Interface()
+		}
 		if err := rows.Scan(row...); err != nil {
 			return nil, err
 		}

flow/e2e/clickhouse_test.go

@@ -1348,6 +1348,9 @@ func (s ClickHouseSuite) Test_Time64() {
 	`, srcFullName)))
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
 		`INSERT INTO %s (t, t_nullable, t_nullable_2) VALUES ('14:21:00', '08:30:00.123456', NULL)`, srcFullName)))
+	// Regression test for precise conversion - these values were losing 1us when tripped through floating point
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (t, t_nullable, t_nullable_2) VALUES ('00:00:00.000249', '00:00:00.000251', '00:00:00.000493')`, srcFullName)))
 
 	connectionGen := FlowConnectionGenerationConfig{
 		FlowJobName:      s.attachSuffix(srcTableName),