Fix column lineage transformation details always appearing as null

api/src/main/java/marquez/db/OpenLineageDao.java

@@ -1049,10 +1049,23 @@ private List<ColumnLineageRow> upsertColumnLineage(
                 return Stream.empty();
               }
 
-              // get field uuids of input columns related to this run
-              List<Pair<UUID, UUID>> inputFields =
+              // Match each field associated with this run to the OpenLineage input field it
+              // corresponds to (if any), along with the transformation description/type that
+              // applies to that specific input -> output edge. Prefer the (non-deprecated)
+              // per-input-field InputField#transformations entry reported by the producer;
+              // fall back to the deprecated, whole-output-column
+              // transformationDescription/transformationType for producers that only report the
+              // old format. Without this fallback/lookup, transformation details reported via the
+              // modern `transformations` array were silently dropped and always appeared as null
+              // (see #3100).
+              //
+              // Input fields are then grouped by the transformation (description, type) that
+              // applies to each of them, so that each distinct transformation is persisted with
+              // its own value while still reusing the existing batch upsertColumnLineageRow(...)
+              // API (which applies a single description/type to every input field passed to it).
+              Map<Pair<String, String>, List<Pair<UUID, UUID>>> inputFieldsByTransformation =
                   runFields.stream()
-                      .filter(
+                      .flatMap(
                           fieldData ->
                               columnLineage.getInputFields().stream()
                                   .filter(
@@ -1061,33 +1074,89 @@ private List<ColumnLineageRow> upsertColumnLineage(
                                               && of.getName().equals(fieldData.getDatasetName())
                                               && of.getField().equals(fieldData.getField()))
                                   .findAny()
-                                  .isPresent())
-                      .map(
-                          fieldData ->
-                              Pair.of(
-                                  fieldData.getDatasetVersionUuid(),
-                                  fieldData.getDatasetFieldUuid()))
-                      .collect(Collectors.toList());
+                                  .map(matchedInputField -> Pair.of(fieldData, matchedInputField))
+                                  .stream())
+                      .collect(
+                          Collectors.groupingBy(
+                              fieldDataAndInputField ->
+                                  transformationOf(
+                                      fieldDataAndInputField.getRight(), columnLineage),
+                              Collectors.mapping(
+                                  fieldDataAndInputField ->
+                                      Pair.of(
+                                          fieldDataAndInputField.getLeft().getDatasetVersionUuid(),
+                                          fieldDataAndInputField.getLeft().getDatasetFieldUuid()),
+                                  Collectors.toList())));
 
               log.debug(
-                  "Adding column lineage on output field '{}' for dataset version '{}' with input fields: {}",
+                  "Adding column lineage on output field '{}' for dataset version '{}' with input fields by transformation: {}",
                   outputField.get().getName(),
                   datasetVersionRow.getUuid(),
-                  inputFields);
-              return daos
-                  .getColumnLineageDao()
-                  .upsertColumnLineageRow(
-                      datasetVersionRow.getUuid(),
-                      outputField.get().getUuid(),
-                      inputFields,
-                      columnLineage.getTransformationDescription(),
-                      columnLineage.getTransformationType(),
-                      now)
-                  .stream();
+                  inputFieldsByTransformation);
+
+              // NOTE: when this output field has 2+ distinct transformation groups (its input
+              // fields resolve to more than one distinct (description, type) pair via
+              // transformationOf(...) above), upsertColumnLineageRow(...) below is invoked once
+              // per group. That method's implementation always returns ALL column_lineage rows
+              // currently persisted for this (outputDatasetVersionUuid, outputDatasetFieldUuid)
+              // pair - not just the rows written by the current call - via
+              // ColumnLineageDao#findColumnLineageByDatasetVersionColumnAndOutputDatasetField.
+              // So across multiple groups, the same underlying row can appear more than once in
+              // the aggregate List<ColumnLineageRow> this method returns for a given output
+              // field.
+              //
+              // As of this writing this has no production impact: the only consumer of this
+              // return value, DatasetRecord#getColumnLineageRows(), is not read anywhere in
+              // production code today (only in tests, and only the single-transformation-group
+              // case is currently exercised). If a real consumer is added later, either dedupe
+              // the combined list here or change upsertColumnLineageRow(...)/ColumnLineageDao to
+              // return only the rows written by that specific call.
+              return inputFieldsByTransformation.entrySet().stream()
+                  .flatMap(
+                      entry ->
+                          daos.getColumnLineageDao()
+                              .upsertColumnLineageRow(
+                                  datasetVersionRow.getUuid(),
+                                  outputField.get().getUuid(),
+                                  entry.getValue(),
+                                  entry.getKey().getLeft(),
+                                  entry.getKey().getRight(),
+                                  now)
+                              .stream());
             })
         .collect(Collectors.toList());
   }
 
