fix(rdf): converge Fuseki state on weekly rebuilds and isolate API latency

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

@@ -1 +1,18 @@
 -- Post data migration script for Task workflow cutover - OpenMetadata 2.0.1
+
+-- RdfIndexApp: switch to weekly Saturday cron and recreate-on-each-run.
+-- Previous defaults (daily, incremental) were producing unbounded triple growth
+-- because relationship-removal paths weren't fully reconciled. With per-run
+-- CLEAR ALL the dataset always converges to the current MySQL state; weekly
+-- cadence keeps the per-run cost from saturating Fuseki.
+UPDATE installed_apps
+SET json = JSON_SET(
+    json,
+    '$.appConfiguration.recreateIndex', CAST('true' AS JSON),
+    '$.appSchedule.cronExpression', '0 0 * * 6'
+)
+WHERE name = 'RdfIndexApp';
+
+UPDATE apps_marketplace
+SET json = JSON_SET(json, '$.appConfiguration.recreateIndex', CAST('true' AS JSON))
+WHERE name = 'RdfIndexApp';

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

@@ -1 +1,18 @@
 -- Post data migration script for Task workflow cutover - OpenMetadata 2.0.1
+
+-- RdfIndexApp: switch to weekly Saturday cron and recreate-on-each-run.
+-- Previous defaults (daily, incremental) were producing unbounded triple growth
+-- because relationship-removal paths weren't fully reconciled. With per-run
+-- CLEAR ALL the dataset always converges to the current MySQL state; weekly
+-- cadence keeps the per-run cost from saturating Fuseki.
+UPDATE installed_apps
+SET json = jsonb_set(
+    jsonb_set(json::jsonb, '{appConfiguration,recreateIndex}', 'true'),
+    '{appSchedule,cronExpression}',
+    '"0 0 * * 6"'
+)
+WHERE name = 'RdfIndexApp';
+
+UPDATE apps_marketplace
+SET json = jsonb_set(json::jsonb, '{appConfiguration,recreateIndex}', 'true')
+WHERE name = 'RdfIndexApp';

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/rdf/RdfIndexApp.java

