diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ColumnRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ColumnRepository.java index e7eabe5cf1b..6750d33fdbe 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ColumnRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ColumnRepository.java @@ -15,6 +15,8 @@ package org.openmetadata.service.jdbi3; import static org.openmetadata.service.Entity.DASHBOARD_DATA_MODEL; import static org.openmetadata.service.Entity.TABLE; +import static org.openmetadata.service.events.ChangeEventHandler.copyChangeEvent; +import static org.openmetadata.service.formatter.util.FormatterUtil.createChangeEventForEntity; import static org.openmetadata.service.resources.tags.TagLabelUtil.addDerivedTags; import jakarta.json.JsonPatch; @@ -22,11 +24,14 @@ import jakarta.ws.rs.core.SecurityContext; import jakarta.ws.rs.core.UriInfo; import java.util.List; import lombok.extern.slf4j.Slf4j; +import org.openmetadata.schema.EntityInterface; import org.openmetadata.schema.api.data.UpdateColumn; import org.openmetadata.schema.entity.data.DashboardDataModel; import org.openmetadata.schema.entity.data.Table; +import org.openmetadata.schema.type.ChangeEvent; import org.openmetadata.schema.type.Column; import org.openmetadata.schema.type.EntityReference; +import org.openmetadata.schema.type.EventType; import org.openmetadata.schema.type.Include; import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; @@ -36,6 +41,7 @@ import org.openmetadata.service.security.policyevaluator.OperationContext; import org.openmetadata.service.security.policyevaluator.ResourceContext; import org.openmetadata.service.security.policyevaluator.ResourceContextInterface; import org.openmetadata.service.util.FullyQualifiedName; +import org.openmetadata.service.util.RestUtil; @Slf4j public class ColumnRepository { @@ -144,7 +150,9 @@ public class ColumnRepository { TABLE, parentEntityRef.getId(), null, ResourceContextInterface.Operation.PATCH); authorizer.authorize(securityContext, operationContext, resourceContext); - tableRepository.patch(uriInfo, parentEntityRef.getId(), user, jsonPatch); + RestUtil.PatchResponse patchResponse = + tableRepository.patch(uriInfo, parentEntityRef.getId(), user, jsonPatch); + triggerParentChangeEvent(patchResponse.entity(), user); return column; } @@ -211,7 +219,9 @@ public class ColumnRepository { ResourceContextInterface.Operation.PATCH); authorizer.authorize(securityContext, operationContext, resourceContext); - dataModelRepository.patch(uriInfo, parentEntityRef.getId(), user, jsonPatch); + RestUtil.PatchResponse patchResponse = + dataModelRepository.patch(uriInfo, parentEntityRef.getId(), user, jsonPatch); + triggerParentChangeEvent(patchResponse.entity(), user); return column; } @@ -263,4 +273,13 @@ public class ColumnRepository { } return null; } + + private void triggerParentChangeEvent(Object parent, String user) { + ChangeEvent changeEvent = + createChangeEventForEntity(user, EventType.ENTITY_UPDATED, (EntityInterface) parent); + Object entity = changeEvent.getEntity(); + changeEvent = copyChangeEvent(changeEvent); + changeEvent.setEntity(JsonUtils.pojoToMaskedJson(entity)); + Entity.getCollectionDAO().changeEventDAO().insert(JsonUtils.pojoToJson(changeEvent)); + } } diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/columns/ColumnResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/columns/ColumnResourceTest.java index ea9a231524f..cbc14236286 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/columns/ColumnResourceTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/columns/ColumnResourceTest.java @@ -15,6 +15,7 @@ package org.openmetadata.service.resources.columns; import static jakarta.ws.rs.core.Response.Status.NOT_FOUND; import static jakarta.ws.rs.core.Response.Status.OK; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -28,11 +29,16 @@ import jakarta.ws.rs.client.WebTarget; import java.io.IOException; import java.util.Arrays; import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.junit.jupiter.api.TestInstance; +import org.openmetadata.schema.EntityInterface; import org.openmetadata.schema.api.classification.CreateClassification; import org.openmetadata.schema.api.classification.CreateTag; import org.openmetadata.schema.api.data.CreateDashboardDataModel; @@ -52,6 +58,7 @@ import org.openmetadata.schema.entity.data.DatabaseSchema; import org.openmetadata.schema.entity.data.Glossary; import org.openmetadata.schema.entity.data.GlossaryTerm; import org.openmetadata.schema.entity.data.Table; +import org.openmetadata.schema.entity.feed.Thread; import org.openmetadata.schema.entity.services.DashboardService; import org.openmetadata.schema.entity.services.DatabaseService; import org.openmetadata.schema.type.Column; @@ -59,11 +66,13 @@ import org.openmetadata.schema.type.ColumnConstraint; import org.openmetadata.schema.type.ColumnDataType; import org.openmetadata.schema.type.DataModelType; import org.openmetadata.schema.type.TagLabel; +import org.openmetadata.service.Entity; import org.openmetadata.service.OpenMetadataApplicationTest; import org.openmetadata.service.resources.databases.DatabaseResourceTest; import org.openmetadata.service.resources.databases.DatabaseSchemaResourceTest; import org.openmetadata.service.resources.databases.TableResourceTest; import org.openmetadata.service.resources.datamodels.DashboardDataModelResourceTest; +import org.openmetadata.service.resources.feeds.FeedResourceTest; import org.openmetadata.service.resources.glossary.GlossaryResourceTest; import org.openmetadata.service.resources.glossary.GlossaryTermResourceTest; import org.openmetadata.service.resources.services.DashboardServiceResourceTest; @@ -71,6 +80,7 @@ import org.openmetadata.service.resources.services.DatabaseServiceResourceTest; import org.openmetadata.service.resources.tags.ClassificationResourceTest; import org.openmetadata.service.resources.tags.TagResourceTest; import org.openmetadata.service.util.FullyQualifiedName; +import org.openmetadata.service.util.ResultList; import org.openmetadata.service.util.TestUtils; @Slf4j @@ -1150,6 +1160,151 @@ class ColumnResourceTest extends OpenMetadataApplicationTest { assertTrue(hasDerivedTag, "Derived tag should be present in persisted data"); } + @Test + void test_tableColumnChangeEvents() throws IOException { + // Create a new table specifically for this test to avoid feed pollution + TableResourceTest isolatedTableTest = new TableResourceTest(); + String testName = "tableColumnFeedTest_" + UUID.randomUUID(); + + DatabaseSchema schema = + new DatabaseSchemaResourceTest() + .createEntity( + new CreateDatabaseSchema() + .withName(testName) + .withDatabase(table.getDatabase().getFullyQualifiedName()), + ADMIN_AUTH_HEADERS); + + List columns = + Arrays.asList( + new Column().withName("id").withDataType(ColumnDataType.INT), + new Column() + .withName("description_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255), + new Column() + .withName("displayname_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255), + new Column() + .withName("tags_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255)); + + CreateTable createIsolatedTable = + new CreateTable() + .withName(testName) + .withDatabaseSchema(schema.getFullyQualifiedName()) + .withColumns(columns); + + Table isolatedTable = isolatedTableTest.createEntity(createIsolatedTable, ADMIN_AUTH_HEADERS); + + // 1. Test description change in feed + String descriptionColFQN = isolatedTable.getFullyQualifiedName() + ".description_col"; + UpdateColumn updateDescription = + new UpdateColumn().withDescription("Test description for feed verification"); + Column updatedDescriptionCol = updateColumnByFQN(descriptionColFQN, updateDescription); + assertEquals("Test description for feed verification", updatedDescriptionCol.getDescription()); + + // Verify description change appears in activity feed + verifyColumnChangeEventInFeed( + isolatedTable, "description_col", List.of("description"), Entity.TABLE); + + // 2. Test tags/terms change in feed + String tagsColFQN = isolatedTable.getFullyQualifiedName() + ".tags_col"; + + // Create tag label for classification tag + TagLabel classificationTag = + new TagLabel() + .withTagFQN(personalDataTag.getFullyQualifiedName()) + .withSource(TagLabel.TagSource.CLASSIFICATION); + + // Create tag label for glossary term + TagLabel glossaryTerm = + new TagLabel() + .withTagFQN(businessTermsGlossaryTerm.getFullyQualifiedName()) + .withSource(TagLabel.TagSource.GLOSSARY); + + // Update column with tags + UpdateColumn updateTags = new UpdateColumn().withTags(List.of(classificationTag, glossaryTerm)); + Column updatedTagsCol = updateColumnByFQN(tagsColFQN, updateTags); + + // Verify tags were added + assertEquals(2, updatedTagsCol.getTags().size()); + + // Verify tags change appears in activity feed + verifyColumnChangeEventInFeed(isolatedTable, "tags_col", List.of("tags"), Entity.TABLE); + } + + @Test + void test_dashboardDataModelColumnChangeEvents() throws IOException { + // Create a new dashboard data model specifically for this test to avoid feed pollution + DashboardDataModelResourceTest isolatedModelTest = new DashboardDataModelResourceTest(); + String testName = "dataModelColumnFeedTest_" + UUID.randomUUID(); + + // Create test data model with simple columns + List columns = + Arrays.asList( + new Column().withName("metric_col").withDataType(ColumnDataType.NUMERIC), + new Column() + .withName("description_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255), + new Column() + .withName("displayname_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255), + new Column() + .withName("tags_col") + .withDataType(ColumnDataType.VARCHAR) + .withDataLength(255)); + + CreateDashboardDataModel createIsolatedModel = + new CreateDashboardDataModel() + .withName(testName) + .withService(dashboardDataModel.getService().getFullyQualifiedName()) + .withDataModelType(DataModelType.MetabaseDataModel) + .withColumns(columns); + + DashboardDataModel isolatedModel = + isolatedModelTest.createEntity(createIsolatedModel, ADMIN_AUTH_HEADERS); + + // 1. Test description change in feed + String descriptionColFQN = isolatedModel.getFullyQualifiedName() + ".description_col"; + UpdateColumn updateDescription = + new UpdateColumn().withDescription("Test data model description for feed verification"); + Column updatedDescriptionCol = + updateColumnByFQN(descriptionColFQN, updateDescription, "dashboardDataModel"); + assertEquals( + "Test data model description for feed verification", + updatedDescriptionCol.getDescription()); + + // 2. Test tags/terms change in feed + String tagsColFQN = isolatedModel.getFullyQualifiedName() + ".tags_col"; + + // Create tag label for classification tag + TagLabel classificationTag = + new TagLabel() + .withTagFQN(businessMetricsTag.getFullyQualifiedName()) + .withSource(TagLabel.TagSource.CLASSIFICATION); + + // Create tag label for glossary term + TagLabel glossaryTerm = + new TagLabel() + .withTagFQN(technicalTermsGlossaryTerm.getFullyQualifiedName()) + .withSource(TagLabel.TagSource.GLOSSARY); + + // Update column with tags + UpdateColumn updateTags = new UpdateColumn().withTags(List.of(classificationTag, glossaryTerm)); + Column updatedTagsCol = updateColumnByFQN(tagsColFQN, updateTags, "dashboardDataModel"); + + // Verify tags were added + assertEquals(2, updatedTagsCol.getTags().size()); + + // Verify tags change appears in activity feed + verifyColumnChangeEventInFeed( + isolatedModel, "tags_col", List.of("tags"), Entity.DASHBOARD_DATA_MODEL); + } + private Column updateColumnByFQN(String columnFQN, UpdateColumn updateColumn, String entityType) throws IOException { WebTarget target = @@ -1160,4 +1315,75 @@ class ColumnResourceTest extends OpenMetadataApplicationTest { private Column updateColumnByFQN(String columnFQN, UpdateColumn updateColumn) throws IOException { return updateColumnByFQN(columnFQN, updateColumn, "table"); } + + private void verifyColumnChangeEventInFeed( + EntityInterface parentEntity, + String columnName, + List expectedFields, + String entityType) + throws IOException { + + String entityLink = + String.format("<#E::%s::%s>", entityType, parentEntity.getFullyQualifiedName()); + + FeedResourceTest feedResourceTest = new FeedResourceTest(); + + AtomicReference> threadsRef = new AtomicReference<>(); + + await() + .pollInterval(2, TimeUnit.SECONDS) + .atMost(90, TimeUnit.SECONDS) + .until( + () -> { + // Poll for threads related to this entity + ResultList fetchedThreads = + feedResourceTest.listThreads(entityLink, null, ADMIN_AUTH_HEADERS); + + if (fetchedThreads == null || fetchedThreads.getData().isEmpty()) { + return false; + } + + // Check if any thread mentions our column + boolean found = + fetchedThreads.getData().stream() + .anyMatch( + thread -> + thread.getMessage() != null + && thread.getMessage().contains(columnName)); + + if (found) { + // Store the result only when we find a match + threadsRef.set(fetchedThreads); + } + + return found; + }); + + ResultList threads = threadsRef.get(); + + // Verify feed contains our column update + assertNotNull(threads); + assertFalse(threads.getData().isEmpty()); + + Optional columnUpdateThread = + threads.getData().stream() + .filter(thread -> thread.getMessage().contains(columnName)) + .findFirst(); + + assertTrue( + columnUpdateThread.isPresent(), + "No activity feed entry found for column update: " + columnName); + + Thread thread = columnUpdateThread.get(); + + // Verify the thread message mentions the expected fields + boolean foundFieldReferences = + expectedFields.stream() + .allMatch(field -> thread.getMessage().toLowerCase().contains(field.toLowerCase())); + + assertTrue( + foundFieldReferences, + "Activity feed doesn't contain references to all expected fields: " + + String.join(", ", expectedFields)); + } }