fix: fix nullpointer on empty bq default columns (#22)

* fix: fix nullpointer on empty bq default columns

* chore: remove unused tablePartitionKey

Co-authored-by: jesry.pandawa <jesry.pandawa@go-jek.com>

Author: jesrypandawa

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;
 
@@ -28,8 +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 Optional<LegacySQLTypeName> partitionKeyDataType;
     private final boolean castAllColumnsToStringDataType;
     private final Map<String, String> metadataColumnsTypesMap;
     private final String bqMetadataNamespace;
@@ -40,15 +37,9 @@ 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));
-        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()) {
@@ -103,11 +94,11 @@ private List<BigQueryError> getBqErrorsWithNoSuchFields(List<BigQueryError> valu
                 ).collect(Collectors.toList());
     }
 
+    /**
+     * This method only used for unknown fields.
+     */
 
     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");
         }

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);
     }
 }