diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/EntityResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/EntityResource.java index 747135cdf7c8..d4443fb4efce 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/EntityResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/EntityResource.java @@ -60,6 +60,7 @@ import org.openmetadata.service.util.AsyncService; import org.openmetadata.service.util.BulkAssetsOperationResponse; import org.openmetadata.service.util.CSVExportResponse; +import org.openmetadata.service.util.CSVImportResponse; import org.openmetadata.service.util.EntityUtil; import org.openmetadata.service.util.EntityUtil.Fields; import org.openmetadata.service.util.RestUtil; @@ -465,6 +466,28 @@ public Response bulkRemoveFromAssetsAsync( return Response.ok().entity(response).type(MediaType.APPLICATION_JSON).build(); } + public Response importCsvInternalAsync( + SecurityContext securityContext, String name, String csv, boolean dryRun) { + OperationContext operationContext = + new OperationContext(entityType, MetadataOperation.EDIT_ALL); + authorizer.authorize(securityContext, operationContext, getResourceContextByName(name)); + String jobId = UUID.randomUUID().toString(); + ExecutorService executorService = AsyncService.getInstance().getExecutorService(); + executorService.submit( + () -> { + try { + CsvImportResult result = importCsvInternal(securityContext, name, csv, dryRun); + WebsocketNotificationHandler.sendCsvImportCompleteNotification( + jobId, securityContext, result); + } catch (Exception e) { + WebsocketNotificationHandler.sendCsvImportFailedNotification( + jobId, securityContext, e.getMessage()); + } + }); + CSVImportResponse response = new CSVImportResponse(jobId, "Import initiated successfully."); + return Response.ok().entity(response).type(MediaType.APPLICATION_JSON).build(); + } + public String exportCsvInternal(SecurityContext securityContext, String name) throws IOException { OperationContext operationContext = new OperationContext(entityType, MetadataOperation.VIEW_ALL); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseResource.java index b64b906ec0d8..e46ce51b8ff7 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseResource.java @@ -505,6 +505,41 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + @Valid + @Operation( + operationId = "importDatabaseAsync", + summary = "Import database schemas from CSV asynchronously", + description = + "Import database schemas from CSV to update database schemas asynchronously (no creation allowed).", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import initiated successfully", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter(description = "Name of the Database", schema = @Schema(type = "string")) + @PathParam("name") + String name, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + @PUT @Path("/{id}/vote") @Operation( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseSchemaResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseSchemaResource.java index 9b6893376684..72bbb666f508 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseSchemaResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/DatabaseSchemaResource.java @@ -476,6 +476,40 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + @Valid + @Operation( + operationId = "importDatabaseSchemaAsync", + summary = + "Import tables from CSV to update database schema asynchronously (no creation allowed)", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import initiated successfully", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter(description = "Name of the Database schema", schema = @Schema(type = "string")) + @PathParam("name") + String name, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + @PUT @Path("/{id}/vote") @Operation( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableResource.java index d8eed3fd1a3d..529aa0769776 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableResource.java @@ -534,6 +534,38 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Valid + @Operation( + operationId = "importTableAsync", + summary = "Import columns from CSV to update table asynchronously (no creation allowed)", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import result", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter(description = "Name of the table", schema = @Schema(type = "string")) + @PathParam("name") + String name, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + @DELETE @Path("/{id}") @Operation( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/glossary/GlossaryResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/glossary/GlossaryResource.java index 018a4288c55b..2947dd330924 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/glossary/GlossaryResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/glossary/GlossaryResource.java @@ -578,6 +578,36 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + @Valid + @Operation( + operationId = "importGlossaryAsync", + summary = "Import glossary in CSV format asynchronously", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import initiated successfully", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter(description = "Name of the glossary", schema = @Schema(type = "string")) + @PathParam("name") + String name, + @RequestBody(description = "CSV data to import", required = true) String csv, + @Parameter(description = "Dry run the import", schema = @Schema(type = "boolean")) + @QueryParam("dryRun") + @DefaultValue("true") + boolean dryRun) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + private Glossary getGlossary(CreateGlossary create, String user) { return getGlossary(repository, create, user); } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/services/database/DatabaseServiceResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/services/database/DatabaseServiceResource.java index aea1a6c441f4..adf3fd7b8d59 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/services/database/DatabaseServiceResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/services/database/DatabaseServiceResource.java @@ -523,6 +523,40 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + @Valid + @Operation( + operationId = "importDatabaseServiceAsync", + summary = + "Import service from CSV to update database service asynchronously (no creation allowed)", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import initiated successfully", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter(description = "Name of the Database Service", schema = @Schema(type = "string")) + @PathParam("name") + String name, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + @DELETE @Path("/{id}") @Operation( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/TeamResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/TeamResource.java index 61c6a6ecce4d..4017a850a6bc 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/TeamResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/TeamResource.java @@ -666,6 +666,37 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, name, csv, dryRun); } + @PUT + @Path("/name/{name}/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + @Valid + @Operation( + operationId = "importTeamsAsync", + summary = "Import from CSV to create, and update teams asynchronously.", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import initiated successfully", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @PathParam("name") String name, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, name, csv, dryRun); + } + private Team getTeam(CreateTeam ct, String user) { if (ct.getTeamType().equals(TeamType.ORGANIZATION)) { throw new IllegalArgumentException(CREATE_ORGANIZATION); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/UserResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/UserResource.java index 85bc14878363..c84754d71151 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/UserResource.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/teams/UserResource.java @@ -1482,6 +1482,41 @@ public CsvImportResult importCsv( return importCsvInternal(securityContext, team, csv, dryRun); } + @PUT + @Path("/importAsync") + @Consumes(MediaType.TEXT_PLAIN) + @Valid + @Operation( + operationId = "importTeamsAsync", + summary = "Import from CSV to create, and update teams asynchronously.", + responses = { + @ApiResponse( + responseCode = "200", + description = "Import result", + content = + @Content( + mediaType = "application/json", + schema = @Schema(implementation = CsvImportResult.class))) + }) + public Response importCsvAsync( + @Context SecurityContext securityContext, + @Parameter( + description = "Name of the team to under which the users are imported to", + required = true, + schema = @Schema(type = "string")) + @QueryParam("team") + String team, + @Parameter( + description = + "Dry-run when true is used for validating the CSV without really importing it. (default=true)", + schema = @Schema(type = "boolean")) + @DefaultValue("true") + @QueryParam("dryRun") + boolean dryRun, + String csv) { + return importCsvInternalAsync(securityContext, team, csv, dryRun); + } + public void validateEmailAlreadyExists(String email) { if (repository.checkEmailAlreadyExists(email)) { throw new CustomExceptionMessage( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/search/security/RBACConditionEvaluator.java b/openmetadata-service/src/main/java/org/openmetadata/service/search/security/RBACConditionEvaluator.java index ac319e8cf9b9..f2f66a008e9f 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/search/security/RBACConditionEvaluator.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/search/security/RBACConditionEvaluator.java @@ -235,7 +235,12 @@ private List extractMethodArguments(MethodReference methodRef) { public void matchAnyTag(List tags, ConditionCollector collector) { List tagQueries = new ArrayList<>(); for (String tag : tags) { - OMQueryBuilder tagQuery = queryBuilderFactory.termQuery("tags.tagFQN", tag); + OMQueryBuilder tagQuery; + if (tag.startsWith("Tier")) { + tagQuery = queryBuilderFactory.termQuery("tier.tagFQN", tag); + } else { + tagQuery = queryBuilderFactory.termQuery("tags.tagFQN", tag); + } tagQueries.add(tagQuery); } OMQueryBuilder tagQueryCombined; diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/security/policyevaluator/PolicyEvaluator.java b/openmetadata-service/src/main/java/org/openmetadata/service/security/policyevaluator/PolicyEvaluator.java index ae45e5ae613d..fe618cd08f04 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/security/policyevaluator/PolicyEvaluator.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/security/policyevaluator/PolicyEvaluator.java @@ -84,7 +84,9 @@ public static void hasDomainPermission( @NonNull ResourceContextInterface resourceContext, OperationContext operationContext) { // If the Resource Does not belong to any Domain, then evaluate via other permissions - if (!nullOrEmpty(resourceContext.getDomain())) { + if (!nullOrEmpty(resourceContext.getDomain()) + && !subjectContext.isAdmin() + && !subjectContext.isBot()) { EntityReference domain = resourceContext.getDomain(); if (!subjectContext.hasDomain(domain)) { throw new AuthorizationException( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/socket/WebSocketManager.java b/openmetadata-service/src/main/java/org/openmetadata/service/socket/WebSocketManager.java index 1988ee5edc89..d4f6ff9ebf01 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/socket/WebSocketManager.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/socket/WebSocketManager.java @@ -29,6 +29,7 @@ public class WebSocketManager { public static final String MENTION_CHANNEL = "mentionChannel"; public static final String ANNOUNCEMENT_CHANNEL = "announcementChannel"; public static final String CSV_EXPORT_CHANNEL = "csvExportChannel"; + public static final String CSV_IMPORT_CHANNEL = "csvImportChannel"; public static final String BULK_ASSETS_CHANNEL = "bulkAssetsChannel"; diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportMessage.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportMessage.java new file mode 100644 index 000000000000..0ed899ad1f77 --- /dev/null +++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportMessage.java @@ -0,0 +1,21 @@ +package org.openmetadata.service.util; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import org.openmetadata.schema.type.csv.CsvImportResult; + +@NoArgsConstructor +public class CSVImportMessage { + @Getter @Setter private String jobId; + @Getter @Setter private String status; + @Getter @Setter private CsvImportResult result; + @Getter @Setter private String error; + + public CSVImportMessage(String jobId, String status, CsvImportResult result, String error) { + this.jobId = jobId; + this.status = status; + this.result = result; + this.error = error; + } +} diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportResponse.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportResponse.java new file mode 100644 index 000000000000..cde3c65dab12 --- /dev/null +++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/CSVImportResponse.java @@ -0,0 +1,16 @@ +package org.openmetadata.service.util; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +@NoArgsConstructor +public class CSVImportResponse { + @Getter @Setter private String jobId; + @Getter @Setter private String message; + + public CSVImportResponse(String jobId, String message) { + this.jobId = jobId; + this.message = message; + } +} diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/WebsocketNotificationHandler.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/WebsocketNotificationHandler.java index bfac09b6e0a6..ceecb6265c7a 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/util/WebsocketNotificationHandler.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/WebsocketNotificationHandler.java @@ -35,6 +35,7 @@ import org.openmetadata.schema.type.Post; import org.openmetadata.schema.type.Relationship; import org.openmetadata.schema.type.api.BulkOperationResult; +import org.openmetadata.schema.type.csv.CsvImportResult; import org.openmetadata.service.Entity; import org.openmetadata.service.jdbi3.CollectionDAO; import org.openmetadata.service.resources.feeds.MessageParser; @@ -194,4 +195,22 @@ private static UUID getUserIdFromSecurityContext(SecurityContext securityContext User user = Entity.getCollectionDAO().userDAO().findEntityByName(username); return user.getId(); } + + public static void sendCsvImportCompleteNotification( + String jobId, SecurityContext securityContext, CsvImportResult result) { + CSVImportMessage message = new CSVImportMessage(jobId, "COMPLETED", result, null); + String jsonMessage = JsonUtils.pojoToJson(message); + UUID userId = getUserIdFromSecurityContext(securityContext); + WebSocketManager.getInstance() + .sendToOne(userId, WebSocketManager.CSV_IMPORT_CHANNEL, jsonMessage); + } + + public static void sendCsvImportFailedNotification( + String jobId, SecurityContext securityContext, String errorMessage) { + CSVExportMessage message = new CSVExportMessage(jobId, "FAILED", null, errorMessage); + String jsonMessage = JsonUtils.pojoToJson(message); + UUID userId = getUserIdFromSecurityContext(securityContext); + WebSocketManager.getInstance() + .sendToOne(userId, WebSocketManager.CSV_IMPORT_CHANNEL, jsonMessage); + } } diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java index e6384bbcbbb7..c00e20069def 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java @@ -219,6 +219,8 @@ import org.openmetadata.service.security.SecurityUtil; import org.openmetadata.service.util.CSVExportMessage; import org.openmetadata.service.util.CSVExportResponse; +import org.openmetadata.service.util.CSVImportMessage; +import org.openmetadata.service.util.CSVImportResponse; import org.openmetadata.service.util.EntityUtil; import org.openmetadata.service.util.FullyQualifiedName; import org.openmetadata.service.util.JsonUtils; @@ -4069,6 +4071,78 @@ private String receiveCsvViaSocketIO(String entityName) throws Exception { return null; } + private String receiveCsvImportViaSocketIO(String entityName, String csv, boolean dryRun) + throws Exception { + UUID userId = getAdminUserId(); + String uri = String.format("http://localhost:%d", APP.getLocalPort()); + + IO.Options options = new IO.Options(); + options.path = "/api/v1/push/feed"; + options.query = "userId=" + userId.toString(); + options.transports = new String[] {"websocket"}; + options.reconnection = false; + options.timeout = 10000; // 10 seconds + + Socket socket = IO.socket(uri, options); + + CountDownLatch connectLatch = new CountDownLatch(1); + CountDownLatch messageLatch = new CountDownLatch(1); + final String[] receivedMessage = new String[1]; + + socket + .on( + Socket.EVENT_CONNECT, + args -> { + System.out.println("Connected to Socket.IO server"); + connectLatch.countDown(); + }) + .on( + "csvImportChannel", + args -> { + receivedMessage[0] = (String) args[0]; + System.out.println("Received message: " + receivedMessage[0]); + messageLatch.countDown(); + socket.disconnect(); + }) + .on( + Socket.EVENT_CONNECT_ERROR, + args -> { + System.err.println("Socket.IO connect error: " + args[0]); + connectLatch.countDown(); + messageLatch.countDown(); + }) + .on( + Socket.EVENT_DISCONNECT, + args -> { + System.out.println("Disconnected from Socket.IO server"); + }); + + socket.connect(); + if (!connectLatch.await(10, TimeUnit.SECONDS)) { + fail("Could not connect to Socket.IO server"); + } + String jobId = initiateImport(entityName, csv, dryRun); + + if (!messageLatch.await(45, TimeUnit.SECONDS)) { + fail("Did not receive CSV import result via Socket.IO within the expected time."); + } + + String receivedJson = receivedMessage[0]; + if (receivedJson == null) { + fail("Received message is null."); + } + + CSVImportMessage csvImportMessage = JsonUtils.readValue(receivedJson, CSVImportMessage.class); + if ("COMPLETED".equals(csvImportMessage.getStatus())) { + return JsonUtils.pojoToJson(csvImportMessage.getResult()); + } else if ("FAILED".equals(csvImportMessage.getStatus())) { + fail("CSV import failed: " + csvImportMessage.getError()); + } else { + fail("Unknown status received: " + csvImportMessage.getStatus()); + } + return null; + } + private UUID getAdminUserId() throws HttpResponseException { UserResourceTest userResourceTest = new UserResourceTest(); User adminUser = userResourceTest.getEntityByName("admin", ADMIN_AUTH_HEADERS); @@ -4083,6 +4157,15 @@ protected String initiateExport(String entityName) throws IOException { return response.getJobId(); } + protected String initiateImport(String entityName, String csv, boolean dryRun) + throws IOException { + WebTarget target = getResourceByName(entityName + "/importAsync"); + target = !dryRun ? target.queryParam("dryRun", false) : target; + CSVImportResponse response = + TestUtils.putCsv(target, csv, CSVImportResponse.class, Status.OK, ADMIN_AUTH_HEADERS); + return response.getJobId(); + } + @SneakyThrows protected void importCsvAndValidate( String entityName, @@ -4092,26 +4175,38 @@ protected void importCsvAndValidate( createRecords = listOrEmpty(createRecords); updateRecords = listOrEmpty(updateRecords); - // Import CSV to create new records and update existing records with dryRun=true first String csv = EntityCsvTest.createCsv(csvHeaders, createRecords, updateRecords); - Awaitility.await().atMost(4, TimeUnit.SECONDS).until(() -> true); - CsvImportResult dryRunResult = importCsv(entityName, csv, true); - Awaitility.await().atMost(4, TimeUnit.SECONDS).until(() -> true); + + CsvImportResult dryRunResultAsync = + JsonUtils.readValue( + receiveCsvImportViaSocketIO(entityName, csv, true), CsvImportResult.class); + CsvImportResult dryRunResultSync = importCsv(entityName, csv, true); // Validate the imported result summary - it should include both created and updated records int totalRows = 1 + createRecords.size() + updateRecords.size(); - assertSummary(dryRunResult, ApiStatus.SUCCESS, totalRows, totalRows, 0); + assertSummary(dryRunResultSync, ApiStatus.SUCCESS, totalRows, totalRows, 0); + assertSummary(dryRunResultAsync, ApiStatus.SUCCESS, totalRows, totalRows, 0); String expectedResultsCsv = EntityCsvTest.createCsvResult(csvHeaders, createRecords, updateRecords); - assertEquals(expectedResultsCsv, dryRunResult.getImportResultsCsv()); + assertEquals(expectedResultsCsv, dryRunResultSync.getImportResultsCsv()); + assertEquals(expectedResultsCsv, dryRunResultAsync.getImportResultsCsv()); - // Import CSV to create new records and update existing records with dryRun=false to really - // import the data - CsvImportResult result = importCsv(entityName, csv, false); - Awaitility.await().atMost(4, TimeUnit.SECONDS).until(() -> true); - assertEquals(dryRunResult.withDryRun(false), result); + CsvImportResult finalResultAsync = + JsonUtils.readValue( + receiveCsvImportViaSocketIO(entityName, csv, false), CsvImportResult.class); + CsvImportResult finalResultSync = importCsv(entityName, csv, false); + + assertEquals(dryRunResultAsync.withDryRun(false), finalResultAsync); + // entities have created in the earlier sync import (dryRun=false) so the next import agan will + // have entities updated + assertEquals( + dryRunResultSync + .withImportResultsCsv( + dryRunResultSync.getImportResultsCsv().replace("Entity created", "Entity updated")) + .withDryRun(false), + finalResultSync); - // Finally, export CSV and ensure the exported CSV is same as imported CSV + // Finally, export CSV and ensure the exported CSV is the same as imported CSV String exportedCsvAsync = receiveCsvViaSocketIO(entityName); CsvUtilTest.assertCsv(csv, exportedCsvAsync); String exportedCsvSync = exportCsv(entityName); diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/teams/UserResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/teams/UserResourceTest.java index bf5f7802f4d6..84de8fe69a33 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/teams/UserResourceTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/teams/UserResourceTest.java @@ -140,6 +140,7 @@ import org.openmetadata.service.security.AuthenticationException; import org.openmetadata.service.security.mask.PIIMasker; import org.openmetadata.service.util.CSVExportResponse; +import org.openmetadata.service.util.CSVImportResponse; import org.openmetadata.service.util.EntityUtil; import org.openmetadata.service.util.JsonUtils; import org.openmetadata.service.util.PasswordUtil; @@ -1576,4 +1577,14 @@ protected String initiateExport(String teamName) throws HttpResponseException { target, CSVExportResponse.class, ADMIN_AUTH_HEADERS, Status.ACCEPTED.getStatusCode()); return response.getJobId(); } + + @Override + protected String initiateImport(String teamName, String csv, boolean dryRun) throws IOException { + WebTarget target = getCollection().path("/importAsync"); + target = target.queryParam("team", teamName); + target = !dryRun ? target.queryParam("dryRun", false) : target; + CSVImportResponse response = + TestUtils.putCsv(target, csv, CSVImportResponse.class, Status.OK, ADMIN_AUTH_HEADERS); + return response.getJobId(); + } } diff --git a/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/DataQualityAndProfiler.spec.ts b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/DataQualityAndProfiler.spec.ts index 13017484b9e8..a1578a45eee4 100644 --- a/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/DataQualityAndProfiler.spec.ts +++ b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/DataQualityAndProfiler.spec.ts @@ -191,7 +191,7 @@ test('Column test case', PLAYWRIGHT_INGESTION_TAG_OBJ, async ({ page }) => { await testDefinitionResponse; await page.fill('#tableTestForm_testName', NEW_COLUMN_TEST_CASE.name); await page.click('#tableTestForm_testTypeId'); - await page.click(`[title="${NEW_COLUMN_TEST_CASE.label}"]`); + await page.click(`[data-testid="${NEW_COLUMN_TEST_CASE.type}"]`); await page.fill( '#tableTestForm_params_minLength', NEW_COLUMN_TEST_CASE.min diff --git a/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/TestCases.spec.ts b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/TestCases.spec.ts index f9bad592a9e4..56bef624f085 100644 --- a/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/TestCases.spec.ts +++ b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Pages/TestCases.spec.ts @@ -56,7 +56,7 @@ test('Table difference test case', async ({ page }) => { const tableListSearchResponse = page.waitForResponse( `/api/v1/search/query?q=*index=table_search_index*` ); - await page.getByTitle('Compare 2 tables for').click(); + await page.getByTestId('tableDiff').click(); await tableListSearchResponse; await page.click('#tableTestForm_params_table2'); const tableSearchResponse = page.waitForResponse( @@ -180,7 +180,7 @@ test('Custom SQL Query', async ({ page }) => { await page.getByTestId('test-case').click(); await page.getByTestId('test-case-name').fill(testCase.name); await page.getByTestId('test-type').click(); - await page.getByTitle('Custom SQL Query').click(); + await page.getByTestId('tableCustomSQLQuery').click(); await page.click('#tableTestForm_params_strategy'); await page.locator('.CodeMirror-scroll').click(); await page @@ -289,7 +289,7 @@ test('Column Values To Be Not Null', async ({ page }) => { NEW_COLUMN_TEST_CASE_WITH_NULL_TYPE.type ); await page.click( - `[title="${NEW_COLUMN_TEST_CASE_WITH_NULL_TYPE.label}"]` + `[data-testid="${NEW_COLUMN_TEST_CASE_WITH_NULL_TYPE.type}"]` ); await page.fill( descriptionBox, diff --git a/openmetadata-ui/src/main/resources/ui/src/assets/svg/no-dimension-icon.svg b/openmetadata-ui/src/main/resources/ui/src/assets/svg/no-dimension-icon.svg new file mode 100644 index 000000000000..c46b8bb69644 --- /dev/null +++ b/openmetadata-ui/src/main/resources/ui/src/assets/svg/no-dimension-icon.svg @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/openmetadata-ui/src/main/resources/ui/src/components/BulkImport/BulkEntityImport.component.tsx b/openmetadata-ui/src/main/resources/ui/src/components/BulkImport/BulkEntityImport.component.tsx index d1067f96632b..fffceaffac1c 100644 --- a/openmetadata-ui/src/main/resources/ui/src/components/BulkImport/BulkEntityImport.component.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/components/BulkImport/BulkEntityImport.component.tsx @@ -18,7 +18,14 @@ import { } from '@inovua/reactdatagrid-community/types'; import { Button, Card, Col, Row, Space, Typography } from 'antd'; import { AxiosError } from 'axios'; -import React, { MutableRefObject, useCallback, useMemo, useState } from 'react'; +import React, { + MutableRefObject, + useCallback, + useEffect, + useMemo, + useRef, + useState, +} from 'react'; import { useTranslation } from 'react-i18next'; import { usePapaParse } from 'react-papaparse'; @@ -27,6 +34,8 @@ import { ENTITY_IMPORT_STEPS, VALIDATION_STEP, } from '../../constants/BulkImport.constant'; +import { SOCKET_EVENTS } from '../../constants/constants'; +import { useWebSocketConnector } from '../../context/WebSocketProvider/WebSocketProvider'; import { CSVImportResult } from '../../generated/type/csvImportResult'; import { getCSVStringFromColumnsAndDataSource, @@ -34,11 +43,16 @@ import { } from '../../utils/CSV/CSV.utils'; import csvUtilsClassBase from '../../utils/CSV/CSVUtilsClassBase'; import { showErrorToast, showSuccessToast } from '../../utils/ToastUtils'; +import Banner from '../common/Banner/Banner'; import { ImportStatus } from '../common/EntityImport/ImportStatus/ImportStatus.component'; import Stepper from '../Settings/Services/Ingestion/IngestionStepper/IngestionStepper.component'; import { UploadFile } from '../UploadFile/UploadFile'; import './bulk-entity-import.style.less'; -import { BulkImportProps } from './BulkEntityImport.interface'; +import { + BulkImportProps, + CSVImportAsyncWebsocketResponse, + CSVImportJobType, +} from './BulkEntityImport.interface'; let inEdit = false; @@ -48,9 +62,17 @@ const BulkEntityImport = ({ onValidateCsvString, onSuccess, }: BulkImportProps) => { + const { socket } = useWebSocketConnector(); + const [activeAsyncImportJob, setActiveAsyncImportJob] = + useState(); + const activeAsyncImportJobRef = useRef(); + const [activeStep, setActiveStep] = useState( VALIDATION_STEP.UPLOAD ); + + const activeStepRef = useRef(VALIDATION_STEP.UPLOAD); + const { t } = useTranslation(); const [isValidating, setIsValidating] = useState(false); const [validationData, setValidationData] = useState(); @@ -80,6 +102,14 @@ const BulkEntityImport = ({ }); }, [setGridRef]); + const handleActiveStepChange = useCallback( + (step: VALIDATION_STEP) => { + setActiveStep(step); + activeStepRef.current = step; + }, + [setActiveStep, activeStepRef] + ); + const onCSVReadComplete = useCallback( (results: { data: string[][] }) => { // results.data is returning data with unknown type @@ -90,10 +120,10 @@ const BulkEntityImport = ({ setDataSource(dataSource); setColumns(columns); - setActiveStep(VALIDATION_STEP.EDIT_VALIDATE); + handleActiveStepChange(VALIDATION_STEP.EDIT_VALIDATE); setTimeout(focusToGrid, 500); }, - [entityType, setDataSource, setColumns, setActiveStep, focusToGrid] + [entityType, setDataSource, setColumns, handleActiveStepChange, focusToGrid] ); const handleLoadData = useCallback( @@ -102,22 +132,14 @@ const BulkEntityImport = ({ const result = e.target?.result as string; const validationResponse = await onValidateCsvString(result, true); - - if (['failure', 'aborted'].includes(validationResponse?.status ?? '')) { - setValidationData(validationResponse); - - setActiveStep(VALIDATION_STEP.UPLOAD); - - return; - } - - if (result) { - readString(result, { - worker: true, - skipEmptyLines: true, - complete: onCSVReadComplete, - }); - } + const jobData: CSVImportJobType = { + ...validationResponse, + type: 'initialLoad', + initialResult: result, + }; + + setActiveAsyncImportJob(jobData); + activeAsyncImportJobRef.current = jobData; } catch (error) { showErrorToast(error as AxiosError); } @@ -137,9 +159,9 @@ const BulkEntityImport = ({ const handleBack = () => { if (activeStep === VALIDATION_STEP.UPDATE) { - setActiveStep(VALIDATION_STEP.EDIT_VALIDATE); + handleActiveStepChange(VALIDATION_STEP.EDIT_VALIDATE); } else { - setActiveStep(VALIDATION_STEP.UPLOAD); + handleActiveStepChange(VALIDATION_STEP.UPLOAD); } }; @@ -155,52 +177,15 @@ const BulkEntityImport = ({ activeStep === VALIDATION_STEP.EDIT_VALIDATE ); - if (activeStep === VALIDATION_STEP.UPDATE) { - if (response?.status === 'failure') { - setValidationData(response); - readString(response?.importResultsCsv ?? '', { - worker: true, - skipEmptyLines: true, - complete: (results) => { - // results.data is returning data with unknown type - setValidateCSVData( - getEntityColumnsAndDataSourceFromCSV( - results.data as string[][], - entityType - ) - ); - }, - }); - setActiveStep(VALIDATION_STEP.UPDATE); - } else { - showSuccessToast( - t('message.entity-details-updated', { - entityType: capitalize(entityType), - fqn, - }) - ); - onSuccess(); - } - } else if (activeStep === VALIDATION_STEP.EDIT_VALIDATE) { - setValidationData(response); - setActiveStep(VALIDATION_STEP.UPDATE); - readString(response?.importResultsCsv ?? '', { - worker: true, - skipEmptyLines: true, - complete: (results) => { - // results.data is returning data with unknown type - setValidateCSVData( - getEntityColumnsAndDataSourceFromCSV( - results.data as string[][], - entityType - ) - ); - }, - }); - } + const jobData: CSVImportJobType = { + ...response, + type: 'onValidate', + }; + + setActiveAsyncImportJob(jobData); + activeAsyncImportJobRef.current = jobData; } catch (error) { showErrorToast(error as AxiosError); - } finally { setIsValidating(false); } }; @@ -286,14 +271,179 @@ const BulkEntityImport = ({ const handleRetryCsvUpload = () => { setValidationData(undefined); - setActiveStep(VALIDATION_STEP.UPLOAD); + handleActiveStepChange(VALIDATION_STEP.UPLOAD); }; + const handleResetImportJob = useCallback(() => { + setActiveAsyncImportJob(undefined); + activeAsyncImportJobRef.current = undefined; + }, [setActiveAsyncImportJob, activeAsyncImportJobRef]); + + const handleImportWebsocketResponseWithActiveStep = useCallback( + (importResults: CSVImportResult) => { + const activeStep = activeStepRef.current; + + if (activeStep === VALIDATION_STEP.UPDATE) { + if (importResults?.status === 'failure') { + setValidationData(importResults); + readString(importResults?.importResultsCsv ?? '', { + worker: true, + skipEmptyLines: true, + complete: (results) => { + // results.data is returning data with unknown type + setValidateCSVData( + getEntityColumnsAndDataSourceFromCSV( + results.data as string[][], + entityType + ) + ); + }, + }); + handleActiveStepChange(VALIDATION_STEP.UPDATE); + setIsValidating(false); + } else { + showSuccessToast( + t('message.entity-details-updated', { + entityType: capitalize(entityType), + fqn, + }) + ); + onSuccess(); + handleResetImportJob(); + setIsValidating(false); + } + } else if (activeStep === VALIDATION_STEP.EDIT_VALIDATE) { + setValidationData(importResults); + handleActiveStepChange(VALIDATION_STEP.UPDATE); + readString(importResults?.importResultsCsv ?? '', { + worker: true, + skipEmptyLines: true, + complete: (results) => { + // results.data is returning data with unknown type + setValidateCSVData( + getEntityColumnsAndDataSourceFromCSV( + results.data as string[][], + entityType + ) + ); + }, + }); + handleResetImportJob(); + setIsValidating(false); + } + }, + [ + activeStepRef, + entityType, + fqn, + onSuccess, + handleResetImportJob, + handleActiveStepChange, + ] + ); + + const handleImportWebsocketResponse = useCallback( + (websocketResponse: CSVImportAsyncWebsocketResponse) => { + if (!websocketResponse.jobId) { + return; + } + + const activeImportJob = activeAsyncImportJobRef.current; + + if (websocketResponse.jobId === activeImportJob?.jobId) { + setActiveAsyncImportJob((job) => { + if (!job) { + return; + } + + return { + ...job, + ...websocketResponse, + }; + }); + + if (websocketResponse.status === 'COMPLETED') { + const importResults = websocketResponse.result; + + // If the job is complete and the status is either failure or aborted + // then reset the validation data and active step + if (['failure', 'aborted'].includes(importResults?.status ?? '')) { + setValidationData(importResults); + + handleActiveStepChange(VALIDATION_STEP.UPLOAD); + + handleResetImportJob(); + + return; + } + + // If the job is complete and the status is success + // and job was for initial load then check if the initial result is available + // and then read the initial result + if ( + activeImportJob.type === 'initialLoad' && + activeImportJob.initialResult + ) { + readString(activeImportJob.initialResult, { + worker: true, + skipEmptyLines: true, + complete: onCSVReadComplete, + }); + + handleResetImportJob(); + + return; + } + + handleImportWebsocketResponseWithActiveStep(importResults); + } + } + }, + [ + activeStepRef, + activeAsyncImportJobRef, + onCSVReadComplete, + setActiveAsyncImportJob, + handleResetImportJob, + handleActiveStepChange, + ] + ); + + useEffect(() => { + if (socket) { + socket.on(SOCKET_EVENTS.CSV_IMPORT_CHANNEL, (importResponse) => { + if (importResponse) { + const importResponseData = JSON.parse( + importResponse + ) as CSVImportAsyncWebsocketResponse; + + handleImportWebsocketResponse(importResponseData); + } + }); + } + + return () => { + socket && socket.off(SOCKET_EVENTS.CSV_IMPORT_CHANNEL); + }; + }, [socket]); + return ( - + + + {activeAsyncImportJob?.jobId && ( + + )} + {activeStep === 0 && ( <> @@ -371,12 +521,14 @@ const BulkEntityImport = ({ )}
{activeStep > 0 && ( - + )} {activeStep < 3 && ( - - - - ) : ( - // added extra margin to prevent data lost due to fixed footer at bottom -
- - - - - {children} - - - - - {!isFailure && ( - - )} - - -
- )} - - )} - - {activeStep > 2 && ( - + + )} + {activeStep === 2 && !isUndefined(csvImportResult) && ( + + {isAborted ? ( - - - - {fileName}{' '} - {`${t('label.successfully-uploaded')}.`} + + {t('label.aborted')}{' '} + {csvImportResult.abortReason} - - )} - - )} + ) : ( + // added extra margin to prevent data lost due to fixed footer at bottom +
+ + {importStartedBanner} + + + + {children} + + + + + {!isFailure && ( + + )} + + +
+ )} + + )} + + {activeStep > 2 && ( + + + + + + + {fileName}{' '} + {`${t('label.successfully-uploaded')}.`} + + + + + + + + )} + ); }; diff --git a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.interface.ts b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.interface.ts index 7a1495a09eb1..1e0142cd4125 100644 --- a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.interface.ts +++ b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.interface.ts @@ -12,6 +12,7 @@ */ import React from 'react'; import { CSVImportResult } from '../../../generated/type/csvImportResult'; +import { CSVImportAsyncResponse } from '../../BulkImport/BulkEntityImport.interface'; export interface EntityImportProps { entityName: string; @@ -19,8 +20,9 @@ export interface EntityImportProps { name: string, data: string, dryRun?: boolean - ) => Promise; + ) => Promise; onSuccess: () => void; onCancel: () => void; + onCsvResultUpdate?: (result: CSVImportResult) => void; children: React.ReactNode; } diff --git a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.test.tsx b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.test.tsx index 751b403ad25b..efb465644424 100644 --- a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.test.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/EntityImport.test.tsx @@ -18,6 +18,7 @@ import { waitForElement, } from '@testing-library/react'; import React from 'react'; +import { useWebSocketConnector } from '../../../context/WebSocketProvider/WebSocketProvider'; import { CSVImportResult, Status, @@ -35,11 +36,16 @@ let mockCsvImportResult = { success,Entity created,,Glossary2 term2,Glossary2 term2,Description data.,,,,\r`, } as CSVImportResult; +const mockAsyncImportJob = { + jobId: '123', + message: 'Import initiated successfully', +}; + const mockProps = { entityName: 'Business Glossary', onImport: jest .fn() - .mockImplementation(() => Promise.resolve(mockCsvImportResult)), + .mockImplementation(() => Promise.resolve(mockAsyncImportJob)), onSuccess: jest.fn(), onCancel: jest.fn(), }; @@ -65,7 +71,26 @@ jest.mock('./ImportStatus/ImportStatus.component', () => ({ ImportStatus: jest.fn().mockImplementation(() =>
ImportStatus
), })); +jest.mock('../../../context/WebSocketProvider/WebSocketProvider', () => ({ + useWebSocketConnector: jest.fn(), +})); + +const mockSocket = { + on: jest.fn(), + off: jest.fn(), +}; + describe('EntityImport component', () => { + beforeEach(() => { + (useWebSocketConnector as jest.Mock).mockReturnValue({ + socket: mockSocket, + }); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + it('Component should render', async () => { render(ImportTableData); @@ -111,6 +136,22 @@ describe('EntityImport component', () => { await flushPromises(); }); + expect( + await screen.findByText(mockAsyncImportJob.message) + ).toBeInTheDocument(); + + const mockResponse = { + jobId: mockAsyncImportJob.jobId, + status: 'COMPLETED', + result: mockCsvImportResult, + error: null, + }; + + const callback = mockSocket.on.mock.calls[0][1]; + act(() => { + callback(JSON.stringify(mockResponse)); + }); + const importButton = await screen.findByTestId('import-button'); expect(await screen.findByTestId('import-results')).toBeInTheDocument(); @@ -120,6 +161,11 @@ describe('EntityImport component', () => { fireEvent.click(importButton); }); + const callback1 = mockSocket.on.mock.calls[0][1]; + act(() => { + callback1(JSON.stringify(mockResponse)); + }); + const successBadge = await screen.findByTestId('success-badge'); const previewButton = await screen.findByTestId('preview-button'); const fileName = await screen.findByTestId('file-name'); @@ -161,6 +207,22 @@ describe('EntityImport component', () => { await flushPromises(); }); + expect( + await screen.findByText(mockAsyncImportJob.message) + ).toBeInTheDocument(); + + const mockResponse = { + jobId: mockAsyncImportJob.jobId, + status: 'COMPLETED', + result: mockCsvImportResult, + error: null, + }; + + const callback = mockSocket.on.mock.calls[0][1]; + act(() => { + callback(JSON.stringify(mockResponse)); + }); + const importButton = await screen.findByTestId('import-button'); expect(await screen.findByTestId('import-results')).toBeInTheDocument(); @@ -170,6 +232,11 @@ describe('EntityImport component', () => { fireEvent.click(importButton); }); + const callback1 = mockSocket.on.mock.calls[0][1]; + act(() => { + callback1(JSON.stringify(mockResponse)); + }); + const successBadge = await screen.findByTestId('success-badge'); const previewButton = await screen.findByTestId('preview-button'); const fileName = await screen.findByTestId('file-name'); @@ -209,6 +276,22 @@ describe('EntityImport component', () => { await flushPromises(); }); + expect( + await screen.findByText(mockAsyncImportJob.message) + ).toBeInTheDocument(); + + const mockResponse = { + jobId: mockAsyncImportJob.jobId, + status: 'COMPLETED', + result: mockCsvImportResult, + error: null, + }; + + const callback = mockSocket.on.mock.calls[0][1]; + act(() => { + callback(JSON.stringify(mockResponse)); + }); + const importButton = screen.queryByTestId('import-button'); const cancelPreviewButton = await screen.findByTestId( 'preview-cancel-button' @@ -249,6 +332,22 @@ describe('EntityImport component', () => { await flushPromises(); }); + expect( + await screen.findByText(mockAsyncImportJob.message) + ).toBeInTheDocument(); + + const mockResponse = { + jobId: mockAsyncImportJob.jobId, + status: 'COMPLETED', + result: mockCsvImportResult, + error: null, + }; + + const callback = mockSocket.on.mock.calls[0][1]; + act(() => { + callback(JSON.stringify(mockResponse)); + }); + const abortedReason = await screen.findByTestId('abort-reason'); const cancelButton = await screen.findByTestId('cancel-button'); diff --git a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/entity-import.style.less b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/entity-import.style.less index 665e1e4ffae9..12df520543eb 100644 --- a/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/entity-import.style.less +++ b/openmetadata-ui/src/main/resources/ui/src/components/common/EntityImport/entity-import.style.less @@ -33,7 +33,6 @@ } background: @white; height: 360px; - margin: 24px 48px 0; width: unset; } } diff --git a/openmetadata-ui/src/main/resources/ui/src/constants/constants.ts b/openmetadata-ui/src/main/resources/ui/src/constants/constants.ts index 591d9de78939..1752e79f02f2 100644 --- a/openmetadata-ui/src/main/resources/ui/src/constants/constants.ts +++ b/openmetadata-ui/src/main/resources/ui/src/constants/constants.ts @@ -301,6 +301,7 @@ export const SOCKET_EVENTS = { SEARCH_INDEX_JOB_BROADCAST_CHANNEL: 'searchIndexJobStatus', DATA_INSIGHTS_JOB_BROADCAST_CHANNEL: 'dataInsightsJobStatus', BULK_ASSETS_CHANNEL: 'bulkAssetsChannel', + CSV_IMPORT_CHANNEL: 'csvImportChannel', }; export const IN_PAGE_SEARCH_ROUTES: Record> = { diff --git a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.interface.tsx b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.interface.tsx index a87737318805..61911a4a29aa 100644 --- a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.interface.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.interface.tsx @@ -47,6 +47,7 @@ export type UpstreamDownstreamData = { export interface LineageContextType { reactFlowInstance?: ReactFlowInstance; + dataQualityLineage?: EntityLineageResponse; nodes: Node[]; edges: Edge[]; tracedNodes: string[]; diff --git a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.test.tsx b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.test.tsx index 33b147c22c02..ac1ed7562f06 100644 --- a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.test.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.test.tsx @@ -11,6 +11,7 @@ * limitations under the License. */ import { act, fireEvent, render, screen } from '@testing-library/react'; +import QueryString from 'qs'; import React, { useEffect } from 'react'; import { Edge } from 'reactflow'; import { EdgeTypeEnum } from '../../components/Entity/EntityLineage/EntityLineage.interface'; @@ -162,6 +163,9 @@ describe('LineageProvider', () => { }); it('getDataQualityLineage should be called if alert is supported', async () => { + mockLocation.search = QueryString.stringify({ + layers: ['DataObservability'], + }); mockIsAlertSupported = true; (getLineageDataByFQN as jest.Mock).mockImplementationOnce(() => Promise.resolve({ @@ -190,6 +194,7 @@ describe('LineageProvider', () => { ); mockIsAlertSupported = false; + mockLocation.search = ''; }); it('should call loadChildNodesHandler', async () => { diff --git a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.tsx b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.tsx index 574f0fa6c0d8..f5ad384ee47c 100644 --- a/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/context/LineageProvider/LineageProvider.tsx @@ -147,6 +147,8 @@ const LineageProvider = ({ children }: LineageProviderProps) => { edges: [], entity: {} as EntityReference, }); + const [dataQualityLineage, setDataQualityLineage] = + useState(); const [updatedEntityLineage, setUpdatedEntityLineage] = useState(null); const [showDeleteModal, setShowDeleteModal] = useState(false); @@ -221,6 +223,25 @@ const LineageProvider = ({ children }: LineageProviderProps) => { [entityLineage] ); + const fetchDataQualityLineage = async ( + fqn: string, + config?: LineageConfig + ) => { + if (isTourOpen || !tableClassBase.getAlertEnableStatus()) { + return; + } + try { + const dqLineageResp = await getDataQualityLineage( + fqn, + config, + queryFilter + ); + setDataQualityLineage(dqLineageResp); + } catch (error) { + setDataQualityLineage(undefined); + } + }; + const fetchLineageData = useCallback( async (fqn: string, entityType: string, config?: LineageConfig) => { if (isTourOpen) { @@ -237,61 +258,25 @@ const LineageProvider = ({ children }: LineageProviderProps) => { config, queryFilter ); - - const dqLineageResp = tableClassBase.getAlertEnableStatus() - ? await getDataQualityLineage(fqn, config, queryFilter) - : { nodes: [], edges: [] }; - if (res) { - const { nodes = [], entity, edges } = res; + const { nodes = [], entity } = res; const allNodes = uniqWith( [...nodes, entity].filter(Boolean), isEqual - ).map((node) => { - return { - ...node, - isDqTestFailure: - dqLineageResp.nodes?.some((dqNode) => dqNode.id === node.id) ?? - false, - }; - }); - - const updatedEntity = { - ...entity, - isDqTestFailure: - dqLineageResp.nodes?.some((dqNode) => dqNode.id === entity.id) ?? - false, - }; - - const updatedEdges = edges?.map((edge) => { - return { - ...edge, - isDqTestFailure: - dqLineageResp.edges?.some( - (dqEdge) => dqEdge?.doc_id === edge?.doc_id - ) ?? false, - }; - }); + ); if ( entityType !== EntityType.PIPELINE && entityType !== EntityType.STORED_PROCEDURE ) { const { map: childMapObj } = getChildMap( - { - ...res, - nodes: allNodes, - edges: updatedEdges, - entity: updatedEntity, - }, + { ...res, nodes: allNodes }, decodedFqn ); setChildMap(childMapObj); const { nodes: newNodes, edges: newEdges } = getPaginatedChildMap( { ...res, - entity: updatedEntity, - edges: updatedEdges, nodes: allNodes, }, childMapObj, @@ -301,14 +286,12 @@ const LineageProvider = ({ children }: LineageProviderProps) => { setEntityLineage({ ...res, - entity: updatedEntity, nodes: newNodes, - edges: [...(updatedEdges ?? []), ...newEdges], + edges: [...(res.edges ?? []), ...newEdges], }); } else { setEntityLineage({ ...res, - entity: updatedEntity, nodes: allNodes, }); } @@ -1299,8 +1282,10 @@ const LineageProvider = ({ children }: LineageProviderProps) => { onAddPipelineClick, onUpdateLayerView, onExportClick, + dataQualityLineage, }; }, [ + dataQualityLineage, isDrawerOpen, loading, isEditMode, @@ -1363,6 +1348,12 @@ const LineageProvider = ({ children }: LineageProviderProps) => { } }, [lineageLayer]); + useEffect(() => { + if (activeLayer.includes(LineageLayer.DataObservability)) { + fetchDataQualityLineage(decodedFqn, lineageConfig); + } + }, [activeLayer, decodedFqn, lineageConfig]); + return (
({ EntityImport: jest .fn() - .mockImplementation(({ children, onImport, onCancel }) => { - return ( -
- {children}{' '} - - -
- ); - }), + .mockImplementation( + ({ children, onImport, onCancel, onCsvResultUpdate }) => { + onCsvResultUpdate(mockCsvImportResult); + + return ( +
+ {children}{' '} + + +
+ ); + } + ), }) ); jest.mock( @@ -184,17 +208,7 @@ describe('ImportTeamsPage', () => { it('TeamImportResult should visible', async () => { (importTeam as jest.Mock).mockImplementationOnce(() => - Promise.resolve({ - data: { - dryRun: true, - status: 'success', - numberOfRowsProcessed: 1, - numberOfRowsPassed: 1, - numberOfRowsFailed: 0, - importResultsCsv: - 'status,details,name*,displayName,description,teamType*,parents*,Owner,isJoinable,defaultRoles,policies\r\nsuccess,Entity updated,Applications,,,Group,Engineering,,true,,\r\n', - }, - }) + Promise.resolve(mockAsyncImportJob) ); act(() => { render(); @@ -243,18 +257,7 @@ describe('ImportTeamsPage', () => { it('UserImportResult should visible', async () => { mockLocation.search = '?type=users'; (importUserInTeam as jest.Mock).mockImplementationOnce(() => - Promise.resolve({ - data: { - dryRun: true, - status: 'success', - numberOfRowsProcessed: 1, - numberOfRowsPassed: 1, - numberOfRowsFailed: 0, - importResultsCsv: - // eslint-disable-next-line max-len - 'status,details,name*,displayName,description,email*,timezone,isAdmin,teams*,Roles\r\nsuccess,Entity updated,aaron_johnson0,Aaron Johnson,,aaron_johnson0@gmail.com,,false,Applications,DataSteward\r\n', - }, - }) + Promise.resolve(mockAsyncImportJob1) ); act(() => { render(); diff --git a/openmetadata-ui/src/main/resources/ui/src/pages/TeamsPage/ImportTeamsPage/ImportTeamsPage.tsx b/openmetadata-ui/src/main/resources/ui/src/pages/TeamsPage/ImportTeamsPage/ImportTeamsPage.tsx index 2507179a5c2c..58b20d044ce6 100644 --- a/openmetadata-ui/src/main/resources/ui/src/pages/TeamsPage/ImportTeamsPage/ImportTeamsPage.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/pages/TeamsPage/ImportTeamsPage/ImportTeamsPage.tsx @@ -91,6 +91,10 @@ const ImportTeamsPage = () => { return ; }, [csvImportResult, type]); + const handleCsvImportResultUpdate = (result: CSVImportResult) => { + setCsvImportResult(result); + }; + const fetchPermissions = async (entityFqn: string) => { setIsPageLoading(true); try { @@ -131,7 +135,6 @@ const ImportTeamsPage = () => { const api = type === ImportType.USERS ? importUserInTeam : importTeam; try { const response = await api(name, data, dryRun); - setCsvImportResult(response); return response; } catch (error) { @@ -208,6 +211,7 @@ const ImportTeamsPage = () => { {importResult} diff --git a/openmetadata-ui/src/main/resources/ui/src/rest/glossaryAPI.ts b/openmetadata-ui/src/main/resources/ui/src/rest/glossaryAPI.ts index 2828f847fd7a..72d817f816e0 100644 --- a/openmetadata-ui/src/main/resources/ui/src/rest/glossaryAPI.ts +++ b/openmetadata-ui/src/main/resources/ui/src/rest/glossaryAPI.ts @@ -14,6 +14,7 @@ import { AxiosResponse } from 'axios'; import { Operation } from 'fast-json-patch'; import { PagingResponse } from 'Models'; +import { CSVImportAsyncResponse } from '../components/BulkImport/BulkEntityImport.interface'; import { CSVExportResponse } from '../components/Entity/EntityExportModalProvider/EntityExportModalProvider.interface'; import { VotingDataProps } from '../components/Entity/Voting/voting.interface'; import { ES_MAX_PAGE_SIZE, PAGE_SIZE_MEDIUM } from '../constants/constants'; @@ -26,7 +27,6 @@ import { EntityReference, Glossary } from '../generated/entity/data/glossary'; import { GlossaryTerm } from '../generated/entity/data/glossaryTerm'; import { BulkOperationResult } from '../generated/type/bulkOperationResult'; import { ChangeEvent } from '../generated/type/changeEvent'; -import { CSVImportResult } from '../generated/type/csvImportResult'; import { EntityHistory } from '../generated/type/entityHistory'; import { ListParams } from '../interface/API.interface'; import { getEncodedFqn } from '../utils/StringsUtils'; @@ -188,8 +188,13 @@ export const importGlossaryInCSVFormat = async ( const configOptions = { headers: { 'Content-type': 'text/plain' }, }; - const response = await APIClient.put>( - `/glossaries/name/${getEncodedFqn(glossaryName)}/import?dryRun=${dryRun}`, + const response = await APIClient.put< + string, + AxiosResponse + >( + `/glossaries/name/${getEncodedFqn( + glossaryName + )}/importAsync?dryRun=${dryRun}`, data, configOptions ); diff --git a/openmetadata-ui/src/main/resources/ui/src/rest/teamsAPI.ts b/openmetadata-ui/src/main/resources/ui/src/rest/teamsAPI.ts index 9dd1dc9f38df..8c587d7a3814 100644 --- a/openmetadata-ui/src/main/resources/ui/src/rest/teamsAPI.ts +++ b/openmetadata-ui/src/main/resources/ui/src/rest/teamsAPI.ts @@ -14,11 +14,11 @@ import { AxiosResponse } from 'axios'; import { Operation } from 'fast-json-patch'; import { PagingResponse, RestoreRequestType } from 'Models'; +import { CSVImportAsyncResponse } from '../components/BulkImport/BulkEntityImport.interface'; import { CSVExportResponse } from '../components/Entity/EntityExportModalProvider/EntityExportModalProvider.interface'; import { CreateTeam } from '../generated/api/teams/createTeam'; import { Team } from '../generated/entity/teams/team'; import { TeamHierarchy } from '../generated/entity/teams/teamHierarchy'; -import { CSVImportResult } from '../generated/type/csvImportResult'; import { ListParams } from '../interface/API.interface'; import { getEncodedFqn } from '../utils/StringsUtils'; import APIClient from './index'; @@ -120,11 +120,10 @@ export const importTeam = async ( dryRun, }, }; - const response = await APIClient.put>( - `/teams/name/${getEncodedFqn(teamName)}/import`, - data, - configOptions - ); + const response = await APIClient.put< + string, + AxiosResponse + >(`/teams/name/${getEncodedFqn(teamName)}/importAsync`, data, configOptions); return response.data; }; @@ -141,11 +140,10 @@ export const importUserInTeam = async ( dryRun, }, }; - const response = await APIClient.put>( - `/users/import`, - data, - configOptions - ); + const response = await APIClient.put< + string, + AxiosResponse + >(`/users/importAsync`, data, configOptions); return response.data; }; diff --git a/openmetadata-ui/src/main/resources/ui/src/styles/components/code-mirror.less b/openmetadata-ui/src/main/resources/ui/src/styles/components/code-mirror.less index 28a49d4eeb28..24cfc45d5df4 100644 --- a/openmetadata-ui/src/main/resources/ui/src/styles/components/code-mirror.less +++ b/openmetadata-ui/src/main/resources/ui/src/styles/components/code-mirror.less @@ -53,4 +53,7 @@ .CodeMirror-linenumber { color: @text-color; } + .CodeMirror-sizer { + margin-left: 40px !important; + } } diff --git a/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.test.ts b/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.test.ts index 030ccfa735f5..c6cde50a0c1d 100644 --- a/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.test.ts +++ b/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.test.ts @@ -22,6 +22,7 @@ import { } from '../generated/type/tagLabel'; import { digitFormatter, + filterSelectOptions, getBase64EncodedString, getIsErrorMatch, getNameFromFQN, @@ -294,4 +295,55 @@ describe('Tests for CommonUtils', () => { expect(isDeleted(null)).toBe(false); }); }); + + describe('filterSelectOptions', () => { + it('should return true if input matches option labelValue', () => { + const input = 'test'; + const option = { + labelValue: 'Test Label', + value: 'testValue', + label: 'Test Label', + }; + + expect(filterSelectOptions(input, option)).toBe(true); + }); + + it('should return true if input matches option value', () => { + const input = 'test'; + const option = { + labelValue: 'Label', + label: 'Label', + value: 'testValue', + }; + + expect(filterSelectOptions(input, option)).toBe(true); + }); + + it('should return false if input does not match option labelValue or value', () => { + const input = 'test'; + const option = { labelValue: 'Label', value: 'value', label: 'Label' }; + + expect(filterSelectOptions(input, option)).toBe(false); + }); + + it('should return false if option is undefined', () => { + const input = 'test'; + + expect(filterSelectOptions(input)).toBe(false); + }); + + it('should handle non-string option value gracefully', () => { + const input = 'test'; + const option = { labelValue: 'Label', value: 123, label: 'Label' }; + + expect(filterSelectOptions(input, option)).toBe(false); + }); + + it('should handle empty input gracefully', () => { + const input = ''; + const option = { labelValue: 'Label', value: 'value', label: 'Label' }; + + expect(filterSelectOptions(input, option)).toBe(true); + }); + }); }); diff --git a/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.tsx b/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.tsx index 47438230fba7..e00278caca2d 100644 --- a/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.tsx +++ b/openmetadata-ui/src/main/resources/ui/src/utils/CommonUtils.tsx @@ -13,6 +13,7 @@ /* eslint-disable @typescript-eslint/ban-types */ +import { DefaultOptionType } from 'antd/lib/select'; import { AxiosError } from 'axios'; import classNames from 'classnames'; import { t } from 'i18next'; @@ -845,11 +846,13 @@ export const getServiceTypeExploreQueryFilter = (serviceType: string) => { export const filterSelectOptions = ( input: string, - option?: { label: string; value: string } + option?: DefaultOptionType ) => { return ( - toLower(option?.label).includes(toLower(input)) || - toLower(option?.value).includes(toLower(input)) + toLower(option?.labelValue).includes(toLower(input)) || + toLower(isString(option?.value) ? option?.value : '').includes( + toLower(input) + ) ); }; diff --git a/openmetadata-ui/src/main/resources/ui/src/utils/DataQuality/DataQualityUtils.ts b/openmetadata-ui/src/main/resources/ui/src/utils/DataQuality/DataQualityUtils.ts index 7c24847b592f..f9959fc49d71 100644 --- a/openmetadata-ui/src/main/resources/ui/src/utils/DataQuality/DataQualityUtils.ts +++ b/openmetadata-ui/src/main/resources/ui/src/utils/DataQuality/DataQualityUtils.ts @@ -11,7 +11,6 @@ * limitations under the License. */ import { cloneDeep, isArray, isUndefined, omit, omitBy } from 'lodash'; -import { ReactComponent as TestCaseIcon } from '../../assets/svg/all-activity-v2.svg'; import { ReactComponent as AccuracyIcon } from '../../assets/svg/ic-accuracy.svg'; import { ReactComponent as CompletenessIcon } from '../../assets/svg/ic-completeness.svg'; import { ReactComponent as ConsistencyIcon } from '../../assets/svg/ic-consistency.svg'; @@ -19,6 +18,7 @@ import { ReactComponent as IntegrityIcon } from '../../assets/svg/ic-integrity.s import { ReactComponent as SqlIcon } from '../../assets/svg/ic-sql.svg'; import { ReactComponent as UniquenessIcon } from '../../assets/svg/ic-uniqueness.svg'; import { ReactComponent as ValidityIcon } from '../../assets/svg/ic-validity.svg'; +import { ReactComponent as NoDimensionIcon } from '../../assets/svg/no-dimension-icon.svg'; import { StatusData } from '../../components/DataQuality/ChartWidgets/StatusCardWidget/StatusCardWidget.interface'; import { TestCaseSearchParams } from '../../components/DataQuality/DataQuality.interface'; import { @@ -235,6 +235,6 @@ export const getDimensionIcon = (dimension: DataQualityDimensions) => { case DataQualityDimensions.Validity: return ValidityIcon; default: - return TestCaseIcon; + return NoDimensionIcon; } };