You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: qrep_avro_sync.go
    path of file: ./repos/peerdb/flow/connectors/bigquery
    
    the code of the file until where you have to start completion: package connbigquery

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"strconv"
	"strings"
	"time"

	"cloud.google.com/go/bigquery"
	"go.temporal.io/sdk/activity"

	"github.com/PeerDB-io/peer-flow/connectors/utils"
	avro "github.com/PeerDB-io/peer-flow/connectors/utils/avro"
	"github.com/PeerDB-io/peer-flow/generated/protos"
	"github.com/PeerDB-io/peer-flow/model"
	"github.com/PeerDB-io/peer-flow/model/numeric"
	"github.com/PeerDB-io/peer-flow/model/qvalue"
	"github.com/PeerDB-io/peer-flow/shared"
)

type QRepAvroSyncMethod struct {
	connector   *BigQueryConnector
	gcsBucket   string
	flowJobName string
}

func NewQRepAvroSyncMethod(connector *BigQueryConnector, gcsBucket string,
	flowJobName string,
) *QRepAvroSyncMethod {
	return &QRepAvroSyncMethod{
		connector:   connector,
		gcsBucket:   gcsBucket,
		flowJobName: flowJobName,
	}
}

func (s *QRepAvroSyncMethod) SyncRecords(
	ctx context.Context,
	req *model.SyncRecordsRequest,
	rawTableName string,
	dstTableMetadata *bigquery.TableMetadata,
	syncBatchID int64,
	stream *model.QRecordStream,
	tableNameRowsMapping map[string]uint32,
) (*model.SyncResponse, error) {
	activity.RecordHeartbeat(ctx,
		fmt.Sprintf("Flow job %s: Obtaining Avro schema"+
			" for destination table %s and sync batch ID %d",
			req.FlowJobName, rawTableName, syncBatchID),
	)
	// You will need to define your Avro schema as a string
	avroSchema, err := DefineAvroSchema(rawTableName, dstTableMetadata, "", "")
	if err != nil {
		return nil, fmt.Errorf("failed to define Avro schema: %w", err)
	}

	stagingTable := fmt.Sprintf("%s_%s_staging", rawTableName, strconv.FormatInt(syncBatchID, 10))
	numRecords, err := s.writeToStage(ctx, strconv.FormatInt(syncBatchID, 10), rawTableName, avroSchema,
		&datasetTable{
			project: s.connector.projectID,
			dataset: s.connector.datasetID,
			table:   stagingTable,
		}, stream, req.FlowJobName)
	if err != nil {
		return nil, fmt.Errorf("failed to push to avro stage: %w", err)
	}

	bqClient := s.connector.client
	datasetID := s.connector.datasetID
	insertStmt := fmt.Sprintf("INSERT INTO `%s` SELECT * FROM `%s`;",
		rawTableName, stagingTable)

	activity.RecordHeartbeat(ctx,
		fmt.Sprintf("Flow job %s: performing insert and update transaction"+
			" for destination table %s and sync batch ID %d",
	
    
func GetAvroType(bqField *bigquery.FieldSchema) (interface{}, error) {
	avroNumericPrecision := int16(bqField.Precision)
	avroNumericScale := int16(bqField.Scale)
	if avroNumericPrecision > 38 || avroNumericPrecision <= 0 ||
		avroNumericScale > 38 || avroNumericScale < 0 {
		avroNumericPrecision = numeric.PeerDBNumericPrecision
		avroNumericScale = numeric.PeerDBNumericScale
	}

	considerRepeated := func(typ string, repeated bool) interface{} {
		if repeated {
			return qvalue.AvroSchemaArray{
				Type:  "array",
				Items: typ,
			}
		} else {
			return typ
		}
	}

	switch bqField.Type {
	case bigquery.StringFieldType, bigquery.GeographyFieldType, bigquery.JSONFieldType:
		return considerRepeated("string", bqField.Repeated), nil
	case bigquery.BytesFieldType:
		return "bytes", nil
	case bigquery.IntegerFieldType:
		return considerRepeated("long", bqField.Repeated), nil
	case bigquery.FloatFieldType:
		return considerRepeated("double", bqField.Repeated), nil
	case bigquery.BooleanFieldType:
		return considerRepeated("boolean", bqField.Repeated), nil
	case bigquery.TimestampFieldType:
		timestampSchema := qvalue.AvroSchemaField{
			Type:        "long",
			LogicalType: "timestamp-micros",
		}
		if bqField.Repeated {
			return qvalue.AvroSchemaComplexArray{
				Type:  "array",
				Items: timestampSchema,
			}, nil
		}
		return timestampSchema, nil
	case bigquery.DateFieldType:
		dateSchema := qvalue.AvroSchemaField{
			Type:        "int",
			LogicalType: "date",
		}
		if bqField.Repeated {
			return qvalue.AvroSchemaComplexArray{
				Type:  "array",
				Items: dateSchema,
			}, nil
		}
		return dateSchema, nil

	case bigquery.TimeFieldType:
		return qvalue.AvroSchemaField{
			Type:        "long",
			LogicalType: "time-micros",
		}, nil
	case bigquery.DateTimeFieldType:
		return qvalue.AvroSchemaRecord{
			Type: "record",
			Name: "datetime",
			Fields: []qvalue.AvroSchemaField{
				{
					Name:        "date",
					Type:        "int",
					LogicalType: "date",
				},
				{
					Name:        "time",
					Type:        "long",
					LogicalType: "time-micros",
				},
			},
		}, nil
	case bigquery.BigNumericFieldType:
		return qvalue.AvroSchemaNumeric{
			Type:        "bytes",
			LogicalType: "decimal",
			Precision:   avroNumericPrecision,
			Scale:       avroNumericScale,
		}, nil
	case bigquery.RecordFieldType:
		avroFields := []qvalue.AvroSchemaField{}
		for _, bqSubField := range bqField.Schema {
			avroType, err := GetAvroType(bqSubField)
			if err != nil {
				return nil, err
			}
			avroFields = append(avroFields, qvalue.AvroSchemaField{
				Name: bqSubField.Name,
				Type: avroType,
			})
		}
		return qvalue.AvroSchemaRecord{
			Type:   "record",
			Name:   bqField.Name,
			Fields: avroFields,
		}, nil

	default:
		return nil, fmt.Errorf("unsupported BigQuery field type: %s", bqField.Type)
	}
}