+  /**
+   * Resolves the (transformationDescription, transformationType) pair that applies to a single
+   * OpenLineage column-lineage input field. Prefers the first entry of the (non-deprecated) {@link
+   * LineageEvent.ColumnLineageInputField#getTransformations()} list reported for that specific
+   * input field; falls back to the deprecated, whole-output-column {@code
+   * transformationDescription}/{@code transformationType} on {@link
+   * LineageEvent.ColumnLineageOutputColumn} for producers that only report the old format.
+   *
+   * <p>If a producer reports more than one transformation for a single input field, only the
+   * first is used, matching the granularity at which Marquez persists a transformation
+   * description/type (one value per input/output field edge).
+   */
+  static Pair<String, String> transformationOf(
+      LineageEvent.ColumnLineageInputField inputField,
+      LineageEvent.ColumnLineageOutputColumn outputColumn) {
+    Optional<LineageEvent.ColumnLineageInputField.Transformation> transformation =
+        Optional.ofNullable(inputField.getTransformations()).stream()
+            .flatMap(List::stream)
+            .findFirst();
+    String transformationDescription =
+        transformation
+            .map(LineageEvent.ColumnLineageInputField.Transformation::getDescription)
+            .orElseGet(outputColumn::getTransformationDescription);
+    String transformationType =
+        transformation
+            .map(LineageEvent.ColumnLineageInputField.Transformation::getType)
+            .orElseGet(outputColumn::getTransformationType);
+    return Pair.of(transformationDescription, transformationType);
+  }
+
   default String formatDatasetName(String name) {
     return name;
   }

api/src/main/java/marquez/service/models/LineageEvent.java

@@ -630,8 +630,6 @@ public static class ColumnLineageOutputColumn extends BaseJsonModel {
     private String transformationType;
   }
 
-  @Builder
-  @AllArgsConstructor
   @NoArgsConstructor
   @Setter
   @Getter
@@ -642,6 +640,44 @@ public static class ColumnLineageInputField extends BaseJsonModel {
     @NotNull private String namespace;
     @NotNull private String name;
     @NotNull private String field;
+
+    /**
+     * The transformations applied to this specific input field in order to compute the output
+     * field it is associated with. This is the current (non-deprecated) way for a producer to
+     * report per-input-field transformation details, as opposed to the deprecated {@link
+     * ColumnLineageOutputColumn#getTransformationDescription()} / {@link
+     * ColumnLineageOutputColumn#getTransformationType()}, which apply (at most) a single,
+     * whole-column value shared by every input field. See
+     * https://github.com/OpenLineage/OpenLineage/blob/main/spec/facets/ColumnLineageDatasetFacet.json
+     */
+    private List<Transformation> transformations;
+
+    public ColumnLineageInputField(String namespace, String name, String field) {
+      this(namespace, name, field, null);
+    }
+
+    @Builder
+    public ColumnLineageInputField(
+        String namespace, String name, String field, List<Transformation> transformations) {
+      this.namespace = namespace;
+      this.name = name;
+      this.field = field;
+      this.transformations = transformations;
+    }
+
+    @Builder
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Setter
+    @Getter
+    @Valid
+    @ToString
+    public static class Transformation {
+      private String type;
+      private String subtype;
+      private String description;
+      private Boolean masking;
+    }
   }
 
   @Builder

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

