Skip to content

fix: fix nullpointer on empty bq default columns #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lavkesh merged 2 commits into from
Jul 25, 2022
Merged

fix: fix nullpointer on empty bq default columns #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fb80b8b
fix: fix nullpointer on empty bq default columns
Jul 25, 2022
fec52c4
chore: remove unused tablePartitionKey
Jul 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 3 additions & 12 deletions src/main/java/io/odpf/depot/bigquery/handler/JsonErrorHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;
Expand All @@ -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()) {
Expand Down Expand Up @@ -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");
}
Expand Down
19 changes: 6 additions & 13 deletions src/main/java/io/odpf/depot/bigquery/json/BigqueryJsonUpdateListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}