Skip to content

Commit

Permalink
feat(openapi-v3): support async and createIfNotExists params on aspect
Browse files Browse the repository at this point in the history
  • Loading branch information
@david-leifker
david-leifker committed Oct 12, 2024
1 parent 4133319 commit 62463c7
Show file tree
Hide file tree
Showing 6 changed files with 110 additions and 10 deletions.
1 change: 1 addition & 0 deletions metadata-io/src/main/java/com/linkedin/metadata/entity/EntityServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1357,6 +1357,7 @@ private Stream<IngestResult> ingestProposalSync(
return IngestResult.builder()
.urn(item.getUrn())
.request(item)
.result(result)
.publishedMCL(result.getMclFuture() != null)
.sqlCommitted(true)
.isUpdate(result.getOldValue() != null)
Expand Down
37 changes: 29 additions & 8 deletions ...servlet/src/main/java/io/datahubproject/openapi/controller/GenericEntitiesController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,9 @@ protected abstract E buildGenericEntity(
@Nonnull UpdateAspectResult updateAspectResult,
boolean withSystemMetadata);

protected abstract E buildGenericEntity(
@Nonnull String aspectName, @Nonnull IngestResult ingestResult, boolean withSystemMetadata);

protected abstract AspectsBatch toMCPBatch(
@Nonnull OperationContext opContext, String entityArrayList, Actor actor)
throws JsonProcessingException, InvalidUrnException;
Expand Down Expand Up @@ -560,8 +563,11 @@ public ResponseEntity<E> createAspect(
@PathVariable("entityName") String entityName,
@PathVariable("entityUrn") String entityUrn,
@PathVariable("aspectName") String aspectName,
@RequestParam(value = "async", required = false, defaultValue = "false") Boolean async,
@RequestParam(value = "systemMetadata", required = false, defaultValue = "false")
Boolean withSystemMetadata,
@RequestParam(value = "createIfEntityNotExists", required = false, defaultValue = "false")
Boolean createIfEntityNotExists,
@RequestParam(value = "createIfNotExists", required = false, defaultValue = "true")
Boolean createIfNotExists,
@RequestBody @Nonnull String jsonAspect)
Expand Down Expand Up @@ -591,24 +597,38 @@ public ResponseEntity<E> createAspect(
opContext.getRetrieverContext().get().getAspectRetriever(),
urn,
aspectSpec,
createIfEntityNotExists,
createIfNotExists,
jsonAspect,
authentication.getActor());

List<UpdateAspectResult> results =
entityService.ingestAspects(
Set<IngestResult> results =
entityService.ingestProposal(
opContext,
AspectsBatchImpl.builder()
.retrieverContext(opContext.getRetrieverContext().get())
.items(List.of(upsert))
.build(),
true,
true);
async);

return ResponseEntity.of(
results.stream()
.findFirst()
.map(result -> buildGenericEntity(aspectName, result, withSystemMetadata)));
if (!async) {
return ResponseEntity.of(
results.stream()
.filter(item -> aspectName.equals(item.getRequest().getAspectName()))
.findFirst()
.map(
result ->
buildGenericEntity(aspectName, result.getResult(), withSystemMetadata)));
} else {
return results.stream()
.filter(item -> aspectName.equals(item.getRequest().getAspectName()))
.map(
result ->
ResponseEntity.accepted()
.body(buildGenericEntity(aspectName, result, withSystemMetadata)))
.findFirst()
.orElse(ResponseEntity.accepted().build());
}
}

@Tag(name = "Generic Aspects")
Expand Down Expand Up @@ -789,6 +809,7 @@ protected abstract ChangeMCP toUpsertItem(
@Nonnull AspectRetriever aspectRetriever,
Urn entityUrn,
AspectSpec aspectSpec,
Boolean createIfEntityNotExists,
Boolean createIfNotExists,
String jsonAspect,
Actor actor)
Expand Down
27 changes: 26 additions & 1 deletion ...enapi-servlet/src/main/java/io/datahubproject/openapi/v2/controller/EntityController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,20 @@ protected GenericEntityV2 buildGenericEntity(
withSystemMetadata ? updateAspectResult.getNewSystemMetadata() : null)));
}

@Override
protected GenericEntityV2 buildGenericEntity(
@Nonnull String aspectName, @Nonnull IngestResult ingestResult, boolean withSystemMetadata) {
return GenericEntityV2.builder()
.urn(ingestResult.getUrn().toString())
.build(
objectMapper,
Map.of(
aspectName,
Pair.of(
ingestResult.getRequest().getRecordTemplate(),
withSystemMetadata ? ingestResult.getRequest().getSystemMetadata() : null)));
}

