From fb80b8ba27e7b1237a9c5115c8921f9b88a66754 Mon Sep 17 00:00:00 2001
From: "jesry.pandawa" <jesry.pandawa@go-jek.com>
Date: Mon, 25 Jul 2022 16:58:09 +0700
Subject: [PATCH 1/2] fix: fix nullpointer on empty bq default columns

---
 build.gradle                                           |  2 +-
 .../odpf/depot/bigquery/handler/JsonErrorHandler.java  | 10 ----------
 2 files changed, 1 insertion(+), 11 deletions(-)

diff --git a/build.gradle b/build.gradle
index ecc56aaf..021dcb72 100644
--- a/build.gradle
+++ b/build.gradle
@@ -22,7 +22,7 @@ plugins {
 }
 
 group 'io.odpf'
-version '0.1.6'
+version '0.1.7'
 
 repositories {
     mavenCentral()
diff --git a/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java b/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
index 6472aec0..466534bd 100644
--- a/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
+++ b/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
@@ -17,7 +17,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -29,7 +28,6 @@ public class JsonErrorHandler implements ErrorHandler {
 
     private final BigQueryClient bigQueryClient;
     private final String tablePartitionKey;
-    private final Optional<LegacySQLTypeName> partitionKeyDataType;
     private final boolean castAllColumnsToStringDataType;
     private final Map<String, String> metadataColumnsTypesMap;
     private final String bqMetadataNamespace;
@@ -44,11 +42,6 @@ public JsonErrorHandler(BigQueryClient bigQueryClient, BigQuerySinkConfig bigQue
         defaultColumnsMap = bigQuerySinkConfig.getSinkBigqueryDefaultColumns()
                 .stream()
                 .collect(Collectors.toMap(TupleString::getFirst, TupleString::getSecond));
-        if (bigQuerySinkConfig.isTablePartitioningEnabled()) {
-            partitionKeyDataType = Optional.of(LegacySQLTypeName.valueOfStrict(defaultColumnsMap.get(tablePartitionKey).toUpperCase()));
-        } else {
-            partitionKeyDataType = Optional.empty();
-        }
         castAllColumnsToStringDataType = bigQuerySinkConfig.getSinkBigqueryDefaultDatatypeStringEnable();
         bqMetadataNamespace = bigQuerySinkConfig.getBqMetadataNamespace();
         if (!bigQuerySinkConfig.shouldAddMetadata()) {
@@ -105,9 +98,6 @@ private List<BigQueryError> getBqErrorsWithNoSuchFields(List<BigQueryError> valu
 
 
     private Field getField(String key) {
-        if (!tablePartitionKey.isEmpty() && tablePartitionKey.equals(key) && partitionKeyDataType.isPresent()) {
-            return Field.of(key, partitionKeyDataType.get());
-        }
         if (!bqMetadataNamespace.isEmpty()) {
             throw new UnsupportedOperationException("metadata namespace is not supported, because nested json structure is not supported");
         }

From fec52c42349d7cee8a2eed9216cb55c65bd63f6c Mon Sep 17 00:00:00 2001
From: "jesry.pandawa" <jesry.pandawa@go-jek.com>
Date: Mon, 25 Jul 2022 18:57:58 +0700
Subject: [PATCH 2/2] chore: remove unused tablePartitionKey

---
 build.gradle                                  |  2 +-
 .../bigquery/handler/JsonErrorHandler.java    |  5 +++--
 .../json/BigqueryJsonUpdateListener.java      | 19 ++++++-------------
 3 files changed, 10 insertions(+), 16 deletions(-)

diff --git a/build.gradle b/build.gradle
index 021dcb72..ecc56aaf 100644
--- a/build.gradle
+++ b/build.gradle
@@ -22,7 +22,7 @@ plugins {
 }
 
 group 'io.odpf'
-version '0.1.7'
+version '0.1.6'
 
 repositories {
     mavenCentral()
diff --git a/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java b/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
index 466534bd..b1dd4329 100644
--- a/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
+++ b/src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
@@ -27,7 +27,6 @@ the job of the class is to handle unknown field errors and then update the bq ta
 public class JsonErrorHandler implements ErrorHandler {
 
     private final BigQueryClient bigQueryClient;
-    private final String tablePartitionKey;
     private final boolean castAllColumnsToStringDataType;
     private final Map<String, String> metadataColumnsTypesMap;
     private final String bqMetadataNamespace;
@@ -38,7 +37,6 @@ public JsonErrorHandler(BigQueryClient bigQueryClient, BigQuerySinkConfig bigQue
 
         this.instrumentation = instrumentation;
         this.bigQueryClient = bigQueryClient;
-        tablePartitionKey = bigQuerySinkConfig.isTablePartitioningEnabled() ? bigQuerySinkConfig.getTablePartitionKey() : "";
         defaultColumnsMap = bigQuerySinkConfig.getSinkBigqueryDefaultColumns()
                 .stream()
                 .collect(Collectors.toMap(TupleString::getFirst, TupleString::getSecond));
@@ -96,6 +94,9 @@ private List<BigQueryError> getBqErrorsWithNoSuchFields(List<BigQueryError> valu
                 ).collect(Collectors.toList());
     }
 
+    /**
+     * This method only used for unknown fields.
+     */
 
     private Field getField(String key) {
         if (!bqMetadataNamespace.isEmpty()) {
diff --git a/src/main/java/io/odpf/depot/bigquery/json/BigqueryJsonUpdateListener.java b/src/main/java/io/odpf/depot/bigquery/json/BigqueryJsonUpdateListener.java
index 10ddf735..d3c00f48 100644
--- a/src/main/java/io/odpf/depot/bigquery/json/BigqueryJsonUpdateListener.java
+++ b/src/main/java/io/odpf/depot/bigquery/json/BigqueryJsonUpdateListener.java
@@ -93,29 +93,22 @@ private void addMetadataFields(HashSet<Field> fieldsToBeUpdated, List<TupleStrin
     private Field getField(TupleString tupleString) {
         String fieldName = tupleString.getFirst();
         LegacySQLTypeName fieldDataType = LegacySQLTypeName.valueOfStrict(tupleString.getSecond().toUpperCase());
-
-        if (isValidPartitionField(fieldName, fieldDataType)) {
-            return Field.of(fieldName, fieldDataType);
-        }
-
-        return Field.of(fieldName, fieldDataType);
+        return checkAndCreateField(fieldName, fieldDataType);
     }
 
     /**
-     * Range Bigquery partitioning is not supported, supported paritition fields have to be of DATE or TIMESTAMP type..
+     * Range BigQuery partitioning is not supported, supported partition fields have to be of DATE or TIMESTAMP type..
      */
-    private boolean isValidPartitionField(String fieldName, LegacySQLTypeName fieldDataType) {
+    private Field checkAndCreateField(String fieldName, LegacySQLTypeName fieldDataType) {
         Boolean isPartitioningEnabled = config.isTablePartitioningEnabled();
         if (!isPartitioningEnabled) {
-            return false;
+            return Field.of(fieldName, fieldDataType);
         }
         String partitionKey = config.getTablePartitionKey();
-
         boolean isValidPartitionDataType = (fieldDataType == LegacySQLTypeName.TIMESTAMP || fieldDataType == LegacySQLTypeName.DATE);
         if (partitionKey.equals(fieldName) && !isValidPartitionDataType) {
-            throw new UnsupportedOperationException(" supported paritition fields have to be of DATE or TIMESTAMP type..");
+            throw new UnsupportedOperationException("supported partition fields have to be of DATE or TIMESTAMP type..");
         }
-
-        return true;
+        return Field.of(fieldName, fieldDataType);
     }
 }