@@ -215,6 +215,21 @@ private void initializeJob(JobExecutionContext jobExecutionContext) {
     rdfIndexStats.set(initializeTotalRecords(jobData.getEntities()));
     jobData.setStats(rdfIndexStats.get());
 
+    // bulkAddGlossaryTermRelations has no per-batch DELETE side, so stale
+    // glossary-term relations would accumulate forever across reindex runs.
+    // When recreateIndex=true clearAll() already wipes everything, so we
+    // only need this targeted cleanup on incremental runs.
+    if (!Boolean.TRUE.equals(jobData.getRecreateIndex())
+        && jobData.getEntities() != null
+        && jobData.getEntities().contains(Entity.GLOSSARY_TERM)) {
+      LOG.info("Clearing existing glossary term relations before re-indexing");
+      try {
+        rdfRepository.clearAllGlossaryTermRelations();
+      } catch (Exception e) {
+        LOG.warn("Failed to clear glossary term relations; continuing with reindex", e);
+      }
+    }
+
     if (Boolean.TRUE.equals(jobData.getUseDistributedIndexing())) {
       sendUpdates(jobExecutionContext, true);
       return;
@@ -242,6 +257,10 @@ private void clearRdfData() {
     try {
       rdfRepository.clearAll();
       LOG.info("Cleared all RDF data");
+      // CLEAR ALL wipes the ontology and shapes graphs as well; reload them
+      // before indexing starts so SPARQL queries that depend on the ontology
+      // (inference, federated, etc.) work after the wipe.
+      rdfRepository.reloadOntologies();
     } catch (Exception e) {
       LOG.error("Failed to clear RDF data", e);
       throw new RuntimeException("Failed to clear RDF data", e);

openmetadata-service/src/main/java/org/openmetadata/service/rdf/RdfRepository.java

@@ -70,6 +70,23 @@ private void loadOntologies() {
     }
   }
 
+  // CLEAR ALL (called by clearAll()) wipes the ontology and shapes graphs too.
+  // Callers that wipe the dataset must invoke this afterwards so SPARQL queries
+  // that depend on the ontology don't break. Unlike loadOntologies() this skips
+  // the "already loaded" guard — areOntologiesLoaded() would return false right
+  // after a CLEAR, but we want to unconditionally reload.
+  public void reloadOntologies() {
+    if (!isEnabled()) {
+      return;
+    }
+    try {
+      new OntologyLoader(this).loadOntologies();
+      LOG.info("Reloaded OpenMetadata ontologies into RDF store");
+    } catch (Exception e) {
+      LOG.error("Failed to reload ontologies", e);
+    }
+  }
+
   public static void initialize(RdfConfiguration config) {
     if (INSTANCE != null) {
       throw new IllegalStateException("RdfRepository already initialized");
@@ -127,24 +144,6 @@ public void createOrUpdate(EntityInterface entity) {
           entity.getName(),
           entity.getId());
       Model rdfModel = translator.toRdf(entity);
-
-      // Preserve existing relationship triples before updating
-      // This prevents postCreate() from overwriting relationships added by storeRelationships()
-      Model existingModel = storageService.getEntity(entityType, entity.getId());
-      if (existingModel != null && !existingModel.isEmpty()) {
-        String entityUri =
-            config.getBaseUri().toString() + "entity/" + entityType + "/" + entity.getId();
-        // Extract and preserve relationship triples (where entity is subject and object is a URI)
-        Model relationshipTriples = extractRelationshipTriples(existingModel, entityUri);
-        if (!relationshipTriples.isEmpty()) {
-          rdfModel.add(relationshipTriples);
-          LOG.debug(
-              "Preserved {} relationship triples for entity {}",
-              relationshipTriples.size(),
-              entity.getId());
-        }
-      }
-
       storageService.storeEntity(entityType, entity.getId(), rdfModel);
       LOG.debug("Created/Updated entity {} in RDF store", entity.getId());
     } catch (Exception e) {
@@ -154,33 +153,10 @@ public void createOrUpdate(EntityInterface entity) {
           entity.getEntityReference().getType(),
           entity.getFullyQualifiedName(),
           e);
-      // Rethrow so callers (e.g. RdfBatchProcessor) can count this as a failure instead of
-      // reporting a false success. Entity-hook callers (RdfUpdater) already wrap in try/catch.
       throw new RuntimeException("Failed to create/update entity in RDF", e);
     }
   }
 
-  private Model extractRelationshipTriples(Model model, String entityUri) {
-    Model relationshipTriples = ModelFactory.createDefaultModel();
-    Resource entityResource = model.createResource(entityUri);
-
-    // Find all triples where entity is subject and object is a URI resource (relationships)
-    model
-        .listStatements(entityResource, null, (org.apache.jena.rdf.model.RDFNode) null)
-        .forEachRemaining(
-            stmt -> {
-              if (stmt.getObject().isURIResource()) {
-                String objectUri = stmt.getObject().asResource().getURI();
-                // Only preserve triples that link to other entities (not type/label predicates)
-                if (objectUri.contains("/entity/")) {
-                  relationshipTriples.add(stmt);
-                }
-              }
-            });
-
-    return relationshipTriples;
-  }
-
   public void delete(EntityReference entityReference) {
     if (!isEnabled()) {
       return;

openmetadata-service/src/main/java/org/openmetadata/service/rdf/RdfUpdater.java

@@ -1,16 +1,23 @@
 package org.openmetadata.service.rdf;
 
 import io.micrometer.core.instrument.Timer;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.EntityInterface;
 import org.openmetadata.schema.api.configuration.rdf.RdfConfiguration;
 import org.openmetadata.schema.type.EntityReference;
 import org.openmetadata.schema.type.EntityRelationship;
 import org.openmetadata.service.monitoring.RequestLatencyContext;
+import org.openmetadata.service.util.AsyncService;
 
 @Slf4j
 public class RdfUpdater {
 
+  private static final int MAX_PENDING_RDF_WRITES = 1000;
+  private static final AtomicInteger pendingWrites = new AtomicInteger(0);
+  private static final AtomicLong droppedWrites = new AtomicLong(0L);
+
   private static RdfRepository rdfRepository;
 
   private RdfUpdater() {}
@@ -26,55 +33,75 @@ public static void initialize(RdfConfiguration config) {
   }
 
   public static void updateEntity(EntityInterface entity) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      Timer.Sample sample = RequestLatencyContext.startRdfOperation();
-      try {
-        rdfRepository.createOrUpdate(entity);
-      } catch (Exception e) {
-        LOG.error("Failed to update entity {} in RDF", entity.getId(), e);
-      } finally {
-        RequestLatencyContext.endRdfOperation(sample);
-      }
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
     }
+    submitAsync(
+        "updateEntity " + entity.getId(),
+        () -> {
+          Timer.Sample sample = RequestLatencyContext.startRdfOperation();
+          try {
+            rdfRepository.createOrUpdate(entity);
+          } catch (Exception e) {
+            LOG.error("Failed to update entity {} in RDF", entity.getId(), e);
+          } finally {
+            RequestLatencyContext.endRdfOperation(sample);
+          }
+        });
   }
 
   public static void deleteEntity(EntityReference entityReference) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      Timer.Sample sample = RequestLatencyContext.startRdfOperation();
-      try {
-        rdfRepository.delete(entityReference);
-      } catch (Exception e) {
-        LOG.error("Failed to delete entity {} in RDF", entityReference.getId(), e);
-      } finally {
-        RequestLatencyContext.endRdfOperation(sample);
-      }
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
     }
+    submitAsync(
+        "deleteEntity " + entityReference.getId(),
+        () -> {
+          Timer.Sample sample = RequestLatencyContext.startRdfOperation();
+          try {
+            rdfRepository.delete(entityReference);
+          } catch (Exception e) {
+            LOG.error("Failed to delete entity {} in RDF", entityReference.getId(), e);
+          } finally {
+            RequestLatencyContext.endRdfOperation(sample);
+          }
+        });
   }
 
   public static void addRelationship(EntityRelationship relationship) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      Timer.Sample sample = RequestLatencyContext.startRdfOperation();
-      try {
-        rdfRepository.addRelationship(relationship);
-      } catch (Exception e) {
-        LOG.error("Failed to add relationship in RDF", e);
-      } finally {
-        RequestLatencyContext.endRdfOperation(sample);
-      }
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
     }
+    submitAsync(
+        "addRelationship",
+        () -> {
+          Timer.Sample sample = RequestLatencyContext.startRdfOperation();
+          try {
+            rdfRepository.addRelationship(relationship);
+          } catch (Exception e) {
+            LOG.error("Failed to add relationship in RDF", e);
+          } finally {
+            RequestLatencyContext.endRdfOperation(sample);
+          }
+        });
   }
 
   public static void removeRelationship(EntityRelationship relationship) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      Timer.Sample sample = RequestLatencyContext.startRdfOperation();