private List<GenericEntityV2> toRecordTemplates(
@Nonnull OperationContext opContext,
SearchEntityArray searchEntities,
Expand Down Expand Up @@ -278,14 +292,25 @@ protected ChangeMCP toUpsertItem(
@Nonnull AspectRetriever aspectRetriever,
Urn entityUrn,
AspectSpec aspectSpec,
Boolean createIfEntityNotExists,
Boolean createIfNotExists,
String jsonAspect,
Actor actor)
throws URISyntaxException {

final ChangeType changeType;
if (Boolean.TRUE.equals(createIfEntityNotExists)) {
changeType = ChangeType.CREATE_ENTITY;
} else if (Boolean.TRUE.equals(createIfNotExists)) {
changeType = ChangeType.CREATE;
} else {
changeType = ChangeType.UPSERT;
}

return ChangeItemImpl.builder()
.urn(entityUrn)
.aspectName(aspectSpec.getName())
.changeType(Boolean.TRUE.equals(createIfNotExists) ? ChangeType.CREATE : ChangeType.UPSERT)
.changeType(changeType)
.auditStamp(AuditStampUtils.createAuditStamp(actor.toUrnStr()))
.recordTemplate(
GenericRecordUtils.deserializeAspect(
Expand Down
22 changes: 22 additions & 0 deletions ...ervice/openapi-servlet/src/main/java/io/datahubproject/openapi/v3/OpenAPIV3Generator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1100,6 +1100,28 @@ private static PathItem buildSingleEntityAspectPath(
new Operation()
.summary(String.format("Create aspect %s on %s ", aspect, upperFirstEntity))
.tags(tags)
.parameters(
List.of(
new Parameter()
.in(NAME_QUERY)
.name("async")
.description("Use async ingestion for high throughput.")
.schema(new Schema().type(TYPE_BOOLEAN)._default(false)),
new Parameter()
.in(NAME_QUERY)
.name(NAME_SYSTEM_METADATA)
.description("Include systemMetadata with response.")
.schema(new Schema().type(TYPE_BOOLEAN)._default(false)),
new Parameter()
.in(NAME_QUERY)
.name("createIfEntityNotExists")
.description("Only create the aspect if the Entity doesn't exist.")
.schema(new Schema().type(TYPE_BOOLEAN)._default(false)),
new Parameter()
.in(NAME_QUERY)
.name("createIfNotExists")
.description("Only create the aspect if the Aspect doesn't exist.")
.schema(new Schema().type(TYPE_BOOLEAN)._default(true))))
.requestBody(requestBody)
.responses(new ApiResponses().addApiResponse("201", successPostResponse));
// Patch Operation
Expand Down
31 changes: 30 additions & 1 deletion ...enapi-servlet/src/main/java/io/datahubproject/openapi/v3/controller/EntityController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,24 @@ protected GenericEntityV3 buildGenericEntity(
.build()));
}

@Override
protected GenericEntityV3 buildGenericEntity(
@Nonnull String aspectName, @Nonnull IngestResult ingestResult, boolean withSystemMetadata) {
return GenericEntityV3.builder()
.build(
objectMapper,
ingestResult.getUrn(),
Map.of(
aspectName,
AspectItem.builder()
.aspect(ingestResult.getRequest().getRecordTemplate())
.systemMetadata(
withSystemMetadata ? ingestResult.getRequest().getSystemMetadata() : null)
.auditStamp(
withSystemMetadata ? ingestResult.getRequest().getAuditStamp() : null)
.build()));
}

private List<GenericEntityV3> toRecordTemplates(
@Nonnull OperationContext opContext,
SearchEntityArray searchEntities,
Expand Down Expand Up @@ -472,16 +490,27 @@ protected ChangeMCP toUpsertItem(
@Nonnull AspectRetriever aspectRetriever,
Urn entityUrn,
AspectSpec aspectSpec,
Boolean createIfEntityNotExists,
Boolean createIfNotExists,
String jsonAspect,
Actor actor)
throws JsonProcessingException {
JsonNode jsonNode = objectMapper.readTree(jsonAspect);
String aspectJson = jsonNode.get("value").toString();

final ChangeType changeType;
if (Boolean.TRUE.equals(createIfEntityNotExists)) {
changeType = ChangeType.CREATE_ENTITY;
} else if (Boolean.TRUE.equals(createIfNotExists)) {
changeType = ChangeType.CREATE;
} else {
changeType = ChangeType.UPSERT;
}

return ChangeItemImpl.builder()
.urn(entityUrn)
.aspectName(aspectSpec.getName())
.changeType(Boolean.TRUE.equals(createIfNotExists) ? ChangeType.CREATE : ChangeType.UPSERT)
.changeType(changeType)
.auditStamp(AuditStampUtils.createAuditStamp(actor.toUrnStr()))
.recordTemplate(
GenericRecordUtils.deserializeAspect(
Expand Down
2 changes: 2 additions & 0 deletions metadata-service/services/src/main/java/com/linkedin/metadata/entity/IngestResult.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import com.linkedin.common.urn.Urn;
import com.linkedin.metadata.aspect.batch.BatchItem;
import javax.annotation.Nullable;
import lombok.Builder;
import lombok.Value;

Expand All @@ -10,6 +11,7 @@
public class IngestResult {
Urn urn;
BatchItem request;
@Nullable UpdateAspectResult result;
boolean publishedMCL;
boolean processedMCL;
boolean publishedMCP;
Expand Down

0 comments on commit 62463c7

Please sign in to comment.