Allow null column type in column-lineage

CHANGELOG.md

@@ -1,9 +1,15 @@
 # Changelog
 
 ## [Unreleased](https://github.com/MarquezProject/marquez/compare/0.28.0...HEAD)
+
+### Added
+
 * Column-lineage endpoints supports point-in-time requests [`#2265`](https://github.com/MarquezProject/marquez/pull/2265) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)
     *Enable requesting `column-lineage` endpoint by a dataset version, job version or dataset field of a specific dataset version.*
 
+### Fixed
+
+* Allow null column type in column-lineage [`#2272`](https://github.com/MarquezProject/marquez/pull/2272) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)
 * Include error message for JSON processing exception [`#2271`](https://github.com/MarquezProject/marquez/pull/2271) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)  
     *In case of JSON processing exceptions Marquez API should return exception message to a client.*
 

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 datasetWithNullDataType = getDatasetB();
+    datasetWithNullDataType.getFacets().getSchema().getFields().get(0).setType(null);
+
+    UpdateLineageRow lineageRow = createLineage(openLineageDao, dataset_A, datasetWithNullDataType);
+    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();