feat(bigquery): add support for AvroOptions in external data config (#4945)

Author: shollyman

Diff for: bigquery/external.go

@@ -143,6 +143,8 @@ func bqToExternalDataConfig(q *bq.ExternalDataConfiguration) (*ExternalDataConfi
 		e.DecimalTargetTypes = append(e.DecimalTargetTypes, DecimalTargetType(v))
 	}
 	switch {
+	case q.AvroOptions != nil:
+		e.Options = bqToAvroOptions(q.AvroOptions)
 	case q.CsvOptions != nil:
 		e.Options = bqToCSVOptions(q.CsvOptions)
 	case q.GoogleSheetsOptions != nil:
@@ -165,6 +167,29 @@ type ExternalDataConfigOptions interface {
 	populateExternalDataConfig(*bq.ExternalDataConfiguration)
 }
 
+// AvroOptions are additional options for Avro external data data sources.
+type AvroOptions struct {
+	// UseAvroLogicalTypes indicates whether to interpret logical types as the
+	// corresponding BigQuery data type (for example, TIMESTAMP), instead of using
+	// the raw type (for example, INTEGER).
+	UseAvroLogicalTypes bool
+}
+
+func (o *AvroOptions) populateExternalDataConfig(c *bq.ExternalDataConfiguration) {
+	c.AvroOptions = &bq.AvroOptions{
+		UseAvroLogicalTypes: o.UseAvroLogicalTypes,
+	}
+}
+
+func bqToAvroOptions(q *bq.AvroOptions) *AvroOptions {
+	if q == nil {
+		return nil
+	}
+	return &AvroOptions{
+		UseAvroLogicalTypes: q.UseAvroLogicalTypes,
+	}
+}
+
 // CSVOptions are additional options for CSV external data sources.
 type CSVOptions struct {
 	// AllowJaggedRows causes missing trailing optional columns to be tolerated

Diff for: bigquery/external_test.go

@@ -91,6 +91,12 @@ func TestExternalDataConfig(t *testing.T) {
 			SourceFormat:       Parquet,
 			DecimalTargetTypes: []DecimalTargetType{BigNumericTargetType, NumericTargetType, StringTargetType},
 		},
+		{
+			SourceFormat: Avro,
+			Options: &AvroOptions{
+				UseAvroLogicalTypes: true,
+			},
+		},
 	} {
 		q := want.toBQ()
 		got, err := bqToExternalDataConfig(&q)

Diff for: bigquery/file.go

@@ -77,6 +77,9 @@ type FileConfig struct {
 
 	// Additional options for Parquet files.
 	ParquetOptions *ParquetOptions
+
+	// Additional options for Avro files.
+	AvroOptions *AvroOptions
 }
 
 func (fc *FileConfig) populateLoadConfig(conf *bq.JobConfigurationLoad) {
@@ -98,6 +101,9 @@ func (fc *FileConfig) populateLoadConfig(conf *bq.JobConfigurationLoad) {
 			EnableListInference: fc.ParquetOptions.EnableListInference,
 		}
 	}
+	if fc.AvroOptions != nil {
+		conf.UseAvroLogicalTypes = fc.AvroOptions.UseAvroLogicalTypes
+	}
 	conf.Quote = fc.quote()
 }
 
@@ -131,6 +137,11 @@ func (fc *FileConfig) populateExternalDataConfig(conf *bq.ExternalDataConfigurat
 	if format == CSV {
 		fc.CSVOptions.populateExternalDataConfig(conf)
 	}
+	if fc.AvroOptions != nil {
+		conf.AvroOptions = &bq.AvroOptions{
+			UseAvroLogicalTypes: fc.AvroOptions.UseAvroLogicalTypes,
+		}
+	}
 	if fc.ParquetOptions != nil {
 		conf.ParquetOptions = &bq.ParquetOptions{
 			EnumAsString:        fc.ParquetOptions.EnumAsString,

Diff for: bigquery/file_test.go

@@ -96,6 +96,19 @@ func TestFileConfigPopulateLoadConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			description: "avro",
+			fileConfig: &FileConfig{
+				SourceFormat: Avro,
+				AvroOptions: &AvroOptions{
+					UseAvroLogicalTypes: true,
+				},
+			},
+			want: &bq.JobConfigurationLoad{
+				SourceFormat:        "AVRO",
+				UseAvroLogicalTypes: true,
+			},
+		},
 	}
 	for _, tc := range testcases {
 		got := &bq.JobConfigurationLoad{}