From 8f9d5de587e78bcb9d35459e2326a1ed59182ea9 Mon Sep 17 00:00:00 2001
From: Fernando Velasquez <fevelasquez@google.com>
Date: Wed, 4 May 2022 19:55:21 -0400
Subject: [PATCH] Fixed type mappings when using the Storage Read API

---
 .../gcp/bigquery/sqlengine/BigQuerySparkDatasetProducer.java  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQuerySparkDatasetProducer.java b/src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQuerySparkDatasetProducer.java
index b4de6b60f3..4eae3fbd25 100644
--- a/src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQuerySparkDatasetProducer.java
+++ b/src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/BigQuerySparkDatasetProducer.java
@@ -123,13 +123,13 @@ private Dataset<Row> convertFieldTypes(Dataset<Row> ds) {
       }
 
       // Handle Int types
-      if (fieldSchema.getType() == Schema.Type.INT && fieldSchema.getLogicalType() == null) {
+      if (fieldSchema.getType() == Schema.Type.INT) {
         LOG.trace("Converting field {} to Integer", fieldName);
         ds = ds.withColumn(fieldName, ds.col(fieldName).cast(DataTypes.IntegerType));
       }
 
       // Handle float types
-      if (fieldSchema.getType() == Schema.Type.FLOAT && fieldSchema.getLogicalType() == null) {
+      if (fieldSchema.getType() == Schema.Type.FLOAT) {
         LOG.trace("Converting field {} to Float", fieldName);
         ds = ds.withColumn(fieldName, ds.col(fieldName).cast(DataTypes.FloatType));
       }