@@ -0,0 +1,153 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.db;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import marquez.service.models.LineageEvent;
+import marquez.service.models.LineageEvent.ColumnLineageInputField;
+import marquez.service.models.LineageEvent.ColumnLineageInputField.Transformation;
+import marquez.service.models.LineageEvent.ColumnLineageOutputColumn;
+import org.apache.commons.lang3.tuple.Pair;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Regression tests for https://github.com/MarquezProject/marquez/issues/3100: transformation
+ * details reported via the (non-deprecated) {@code InputField.transformations[]} array of the
+ * OpenLineage {@code ColumnLineageDatasetFacet} were always dropped, so {@code
+ * transformation_description}/{@code transformation_type} always ended up null in Marquez,
+ * regardless of what the producer actually reported.
+ *
+ * <p>These tests exercise {@link OpenLineageDao#transformationOf(ColumnLineageInputField,
+ * ColumnLineageOutputColumn)} directly - the pure function responsible for resolving which
+ * transformation applies to a given input field - so they can run without a database.
+ */
+@org.junit.jupiter.api.Tag("UnitTests")
+class OpenLineageDaoColumnLineageTransformationTest {
+
+  @Test
+  void prefersPerInputFieldTransformationOverDeprecatedOutputColumnFields() {
+    // Reproduces the exact scenario from the issue: the output column itself does not set the
+    // deprecated transformationDescription/transformationType, but each input field reports its
+    // own transformations[] entry.
+    ColumnLineageInputField inputField =
+        ColumnLineageInputField.builder()
+            .namespace("bookstore2")
+            .name("customers")
+            .field("customer_email")
+            .transformations(
+                Collections.singletonList(
+                    Transformation.builder()
+                        .type("DIRECT")
+                        .subtype("TRANSFORMATION")
+                        .description("concat(customers.customer_name, ' - ', customers.customer_email)")
+                        .build()))
+            .build();
+    ColumnLineageOutputColumn outputColumn =
+        ColumnLineageOutputColumn.builder().inputFields(List.of(inputField)).build();
+
+    Pair<String, String> transformation = OpenLineageDao.transformationOf(inputField, outputColumn);
+
+    assertThat(transformation.getLeft())
+        .isEqualTo("concat(customers.customer_name, ' - ', customers.customer_email)");
+    assertThat(transformation.getRight()).isEqualTo("DIRECT");
+  }
+
+  @Test
+  void distinctInputFieldsOnTheSameOutputColumnCanHaveDifferentTransformations() {
+    // customer_full in the issue is fed by two input fields, each with a different underlying
+    // source field but (in the issue's example) the same description; verify the resolution is
+    // genuinely per-input-field rather than picking a single value for the whole output column.
+    ColumnLineageInputField emailField =
+        ColumnLineageInputField.builder()
+            .namespace("bookstore2")
+            .name("customers")
+            .field("customer_email")
+            .transformations(
+                Collections.singletonList(
+                    Transformation.builder().type("DIRECT").description("descriptionA").build()))
+            .build();
+    ColumnLineageInputField nameField =
+        ColumnLineageInputField.builder()
+            .namespace("bookstore2")
+            .name("customers")
+            .field("customer_name")
+            .transformations(
+                Collections.singletonList(
+                    Transformation.builder().type("INDIRECT").description("descriptionB").build()))
+            .build();
+    ColumnLineageOutputColumn outputColumn =
+        ColumnLineageOutputColumn.builder()
+            .inputFields(Arrays.asList(emailField, nameField))
+            .build();
+
+    Pair<String, String> emailTransformation =
+        OpenLineageDao.transformationOf(emailField, outputColumn);
+    Pair<String, String> nameTransformation =
+        OpenLineageDao.transformationOf(nameField, outputColumn);
+
+    assertThat(emailTransformation).isEqualTo(Pair.of("descriptionA", "DIRECT"));
+    assertThat(nameTransformation).isEqualTo(Pair.of("descriptionB", "INDIRECT"));
+  }
+
+  @Test
+  void fallsBackToDeprecatedOutputColumnFieldsWhenInputFieldHasNoTransformations() {
+    // Producers using the old (deprecated) format never set inputField.transformations at all;
+    // this must keep working exactly as before.
+    ColumnLineageInputField inputField =
+        ColumnLineageInputField.builder()
+            .namespace("ns")
+            .name("upstream")
+            .field("col")
+            .build(); // no transformations set
+    ColumnLineageOutputColumn outputColumn =
+        ColumnLineageOutputColumn.builder()
+            .inputFields(List.of(inputField))
+            .transformationDescription("legacy description")
+            .transformationType("IDENTITY")
+            .build();
+
+    Pair<String, String> transformation = OpenLineageDao.transformationOf(inputField, outputColumn);
+
+    assertThat(transformation).isEqualTo(Pair.of("legacy description", "IDENTITY"));
+  }
+
+  @Test
+  void returnsNullPairWhenNeitherFormatIsReported() {
+    ColumnLineageInputField inputField =
+        ColumnLineageInputField.builder().namespace("ns").name("upstream").field("col").build();
+    ColumnLineageOutputColumn outputColumn =
+        ColumnLineageOutputColumn.builder().inputFields(List.of(inputField)).build();
+
+    Pair<String, String> transformation = OpenLineageDao.transformationOf(inputField, outputColumn);
+
+    assertThat(transformation.getLeft()).isNull();
+    assertThat(transformation.getRight()).isNull();
+  }
+
+  @Test
+  void usesOnlyFirstTransformationWhenMultipleAreReportedForTheSameInputField() {
+    ColumnLineageInputField inputField =
+        ColumnLineageInputField.builder()
+            .namespace("ns")
+            .name("upstream")
+            .field("col")
+            .transformations(
+                Arrays.asList(
+                    Transformation.builder().type("DIRECT").description("first").build(),
+                    Transformation.builder().type("INDIRECT").description("second").build()))
+            .build();
+    ColumnLineageOutputColumn outputColumn =
+        ColumnLineageOutputColumn.builder().inputFields(List.of(inputField)).build();
+
+    Pair<String, String> transformation = OpenLineageDao.transformationOf(inputField, outputColumn);
+
+    assertThat(transformation).isEqualTo(Pair.of("first", "DIRECT"));
+  }
+}