-      try {
-        rdfRepository.removeRelationship(relationship);
-      } catch (Exception e) {
-        LOG.error("Failed to remove relationship in RDF", e);
-      } finally {
-        RequestLatencyContext.endRdfOperation(sample);
-      }
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
     }
+    submitAsync(
+        "removeRelationship",
+        () -> {
+          Timer.Sample sample = RequestLatencyContext.startRdfOperation();
+          try {
+            rdfRepository.removeRelationship(relationship);
+          } catch (Exception e) {
+            LOG.error("Failed to remove relationship in RDF", e);
+          } finally {
+            RequestLatencyContext.endRdfOperation(sample);
+          }
+        });
   }
 
   public static boolean isEnabled() {
@@ -89,33 +116,79 @@ public static void disable() {
 
   public static void addGlossaryTermRelation(
       java.util.UUID fromTermId, java.util.UUID toTermId, String relationType) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      try {
-        rdfRepository.addGlossaryTermRelation(fromTermId, toTermId, relationType);
-      } catch (Exception e) {
-        LOG.error(
-            "Failed to add glossary term relation {} -> {} ({}) to RDF",
-            fromTermId,
-            toTermId,
-            relationType,
-            e);
-      }
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
     }
+    submitAsync(
+        "addGlossaryTermRelation",
+        () -> {
+          try {
+            rdfRepository.addGlossaryTermRelation(fromTermId, toTermId, relationType);
+          } catch (Exception e) {
+            LOG.error(
+                "Failed to add glossary term relation {} -> {} ({}) to RDF",
+                fromTermId,
+                toTermId,
+                relationType,
+                e);
+          }
+        });
   }
 
   public static void removeGlossaryTermRelation(
       java.util.UUID fromTermId, java.util.UUID toTermId, String relationType) {
-    if (rdfRepository != null && rdfRepository.isEnabled()) {
-      try {
-        rdfRepository.removeGlossaryTermRelation(fromTermId, toTermId, relationType);
-      } catch (Exception e) {
-        LOG.error(
-            "Failed to remove glossary term relation {} -> {} ({}) from RDF",
-            fromTermId,
-            toTermId,
-            relationType,
-            e);
+    if (rdfRepository == null || !rdfRepository.isEnabled()) {
+      return;
+    }
+    submitAsync(
+        "removeGlossaryTermRelation",
+        () -> {
+          try {
+            rdfRepository.removeGlossaryTermRelation(fromTermId, toTermId, relationType);
+          } catch (Exception e) {
+            LOG.error(
+                "Failed to remove glossary term relation {} -> {} ({}) from RDF",
+                fromTermId,
+                toTermId,
+                relationType,
+                e);
+          }
+        });
+  }
+
+  // Bounded fire-and-forget submission: a request thread that triggers an RDF
+  // write must NOT wait for Fuseki. We submit to AsyncService (virtual-thread
+  // pool) but gate first on a soft cap of in-flight writes so that, if Fuseki
+  // is unreachable and tasks pile up, we drop with a logged warning instead
+  // of spawning unbounded virtual threads. RDF is a derived index — missed
+  // writes are reconciled by the weekly RdfIndexApp run.
+  private static void submitAsync(String description, Runnable task) {
+    int newCount = pendingWrites.incrementAndGet();
+    if (newCount > MAX_PENDING_RDF_WRITES) {
+      pendingWrites.decrementAndGet();
+      long dropped = droppedWrites.incrementAndGet();
+      if (dropped == 1 || dropped % 100 == 0) {
+        LOG.warn(
+            "Dropping RDF {} due to backpressure (pending={}, total dropped={})",
+            description,
+            newCount - 1,
+            dropped);
       }
+      return;
+    }
+    try {
+      AsyncService.getInstance()
+          .execute(
+              () -> {
+                try {
+                  task.run();
+                } finally {
+                  pendingWrites.decrementAndGet();
+                }
+              });
+    } catch (RuntimeException e) {
+      pendingWrites.decrementAndGet();
+      LOG.error("Failed to submit RDF {} to async executor", description, e);
     }
   }
 }