Merge branch 'datahub-project:master' into master

Author: anshbansal

.github/workflows/airflow-plugin.yml

@@ -34,11 +34,8 @@ jobs:
         include:
           # Note: this should be kept in sync with tox.ini.
           - python-version: "3.8"
-            extra_pip_requirements: "apache-airflow~=2.3.4"
-            extra_pip_extras: test-airflow23
-          - python-version: "3.10"
-            extra_pip_requirements: "apache-airflow~=2.4.3"
-            extra_pip_extras: test-airflow24
+            extra_pip_extras: test-airflow25
+            # extra_pip_requirements: "apache-airflow~=2.5.3 -c https://raw.githubusercontent.com/apache/airflow/constraints-2.5.3/constraints-3.8.txt"
           - python-version: "3.10"
             extra_pip_requirements: "apache-airflow~=2.6.3 -c https://raw.githubusercontent.com/apache/airflow/constraints-2.6.3/constraints-3.10.txt"
           - python-version: "3.10"

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/BuildIndicesConfig.java

@@ -12,10 +12,12 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
 
 @Configuration
 @Conditional(SystemUpdateCondition.BlockingSystemUpdateCondition.class)
 public class BuildIndicesConfig {
+  @Order(1)
   @Bean(name = "buildIndices")
   public BlockingSystemUpgrade buildIndices(
       final SystemMetadataService systemMetadataService,

datahub-upgrade/src/test/java/com/linkedin/datahub/upgrade/DatahubUpgradeBlockingTest.java

@@ -4,8 +4,10 @@
 import static org.testng.Assert.assertTrue;
 import static org.testng.AssertJUnit.assertNotNull;
 
+import com.linkedin.datahub.upgrade.system.BlockingSystemUpgrade;
 import com.linkedin.datahub.upgrade.system.SystemUpdateBlocking;
 import com.linkedin.datahub.upgrade.system.bootstrapmcps.BootstrapMCPStep;
+import com.linkedin.datahub.upgrade.system.elasticsearch.BuildIndices;
 import java.util.List;
 import java.util.stream.Collectors;
 import javax.inject.Named;
@@ -25,6 +27,14 @@ public class DatahubUpgradeBlockingTest extends AbstractTestNGSpringContextTests
   @Named("systemUpdateBlocking")
   private SystemUpdateBlocking systemUpdateBlocking;
 
+  @Autowired private List<BlockingSystemUpgrade> blockingSystemUpgrades;
+
+  @Test
+  public void testBuildIndicesOrder() {
+    assertNotNull(blockingSystemUpgrades);
+    assertTrue(blockingSystemUpgrades.get(0) instanceof BuildIndices);
+  }
+
   @Test
   public void testNBlockingBootstrapMCP() {
     assertNotNull(systemUpdateBlocking);

datahub-web-react/src/app/entityV2/shared/tabs/Incident/__tests__/utils.test.tsx

@@ -1,5 +1,6 @@
 import { format } from 'date-fns';
 import { SortingState } from '@src/alchemy-components/components/Table/types';
+import { EntityType } from '@src/types.generated';
 import {
     getFilteredTransformedIncidentData,
     getLinkedAssetsCount,
@@ -10,6 +11,7 @@ import {
     validateForm,
     getSortedIncidents,
     getExistingIncidents,
+    useSiblingOptionsForIncidentBuilder,
 } from '../utils';
 import { IncidentListFilter } from '../types';
 
@@ -119,4 +121,122 @@ describe('Utility Functions', () => {
         };
         expect(getExistingIncidents(currData)).toEqual([{ id: 1 }, { id: 2 }]);
     });
+
+    test('should return main entity data in options', () => {
+        const mockEntityData = {
+            urn: 'urn:li:dataset:(urn:li:dataPlatform:bigquery,my_table,PROD)',
+            platform: {
+                properties: {
+                    displayName: 'BigQuery',
+                },
+                name: 'bigquery',
+                urn: 'urn:li:dataPlatform:bigquery',
+            },
+            dataPlatformInstance: {
+                platform: {
+                    name: 'BigQueryInstance',
+                },
+            },
+            siblingsSearch: {
+                searchResults: [
+                    {
+                        entity: {
+                            urn: 'urn:li:dataset:(urn:li:dataPlatform:snowflake,my_table,PROD)',
+                            platform: {
+                                properties: {
+                                    displayName: 'Snowflake',
+                                },
+                                name: 'snowflake',
+                                urn: 'urn:li:dataPlatform:snowflake',
+                            },
+                            dataPlatformInstance: {
+                                platform: {
+                                    name: 'SnowflakeInstance',
+                                },
+                            },
+                        },
+                    },
+                ],
+            },
+        } as any;
+        const result = useSiblingOptionsForIncidentBuilder(
+            mockEntityData,
+            'urn:li:dataset:main',
+            'DATASET' as EntityType,
+        );
+        expect(result[0]).toEqual({
+            title: 'BigQuery',
+            urn: 'urn:li:dataset:main',
+            platform: mockEntityData.platform,
+            entityType: 'DATASET',
+        });
+    });
+
+    test('should include siblings data in options', () => {
+        const mockEntityData = {
+            urn: 'urn:li:dataset:(urn:li:dataPlatform:bigquery,my_table,PROD)',
+            platform: {
+                properties: { displayName: 'BigQuery' },
+                name: 'bigquery',
+                urn: 'urn:li:dataPlatform:bigquery',
+            },
+            dataPlatformInstance: {
+                platform: { name: 'BigQueryInstance' },
+            },
+            siblingsSearch: {
+                searchResults: [
+                    {
+                        entity: {
+                            urn: 'urn:li:dataset:(urn:li:dataPlatform:snowflake,my_table1,PROD)',
+                            platform: {
+                                properties: { displayName: 'Snowflake' },
+                                name: 'snowflake',
+                                urn: 'urn:li:dataPlatform:snowflake',
+                            },
+                            dataPlatformInstance: { platform: { name: 'SnowflakeInstance' } },
+                            type: 'DATASET',
+                        },
+                    },
+                    {
+                        entity: {
+                            urn: 'urn:li:dataset:(urn:li:dataPlatform:redshift,my_table2,PROD)',
+                            platform: {
+                                properties: { displayName: 'Redshift' },
+                                name: 'redshift',
+                                urn: 'urn:li:dataPlatform:redshift',
+                            },
+                            dataPlatformInstance: { platform: { name: 'RedshiftInstance' } },
+                            type: 'DATASET',
+                        },
+                    },
+                ],
+            },
+        } as any;
+
+        const result = useSiblingOptionsForIncidentBuilder(mockEntityData, mockEntityData.urn, 'DATASET' as EntityType);
+
+        // Expect 1 main entity + 2 siblings
+        expect(result.length).toBe(3);
+
+        expect(result[0]).toEqual({
+            urn: mockEntityData.urn,
+            title: 'BigQuery',
+            platform: mockEntityData.platform,
+            entityType: 'DATASET',
+        });
+
+        expect(result[1]).toEqual({
+            urn: 'urn:li:dataset:(urn:li:dataPlatform:snowflake,my_table1,PROD)',
+            title: 'Snowflake',
+            platform: mockEntityData.siblingsSearch.searchResults[0].entity.platform,
+            entityType: 'DATASET',
+        });
+
+        expect(result[2]).toEqual({
+            urn: 'urn:li:dataset:(urn:li:dataPlatform:redshift,my_table2,PROD)',
+            title: 'Redshift',
+            platform: mockEntityData.siblingsSearch.searchResults[1].entity.platform,
+            entityType: 'DATASET',
+        });
+    });
 });

datahub-web-react/src/app/lineageV2/useSearchAcrossLineage.ts

@@ -85,7 +85,7 @@ export default function useSearchAcrossLineage(
         lineageFlags: {
             startTimeMillis,
             endTimeMillis,
-            entitiesExploredPerHopLimit: PER_HOP_LIMIT,
+            entitiesExploredPerHopLimit: maxDepth ? PER_HOP_LIMIT : undefined,
             ignoreAsHops: DEFAULT_IGNORE_AS_HOPS,
         },
         searchFlags: {

docs/how/updating-datahub.md

@@ -20,6 +20,8 @@ This file documents any backwards-incompatible changes in DataHub and assists pe
 
 ### Breaking Changes
 
+- #13004: The `acryl-datahub-airflow-plugin` dropped support for Airflow 2.3 and 2.4.
+
 ### Potential Downtime
 
 ### Deprecations
@@ -42,7 +44,6 @@ This file documents any backwards-incompatible changes in DataHub and assists pe
 
 - #12797: Previously endpoints when used in ASYNC mode would not validate URNs, entity & aspect names immediately. Starting with this release, even in ASYNC mode, these requests will be returned with http code 400.
 
-
 ### Known Issues
 
 - #12601: Jetty 12 introduces a stricter handling of url encoding. We are currently applying a workaround to prevent a regression, while technically breaking the official specifications.
@@ -84,7 +85,7 @@ This file documents any backwards-incompatible changes in DataHub and assists pe
 - #12020 - Removed `sql_parser` configuration from the Redash source, as Redash now exclusively uses the sqlglot-based parser for lineage extraction.
 - #12020 - Removed `datahub.utilities.sql_parser`, `datahub.utilities.sql_parser_base` and `datahub.utilities.sql_lineage_parser_impl` module along with `SqlLineageSQLParser` and `DefaultSQLParser`. Use `create_lineage_sql_parsed_result` from `datahub.sql_parsing.sqlglot_lineage` module instead.
 - #11518 - DataHub Garbage Collection: Various entities that are soft-deleted
-  (after 10d) or are timeseries *entities* (dataprocess, execution requests)
+  (after 10d) or are timeseries _entities_ (dataprocess, execution requests)
   will be removed automatically using logic in the `datahub-gc` ingestion
   source.
 - #12067 - Default behavior of DataJobPatchBuilder in Python sdk has been

docs/lineage/airflow.md

@@ -17,10 +17,10 @@ There's two implementations of the plugin, with different Airflow version suppor
 
 | Approach  | Airflow Versions | Notes                                                                                   |
 | --------- | ---------------- | --------------------------------------------------------------------------------------- |
-| Plugin v2 | 2.3.4+           | Recommended. Requires Python 3.8+                                                       |
-| Plugin v1 | 2.3 - 2.8        | Deprecated. No automatic lineage extraction; may not extract lineage if the task fails. |
+| Plugin v2 | 2.5+             | Recommended. Requires Python 3.8+                                                       |
+| Plugin v1 | 2.5 - 2.8        | Deprecated. No automatic lineage extraction; may not extract lineage if the task fails. |
 
-If you're using Airflow older than 2.3, it's possible to use the v1 plugin with older versions of `acryl-datahub-airflow-plugin`. See the [compatibility section](#compatibility) for more details.
+If you're using Airflow older than 2.5, it's possible to use the plugin with older versions of `acryl-datahub-airflow-plugin`. See the [compatibility section](#compatibility) for more details.
 
 <!-- TODO: Update the local Airflow guide and link to it here. -->
 <!-- If you are looking to run Airflow and DataHub using docker locally, follow the guide [here](../../docker/airflow/local_airflow.md). -->
@@ -376,12 +376,15 @@ This will immediately disable the plugin without requiring a restart.
 
 ## Compatibility
 
-We no longer officially support Airflow <2.3. However, you can use older versions of `acryl-datahub-airflow-plugin` with older versions of Airflow.
-The first two options support Python 3.7+, and the last option supports Python 3.8+.
+We try to support Airflow releases for ~2 years after their release. This is a best-effort guarantee - it's not always possible due to dependency / security issues cropping up in older versions.
+
+We no longer officially support Airflow <2.5. However, you can use older versions of `acryl-datahub-airflow-plugin` with older versions of Airflow.
+The first two options support Python 3.7+, and the others require Python 3.8+.
 
 - Airflow 1.10.x, use DataHub plugin v1 with acryl-datahub-airflow-plugin <= 0.9.1.0.
 - Airflow 2.0.x, use DataHub plugin v1 with acryl-datahub-airflow-plugin <= 0.11.0.1.
 - Airflow 2.2.x, use DataHub plugin v2 with acryl-datahub-airflow-plugin <= 0.14.1.5.
+- Airflow 2.3 - 2.4.3, use DataHub plugin v2 with acryl-datahub-airflow-plugin <= 1.0.0.
 
 DataHub also previously supported an Airflow [lineage backend](https://airflow.apache.org/docs/apache-airflow/2.2.0/lineage.html#lineage-backend) implementation. While the implementation is still in our codebase, it is deprecated and will be removed in a future release.
 Note that the lineage backend did not support automatic lineage extraction, did not capture task failures, and did not work in AWS MWAA.

entity-registry/src/main/java/com/linkedin/metadata/aspect/hooks/OwnershipOwnerTypes.java

@@ -0,0 +1,102 @@
+package com.linkedin.metadata.aspect.hooks;
+
+import static com.linkedin.metadata.Constants.OWNERSHIP_ASPECT_NAME;
+
+import com.linkedin.common.*;
+import com.linkedin.common.urn.Urn;
+import com.linkedin.metadata.aspect.ReadItem;
+import com.linkedin.metadata.aspect.RetrieverContext;
+import com.linkedin.metadata.aspect.batch.ChangeMCP;
+import com.linkedin.metadata.aspect.plugins.config.AspectPluginConfig;
+import com.linkedin.metadata.aspect.plugins.hooks.MutationHook;
+import com.linkedin.util.Pair;
+import java.util.*;
+import java.util.stream.Stream;
+import javax.annotation.Nonnull;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Getter
+@Setter
+@Accessors(chain = true)
+public class OwnershipOwnerTypes extends MutationHook {
+  @Nonnull private AspectPluginConfig config;
+
+  protected Stream<Pair<ChangeMCP, Boolean>> writeMutation(
+      @Nonnull Collection<ChangeMCP> changeMCPS, @Nonnull RetrieverContext retrieverContext) {
+    return changeMCPS.stream()
+        .map(
+            item ->
+                aspectFilter(item)
+                    ? Pair.of(item, processOwnershipAspect(item))
+                    : Pair.of(item, false));
+  }
+
+  private static boolean aspectFilter(ReadItem item) {
+    return item.getAspectName().equals(OWNERSHIP_ASPECT_NAME);
+  }
+
+  public static Map<String, List<Urn>> toMap(UrnArrayMap map) {
+    Map<String, List<Urn>> result = new HashMap<>();
+    map.forEach(
+        ((s, urns) -> {
+          result.put(s, new ArrayList<>(urns));
+        }));
+    return result;
+  }
+
+  public static UrnArrayMap toUrnArrayMap(Map<String, List<Urn>> map) {
+    UrnArrayMap result = new UrnArrayMap();
+    map.forEach(
+        (key, urns) -> {
+          result.put(key, new UrnArray(urns));
+        });
+    return result;
+  }
+
+  public static boolean processOwnershipAspect(ChangeMCP item) {
+    boolean mutated = false;
+    Ownership ownership = item.getAspect(Ownership.class);
+    if (ownership == null) {
+      return false;
+    }
+    UrnArrayMap ownerTypes = ownership.getOwnerTypes();
+    Map<String, List<Urn>> ownerTypesMap;
+    if (ownerTypes == null) {
+      ownerTypesMap = new HashMap<>();
+      mutated = true;
+    } else {
+      ownerTypesMap = toMap(ownerTypes);
+    }
+    OwnerArray owners = ownership.getOwners();
+    for (Owner owner : owners) {
+      String typeKey =
+          Optional.ofNullable(owner.getTypeUrn())
+              .map(Urn::toString)
+              .orElseGet(
+                  () ->
+                      "urn:li:ownershipType:__system__" + owner.getType().toString().toLowerCase());
+
+      List<Urn> ownerOfType;
+      if (ownerTypesMap.containsKey(typeKey)) {
+        ownerOfType = ownerTypesMap.get(typeKey);
+      } else {
+        ownerOfType = new ArrayList<>();
+        ownerTypesMap.put(typeKey, ownerOfType);
+        mutated = true;
+      }
+      Urn ownerUrn = owner.getOwner();
+      if (!ownerOfType.contains(ownerUrn)) {
+        ownerOfType.add(ownerUrn);
+        mutated = true;
+      }
+    }
+    if (mutated) {
+      ownership.setOwnerTypes((toUrnArrayMap(ownerTypesMap)));
+    }
+    return mutated;
+  }
+}