Fixes #30908: Add Conversation V2 and activity replies

.github/playwright/impact-map.json

@@ -344,7 +344,7 @@
         "openmetadata-ui/src/main/resources/ui/src/components/Auth/**",
         "openmetadata-ui/src/main/resources/ui/src/pages/Login/**"
       ],
-      "projects": ["chromium", "Basic", "SearchRBAC", "ImportExport"],
+      "projects": ["chromium", "Basic", "Ingestion", "SearchRBAC", "ImportExport"],
       "specs": [
         "playwright/e2e/**/*Permission*.spec.ts",
         "playwright/e2e/Pages/Login*.spec.ts",

.github/scripts/select_playwright_tests.py

@@ -59,6 +59,14 @@ def is_mapped_file(path: str, impact_map: dict[str, Any]) -> bool:
     )
 
 
+def remove_delegated_specs(
+    selected: dict[str, set[str]], delegated_patterns: list[str]
+) -> None:
+    for spec in list(selected):
+        if matches(spec, delegated_patterns):
+            del selected[spec]
+
+
 def write_github_output(path: Path, plan: dict[str, Any]) -> None:
     direct_changed_specs = plan.get("directChangedSpecs", [])
     lineage_representative_only = (
@@ -144,12 +152,7 @@ def main() -> None:
             for entry in impact_map["canary"]:
                 add_selection(selected, entry, repo_root)
 
-        delegated_patterns = impact_map.get("delegatedSpecs", [])
-        selected = {
-            spec: projects
-            for spec, projects in selected.items()
-            if not matches(spec, delegated_patterns)
-        }
+        remove_delegated_specs(selected, impact_map.get("delegatedSpecs", []))
 
         plan = {
             "version": 1,

.github/scripts/tests/test_playwright_ci_planning.py

@@ -2375,6 +2375,75 @@ def test_changed_visual_regression_spec_is_delegated_not_selected(tmp_path, monk
     assert selection["unmappedFiles"] == []
 
 
+def test_impact_mapping_cannot_reselect_a_delegated_spec(tmp_path, monkeypatch):
+    selector = load_script("select_playwright_tests")
+    spec_dir = tmp_path / selector.UI_ROOT / "playwright/e2e/Features"
+    spec_dir.mkdir(parents=True)
+    spec_path = spec_dir / "Delegated.spec.ts"
+    spec_path.write_text("test('delegated', () => undefined);\n")
+    impact_map = tmp_path / "impact-map.json"
+    impact_map.write_text(
+        json.dumps(
+            {
+                "smoke": [],
+                "canary": [],
+                "delegatedSpecs": ["playwright/e2e/Features/Delegated.spec.ts"],
+                "sharedInfrastructure": [],
+                "mappings": [
+                    {
+                        "sources": ["src/rdf/**"],
+                        "projects": ["chromium"],
+                        "specs": ["playwright/e2e/Features/Delegated.spec.ts"],
+                    }
+                ],
+            }
+        )
+    )
+    changed = tmp_path / "changed.txt"
+    changed.write_text("src/rdf/Processor.java\n")
+    output = tmp_path / "selection.json"
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("GITHUB_OUTPUT", raising=False)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "select_playwright_tests.py",
+            "--event-name",
+            "pull_request_target",
+            "--changed-files",
+            str(changed),
+            "--impact-map",
+            str(impact_map),
+            "--output",
+            str(output),
+        ],
+    )
+
+    selector.main()
+
+    selection = json.loads(output.read_text())
+    assert selection["selectors"] == []
+
+
+def test_security_impact_mapping_includes_ingestion_permission_specs():
+    selector = load_script("select_playwright_tests")
+    impact_map = json.loads(
+        (SCRIPTS.parents[0] / "playwright/impact-map.json").read_text()
+    )
+    mapping = next(
+        entry
+        for entry in impact_map["mappings"]
+        if "openmetadata-service/src/main/java/org/openmetadata/service/security/**"
+        in entry["sources"]
+    )
+
+    assert "Ingestion" in mapping["projects"]
+    assert selector.matches(
+        "playwright/e2e/Flow/ServiceCreationPermissions.spec.ts", mapping["specs"]
+    )
+
+
 def test_summary_reconciles_results_and_evaluates_performance_independently():
     # The playwright-summary job lives in the postgres PR caller (not the
     # reusable) so branch protection can require its unprefixed check name.

bootstrap/sql/migrations/native/2.1.0/mysql/postDataMigrationSQLScript.sql

@@ -26,3 +26,18 @@ ON DUPLICATE KEY UPDATE
   test_case_incident.createdAt = LEAST(test_case_incident.createdAt, VALUES(createdAt)),
   updatedAt = VALUES(updatedAt),
   latestRecordId = VALUES(latestRecordId);
+
+-- Activity comments are retained indefinitely unless an administrator explicitly configures a
+-- positive retention period. Preserve any value already chosen by an administrator.
+UPDATE installed_apps
+SET json = JSON_INSERT(json, '$.appConfiguration.activityCommentsRetentionPeriod', 0)
+WHERE name = 'DataRetentionApplication';
+
+UPDATE apps_marketplace
+SET json = JSON_INSERT(json, '$.appConfiguration.activityCommentsRetentionPeriod', 0)
+WHERE name = 'DataRetentionApplication';
+
+UPDATE entity_extension
+SET json = JSON_INSERT(json, '$.appConfiguration.activityCommentsRetentionPeriod', 0)
+WHERE extension LIKE 'app.version.%'
+  AND json->>'$.name' = 'DataRetentionApplication';

bootstrap/sql/migrations/native/2.1.0/mysql/schemaChanges.sql

@@ -36,6 +36,104 @@ CREATE TABLE IF NOT EXISTS test_case_incident (
     INDEX idx_tci_updated (updatedAt)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
 
+-- Conversation V2 stores bounded roots and replies as schema-first JSON. Indexed mentions and
+-- domains remain normalized because they participate in filters and authorization.
+CREATE TABLE IF NOT EXISTS conversation_entity (
+    id varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.id'))) STORED NOT NULL,
+    source varchar(16) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.source'))) STORED NOT NULL,
+    entityType varchar(64) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.entityRef.type'))) STORED NOT NULL,
+    entityId varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.entityRef.id'))) STORED NOT NULL,
+    entityFqnHash varchar(768) CHARACTER SET ascii COLLATE ascii_bin,
+    about varchar(2048) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.about'))) STORED NOT NULL,
+    aboutFqnHash varchar(768) CHARACTER SET ascii COLLATE ascii_bin,
+    activityEventId varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.activityEventId'))) STORED,
+    creatorId varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.createdBy.id'))) STORED,
+    createdAt bigint GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.createdAt'))) STORED NOT NULL,
+    updatedAt bigint GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.updatedAt'))) STORED NOT NULL,
+    resolved tinyint(1) GENERATED ALWAYS AS
+      (json_extract(json, _utf8mb4'$.resolved')) STORED NOT NULL,
+    replyCount int GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.replyCount'))) STORED NOT NULL,
+    json json NOT NULL,
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_conversation_activity_event (activityEventId),
+    KEY idx_conversation_entity (entityType, entityId, updatedAt, id),
+    KEY idx_conversation_entity_fqn (entityFqnHash, updatedAt, id),
+    KEY idx_conversation_about (aboutFqnHash, updatedAt, id),
+    KEY idx_conversation_creator (creatorId, updatedAt, id),
+    KEY idx_conversation_source_updated (source, updatedAt, id),
+    KEY idx_conversation_created (createdAt, id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
+
+CREATE TABLE IF NOT EXISTS conversation_reply (
+    id varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.id'))) STORED NOT NULL,
+    conversationId varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.conversationId'))) STORED NOT NULL,
+    authorId varchar(36) GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.author.id'))) STORED NOT NULL,
+    createdAt bigint GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.createdAt'))) STORED NOT NULL,
+    updatedAt bigint GENERATED ALWAYS AS
+      (json_unquote(json_extract(json, _utf8mb4'$.updatedAt'))) STORED NOT NULL,
+    json json NOT NULL,
+    PRIMARY KEY (id),
+    CONSTRAINT fk_conversation_reply_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE,
+    KEY idx_conversation_reply_cursor (conversationId, createdAt, id),
+    KEY idx_conversation_reply_author (authorId, createdAt, id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
+
+CREATE TABLE IF NOT EXISTS conversation_mention (
+    conversationId varchar(36) NOT NULL,
+    targetType varchar(16) NOT NULL,
+    targetId varchar(36) NOT NULL,
+    mentionedEntityType varchar(64) NOT NULL,
+    mentionedEntityId varchar(36) NOT NULL,
+    createdAt bigint NOT NULL,
+    PRIMARY KEY (targetType, targetId, mentionedEntityType, mentionedEntityId),
+    CONSTRAINT fk_conversation_mention_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE,
+    KEY idx_conversation_mention_lookup
+      (mentionedEntityType, mentionedEntityId, createdAt, conversationId),
+    KEY idx_conversation_mention_conversation (conversationId, targetType, targetId)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
+
+CREATE TABLE IF NOT EXISTS conversation_domain (
+    conversationId varchar(36) NOT NULL,
+    domainId varchar(36) NOT NULL,
+    PRIMARY KEY (conversationId, domainId),
+    CONSTRAINT fk_conversation_domain_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE,
+    KEY idx_conversation_domain_lookup (domainId, conversationId)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
+
+SET @drop_conversation_activity_timestamp_ddl = (
+  SELECT IF(
+    EXISTS (
+      SELECT 1
+      FROM information_schema.columns
+      WHERE table_schema = DATABASE()
+        AND table_name = 'conversation_entity'
+        AND column_name = 'activityTimestamp'
+    ),
+    'ALTER TABLE conversation_entity DROP COLUMN activityTimestamp',
+    'SELECT 1'
+  )
+);
+PREPARE drop_conversation_activity_timestamp_stmt
+  FROM @drop_conversation_activity_timestamp_ddl;
+EXECUTE drop_conversation_activity_timestamp_stmt;
+DEALLOCATE PREPARE drop_conversation_activity_timestamp_stmt;
 -- Pipeline-backed lineage is the only relationship lookup whose selective identifier lives in JSON.
 -- Pairing it with relation serves every pipeline lineage path without widening the generic table schema.
 CREATE INDEX idx_entity_relationship_pipeline_relation

bootstrap/sql/migrations/native/2.1.0/postgres/postDataMigrationSQLScript.sql

@@ -26,3 +26,24 @@ ON CONFLICT (stateId) DO UPDATE SET
   createdAt = LEAST(test_case_incident.createdAt, EXCLUDED.createdAt),
   updatedAt = EXCLUDED.updatedAt,
   latestRecordId = EXCLUDED.latestRecordId;
+
+-- Activity comments are retained indefinitely unless an administrator explicitly configures a
+-- positive retention period. Preserve any value already chosen by an administrator.
+UPDATE installed_apps
+SET json = jsonb_set(
+    json::jsonb, '{appConfiguration,activityCommentsRetentionPeriod}', '0'::jsonb, true)
+WHERE name = 'DataRetentionApplication'
+  AND NOT jsonb_exists(json::jsonb #> '{appConfiguration}', 'activityCommentsRetentionPeriod');
+
+UPDATE apps_marketplace
+SET json = jsonb_set(
+    json::jsonb, '{appConfiguration,activityCommentsRetentionPeriod}', '0'::jsonb, true)
+WHERE name = 'DataRetentionApplication'
+  AND NOT jsonb_exists(json::jsonb #> '{appConfiguration}', 'activityCommentsRetentionPeriod');
+
+UPDATE entity_extension
+SET json = jsonb_set(
+    json::jsonb, '{appConfiguration,activityCommentsRetentionPeriod}', '0'::jsonb, true)
+WHERE extension LIKE 'app.version.%'
+  AND json::jsonb ->> 'name' = 'DataRetentionApplication'
+  AND NOT jsonb_exists(json::jsonb #> '{appConfiguration}', 'activityCommentsRetentionPeriod');

bootstrap/sql/migrations/native/2.1.0/postgres/schemaChanges.sql

@@ -36,6 +36,98 @@ CREATE INDEX IF NOT EXISTS idx_tci_fqn ON test_case_incident (entityFQNHash);
 CREATE INDEX IF NOT EXISTS idx_tci_assignee ON test_case_incident (assignee, testCaseResolutionStatusType);
 CREATE INDEX IF NOT EXISTS idx_tci_updated ON test_case_incident (updatedAt);
 
+-- Conversation V2 stores bounded roots and replies as schema-first JSON. Indexed mentions and
+-- domains remain normalized because they participate in filters and authorization.
+CREATE TABLE IF NOT EXISTS conversation_entity (
+    id character varying(36) GENERATED ALWAYS AS ((json ->> 'id'::text)) STORED NOT NULL,
+    source character varying(16) GENERATED ALWAYS AS ((json ->> 'source'::text)) STORED NOT NULL,
+    entityType character varying(64) GENERATED ALWAYS AS
+      ((json #>> '{entityRef,type}'::text[])) STORED NOT NULL,
+    entityId character varying(36) GENERATED ALWAYS AS
+      ((json #>> '{entityRef,id}'::text[])) STORED NOT NULL,
+    entityFqnHash character varying(768),
+    about character varying(2048) GENERATED ALWAYS AS ((json ->> 'about'::text)) STORED NOT NULL,
+    aboutFqnHash character varying(768),
+    activityEventId character varying(36) GENERATED ALWAYS AS
+      ((json ->> 'activityEventId'::text)) STORED,
+    creatorId character varying(36) GENERATED ALWAYS AS
+      ((json #>> '{createdBy,id}'::text[])) STORED,
+    createdAt bigint GENERATED ALWAYS AS
+      (((json ->> 'createdAt'::text))::bigint) STORED NOT NULL,
+    updatedAt bigint GENERATED ALWAYS AS
+      (((json ->> 'updatedAt'::text))::bigint) STORED NOT NULL,
+    resolved boolean GENERATED ALWAYS AS
+      (((json ->> 'resolved'::text))::boolean) STORED NOT NULL,
+    replyCount integer GENERATED ALWAYS AS
+      (((json ->> 'replyCount'::text))::integer) STORED NOT NULL,
+    json jsonb NOT NULL,
+    PRIMARY KEY (id),
+    CONSTRAINT uk_conversation_activity_event UNIQUE (activityEventId)
+);
+
+CREATE INDEX IF NOT EXISTS idx_conversation_entity
+    ON conversation_entity (entityType, entityId, updatedAt DESC, id DESC);
+CREATE INDEX IF NOT EXISTS idx_conversation_entity_fqn
+    ON conversation_entity (entityFqnHash, updatedAt DESC, id DESC);
+CREATE INDEX IF NOT EXISTS idx_conversation_about
+    ON conversation_entity (aboutFqnHash, updatedAt DESC, id DESC);
+CREATE INDEX IF NOT EXISTS idx_conversation_creator
+    ON conversation_entity (creatorId, updatedAt DESC, id DESC);
+CREATE INDEX IF NOT EXISTS idx_conversation_source_updated
+    ON conversation_entity (source, updatedAt DESC, id DESC);
+CREATE INDEX IF NOT EXISTS idx_conversation_created
+    ON conversation_entity (createdAt, id);
+
+CREATE TABLE IF NOT EXISTS conversation_reply (
+    id character varying(36) GENERATED ALWAYS AS ((json ->> 'id'::text)) STORED NOT NULL,
+    conversationId character varying(36) GENERATED ALWAYS AS
+      ((json ->> 'conversationId'::text)) STORED NOT NULL,
+    authorId character varying(36) GENERATED ALWAYS AS
+      ((json #>> '{author,id}'::text[])) STORED NOT NULL,
+    createdAt bigint GENERATED ALWAYS AS
+      (((json ->> 'createdAt'::text))::bigint) STORED NOT NULL,
+    updatedAt bigint GENERATED ALWAYS AS
+      (((json ->> 'updatedAt'::text))::bigint) STORED NOT NULL,
+    json jsonb NOT NULL,
+    PRIMARY KEY (id),
+    CONSTRAINT fk_conversation_reply_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_conversation_reply_cursor
+    ON conversation_reply (conversationId, createdAt, id);
+CREATE INDEX IF NOT EXISTS idx_conversation_reply_author
+    ON conversation_reply (authorId, createdAt, id);
+
+CREATE TABLE IF NOT EXISTS conversation_mention (
+    conversationId character varying(36) NOT NULL,
+    targetType character varying(16) NOT NULL,
+    targetId character varying(36) NOT NULL,
+    mentionedEntityType character varying(64) NOT NULL,
+    mentionedEntityId character varying(36) NOT NULL,
+    createdAt bigint NOT NULL,
+    PRIMARY KEY (targetType, targetId, mentionedEntityType, mentionedEntityId),
+    CONSTRAINT fk_conversation_mention_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_conversation_mention_lookup
+    ON conversation_mention (mentionedEntityType, mentionedEntityId, createdAt, conversationId);
+CREATE INDEX IF NOT EXISTS idx_conversation_mention_conversation
+    ON conversation_mention (conversationId, targetType, targetId);
+
+CREATE TABLE IF NOT EXISTS conversation_domain (
+    conversationId character varying(36) NOT NULL,
+    domainId character varying(36) NOT NULL,
+    PRIMARY KEY (conversationId, domainId),
+    CONSTRAINT fk_conversation_domain_conversation
+      FOREIGN KEY (conversationId) REFERENCES conversation_entity(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_conversation_domain_lookup
+    ON conversation_domain (domainId, conversationId);
+
+ALTER TABLE conversation_entity DROP COLUMN IF EXISTS activityTimestamp;
 -- Pipeline-backed lineage is the only relationship lookup whose selective identifier lives in JSON.
 -- The partial index avoids write amplification for relationships that have no pipeline metadata.
 CREATE INDEX IF NOT EXISTS idx_entity_relationship_pipeline_relation