-
Notifications
You must be signed in to change notification settings - Fork 2.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add-support-for-ingesting-schemas-from-schema-registry
- Loading branch information
1 parent
776e5c4
commit 1a57f3e
Showing
319 changed files
with
9,555 additions
and
1,112 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
...-core/src/main/java/com/linkedin/datahub/graphql/concurrency/GraphQLConcurrencyUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package com.linkedin.datahub.graphql.concurrency; | ||
|
|
||
| import com.codahale.metrics.MetricRegistry; | ||
| import com.linkedin.metadata.utils.metrics.MetricUtils; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public class GraphQLConcurrencyUtils { | ||
| private GraphQLConcurrencyUtils() {} | ||
|
|
||
| private static ExecutorService graphQLExecutorService = null; | ||
|
|
||
| public static ExecutorService getExecutorService() { | ||
| return GraphQLConcurrencyUtils.graphQLExecutorService; | ||
| } | ||
|
|
||
| public static void setExecutorService(ExecutorService executorService) { | ||
| GraphQLConcurrencyUtils.graphQLExecutorService = executorService; | ||
| } | ||
|
|
||
| public static <T> CompletableFuture<T> supplyAsync( | ||
| Supplier<T> supplier, String caller, String task) { | ||
| MetricUtils.counter( | ||
| MetricRegistry.name( | ||
| GraphQLConcurrencyUtils.class.getSimpleName(), "supplyAsync", caller, task)) | ||
| .inc(); | ||
| if (GraphQLConcurrencyUtils.graphQLExecutorService == null) { | ||
| return CompletableFuture.supplyAsync(supplier); | ||
| } else { | ||
| return CompletableFuture.supplyAsync( | ||
| supplier, GraphQLConcurrencyUtils.graphQLExecutorService); | ||
| } | ||
| } | ||
| } |
29 changes: 29 additions & 0 deletions
29
...rc/main/java/com/linkedin/datahub/graphql/concurrency/GraphQLWorkerPoolThreadFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package com.linkedin.datahub.graphql.concurrency; | ||
|
|
||
| import java.util.concurrent.ThreadFactory; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| public class GraphQLWorkerPoolThreadFactory implements ThreadFactory { | ||
|
|
||
| private static final AtomicLong THREAD_INIT_NUMBER = new AtomicLong(); | ||
| public static final String GRAPHQL_THREAD_POOL_GROUP_NAME = "graphQLThreadGroup"; | ||
| public static final ThreadGroup GRAPHQL_THREAD_POOL_GROUP = | ||
| new ThreadGroup(GRAPHQL_THREAD_POOL_GROUP_NAME); | ||
|
|
||
| private static long nextThreadNum() { | ||
| return THREAD_INIT_NUMBER.getAndIncrement(); | ||
| } | ||
|
|
||
| private long stackSize; | ||
|
|
||
| public GraphQLWorkerPoolThreadFactory(long stackSize) { | ||
| this.stackSize = stackSize; | ||
| } | ||
|
|
||
| @Override | ||
| public final Thread newThread(Runnable runnable) { | ||
|
|
||
| return new Thread( | ||
| GRAPHQL_THREAD_POOL_GROUP, runnable, "GraphQLWorkerThread-" + nextThreadNum(), stackSize); | ||
| } | ||
| } |
74 changes: 74 additions & 0 deletions
74
...n/java/com/linkedin/datahub/graphql/instrumentation/DataHubFieldComplexityCalculator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| package com.linkedin.datahub.graphql.instrumentation; | ||
|
|
||
| import graphql.analysis.FieldComplexityCalculator; | ||
| import graphql.analysis.FieldComplexityEnvironment; | ||
| import graphql.language.Field; | ||
| import graphql.language.FragmentSpread; | ||
| import graphql.language.Selection; | ||
| import graphql.language.SelectionSet; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| @Slf4j | ||
| public class DataHubFieldComplexityCalculator implements FieldComplexityCalculator { | ||
|
|
||
| private static final String COUNT_ARG = "count"; | ||
| private static final String INPUT_ARG = "input"; | ||
| private static final String SEARCH_RESULTS_FIELD = "searchResults"; | ||
| private static final String ENTITY_FIELD = "entity"; | ||
| private static final String SEARCH_RESULT_FIELDS_FIELD = "searchResultFields"; | ||
| private static final String GRAPHQL_QUERY_TYPE = "Query"; | ||
|
|
||
| @SuppressWarnings("rawtypes") | ||
| @Override | ||
| public int calculate(FieldComplexityEnvironment environment, int childComplexity) { | ||
| int complexity = 1; | ||
| Map<String, Object> args = environment.getArguments(); | ||
| if (args.containsKey(INPUT_ARG)) { | ||
| Map<String, Object> input = (Map<String, Object>) args.get(INPUT_ARG); | ||
| if (input.containsKey(COUNT_ARG) && (Integer) input.get(COUNT_ARG) > 1) { | ||
| Integer count = (Integer) input.get(COUNT_ARG); | ||
| Field field = environment.getField(); | ||
| complexity += countRecursiveLineageComplexity(count, field); | ||
| } | ||
| } | ||
| if (GRAPHQL_QUERY_TYPE.equals(environment.getParentType().getName())) { | ||
| log.info( | ||
| "Query complexity for query: {} is {}", | ||
| environment.getField().getName(), | ||
| complexity + childComplexity); | ||
| } | ||
| return complexity + childComplexity; | ||
| } | ||
|
|
||
| private int countRecursiveLineageComplexity(Integer count, Field field) { | ||
| List<Selection> subFields = field.getSelectionSet().getSelections(); | ||
| Optional<FragmentSpread> searchResultsFieldsField = | ||
| subFields.stream() | ||
| .filter(selection -> selection instanceof Field) | ||
| .map(selection -> (Field) selection) | ||
| .filter(subField -> SEARCH_RESULTS_FIELD.equals(subField.getName())) | ||
| .map(Field::getSelectionSet) | ||
| .map(SelectionSet::getSelections) | ||
| .flatMap(List::stream) | ||
| .filter(selection -> selection instanceof Field) | ||
| .map(selection -> (Field) selection) | ||
| .filter(subField -> ENTITY_FIELD.equals(subField.getName())) | ||
| .map(Field::getSelectionSet) | ||
| .map(SelectionSet::getSelections) | ||
| .flatMap(List::stream) | ||
| .filter(selection -> selection instanceof FragmentSpread) | ||
| .map(selection -> (FragmentSpread) selection) | ||
| .filter(subField -> SEARCH_RESULT_FIELDS_FIELD.equals(subField.getName())) | ||
| .findFirst(); | ||
| if (searchResultsFieldsField.isPresent()) { | ||
| // This fragment includes 2 lineage queries, we account for this additional complexity by | ||
| // multiplying | ||
| // by the count of entities attempting to be returned | ||
| return 2 * count; | ||
| } | ||
| return 0; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.