allow nullable column type in column-lineaeg

Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com>

Author: pawel-big-lebowski

api/src/main/java/marquez/db/mappers/ColumnLineageNodeDataMapper.java

@@ -39,7 +39,7 @@ public ColumnLineageNodeData map(ResultSet results, StatementContext ctx) throws
         stringOrThrow(results, Columns.DATASET_NAME),
         uuidOrThrow(results, Columns.DATASET_VERSION_UUID),
         stringOrThrow(results, Columns.FIELD_NAME),
-        stringOrThrow(results, Columns.TYPE),
+        stringOrNull(results, Columns.TYPE),
         stringOrNull(results, TRANSFORMATION_DESCRIPTION),
         stringOrNull(results, TRANSFORMATION_TYPE),
         toInputFields(results, "inputFields"));

api/src/main/java/marquez/db/models/ColumnLineageNodeData.java

@@ -19,7 +19,7 @@ public class ColumnLineageNodeData implements NodeData {
   @NonNull String dataset;
   @Nullable UUID datasetVersion;
   @NonNull String field;
-  @NonNull String fieldType;
+  @Nullable String fieldType;
   String transformationDescription;
   String transformationType;
   @NonNull List<InputFieldNodeData> inputFields;

api/src/test/java/marquez/db/ColumnLineageDaoTest.java

@@ -474,6 +474,15 @@ void testGetLineageWhenJobRunMultipleTimes() {
         .isEqualTo(lineageRow.getInputs().get().get(0).getDatasetVersionRow().getUuid());
   }
 
+  @Test
+  void testGetLineageWhenDataTypeIsEmpty() {
+    Dataset datasetBWithNullType = getDatasetB();
+    datasetBWithNullType.getFacets().getSchema().getFields().get(0).setType(null);
+
+    UpdateLineageRow lineageRow = createLineage(openLineageDao, dataset_A, datasetBWithNullType);
+    getColumnLineage(lineageRow, "col_c");
+  }
+
   private Set<ColumnLineageNodeData> getColumnLineage(UpdateLineageRow lineageRow, String field) {
     UpdateLineageRow.DatasetRecord datasetRecord = lineageRow.getOutputs().get().get(0);
     UUID field_UUID = fieldDao.findUuid(datasetRecord.getDatasetRow().getUuid(), field).get();