RDF Knowledge Graph epic with requirements including structured RDF modeling, PROV-O lineage representation, idempotent indexing, inference engine support, and scalable indexing architecture. (#24911)

* RDF Knowledge Graph epic with requirements including structured RDF modeling, PROV-O lineage representation, idempotent indexing, inference engine support, and scalable indexing architecture.

* Update generated TypeScript types

* RDF Knowledge Graph epic with requirements including structured RDF modeling, PROV-O lineage representation, idempotent indexing, inference engine support, and scalable indexing architecture.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: harshach

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

@@ -44,7 +44,6 @@ private RdfRepository(RdfConfiguration config) {
           new JsonLdTranslator(JsonUtils.getObjectMapper(), config.getBaseUri().toString());
       LOG.info("RDF Repository initialized with {} storage", config.getStorageType());
 
-      // Load ontologies on initialization
       loadOntologies();
     } else {
       this.storageService = null;
@@ -124,7 +123,6 @@ public void delete(EntityReference entityReference) {
               + "/"
               + entityReference.getId();
 
-      // Remove entity and all its relationships from the knowledge graph
       String sparqlUpdate =
           String.format(
               "DELETE WHERE { GRAPH <%s> { <%s> ?p ?o } }; "
@@ -144,10 +142,8 @@ public void addRelationship(EntityRelationship relationship) {
     }
 
     try {
-      // Create a model for the relationship with proper RDF predicates
       Model relationshipModel = createRelationshipModel(relationship);
 
-      // Store the relationship model
       String fromUri =
           config.getBaseUri().toString()
               + "entity/"
@@ -194,7 +190,6 @@ public void addRelationship(EntityRelationship relationship) {
   private Model createRelationshipModel(EntityRelationship relationship) {
     Model model = ModelFactory.createDefaultModel();
 
-    // Set namespace prefixes
     model.setNsPrefix("om", "https://open-metadata.org/ontology/");
     model.setNsPrefix("prov", "http://www.w3.org/ns/prov#");
     model.setNsPrefix("dct", "http://purl.org/dc/terms/");
@@ -215,18 +210,15 @@ private Model createRelationshipModel(EntityRelationship relationship) {
     Resource fromResource = model.createResource(fromUri);
     Resource toResource = model.createResource(toUri);
 
-    // Map relationship type to RDF predicate based on entityRelationship context
     String relationshipType = relationship.getRelationshipType().value();
     Property predicate = getRelationshipPredicate(relationshipType, model);
 
-    // Add the relationship triple
     fromResource.addProperty(predicate, toResource);
 
     return model;
   }
 
   private Property getRelationshipPredicate(String relationshipType, Model model) {
-    // Map OpenMetadata relationship types to RDF predicates from context
     return switch (relationshipType.toLowerCase()) {
       case "contains" -> model.createProperty("https://open-metadata.org/ontology/", "contains");
       case "uses" -> model.createProperty("http://www.w3.org/ns/prov#", "used");
@@ -268,6 +260,189 @@ public void bulkAddRelationships(List<EntityRelationship> relationships) {
     }
   }
 
+  /**
+   * Add a lineage relationship with full details (SQL query, pipeline, column lineage). This stores
+   * the lineage as structured RDF triples instead of a single JSON literal, enabling rich SPARQL
+   * queries like: "Find all tables derived from table X via pipeline Y" or "What columns from
+   * source table feed into column Z"
+   */
+  public void addLineageWithDetails(
+      String fromType,
+      UUID fromId,
+      String toType,
+      UUID toId,
+      org.openmetadata.schema.type.LineageDetails lineageDetails) {
+    if (!isEnabled()) {
+      return;
+    }
+
+    try {
+      Model model = ModelFactory.createDefaultModel();
+
+      model.setNsPrefix("om", "https://open-metadata.org/ontology/");
+      model.setNsPrefix("prov", "http://www.w3.org/ns/prov#");
+      model.setNsPrefix("dct", "http://purl.org/dc/terms/");
+
+      String fromUri = config.getBaseUri().toString() + "entity/" + fromType + "/" + fromId;
+      String toUri = config.getBaseUri().toString() + "entity/" + toType + "/" + toId;
+
+      Resource fromResource = model.createResource(fromUri);
+      Resource toResource = model.createResource(toUri);
+
+      // PROV-O: to wasDerivedFrom from (reverse direction for semantic correctness)
+      Property derivedFrom = model.createProperty("http://www.w3.org/ns/prov#", "wasDerivedFrom");
+      toResource.addProperty(derivedFrom, fromResource);
+
+      // OpenMetadata-specific upstream for compatibility
+      Property upstream = model.createProperty("https://open-metadata.org/ontology/", "UPSTREAM");
+      fromResource.addProperty(upstream, toResource);
+
+      if (lineageDetails != null) {
+        String detailsUri =
+            config.getBaseUri().toString()
+                + "lineageDetails/"
+                + fromId
+                + "/"
+                + toId
+                + "/"
+                + System.currentTimeMillis();
+        Resource detailsResource = model.createResource(detailsUri);
+
+        Property hasLineageDetails =
+            model.createProperty("https://open-metadata.org/ontology/", "hasLineageDetails");
+        fromResource.addProperty(hasLineageDetails, detailsResource);
+
+        detailsResource.addProperty(
+            model.createProperty("http://www.w3.org/1999/02/22-rdf-syntax-ns#", "type"),
+            model.createResource("https://open-metadata.org/ontology/LineageDetails"));
+
+        if (lineageDetails.getSqlQuery() != null && !lineageDetails.getSqlQuery().isEmpty()) {
+          detailsResource.addProperty(
+              model.createProperty("https://open-metadata.org/ontology/", "sqlQuery"),
+              lineageDetails.getSqlQuery());
+        }
+
+        if (lineageDetails.getSource() != null) {
+          detailsResource.addProperty(
+              model.createProperty("https://open-metadata.org/ontology/", "lineageSource"),
+              lineageDetails.getSource().value());
+        }
+
+        if (lineageDetails.getDescription() != null && !lineageDetails.getDescription().isEmpty()) {
+          detailsResource.addProperty(
+              model.createProperty("http://purl.org/dc/terms/", "description"),
+              lineageDetails.getDescription());
+        }
+
+        if (lineageDetails.getPipeline() != null && lineageDetails.getPipeline().getId() != null) {
+          EntityReference pipeline = lineageDetails.getPipeline();
+          String pipelineType = pipeline.getType() != null ? pipeline.getType() : "pipeline";
+          String pipelineUri =
+              config.getBaseUri().toString() + "entity/" + pipelineType + "/" + pipeline.getId();
+          Resource pipelineResource = model.createResource(pipelineUri);
+
+          detailsResource.addProperty(
+              model.createProperty("http://www.w3.org/ns/prov#", "wasGeneratedBy"),
+              pipelineResource);
+        }
+
+        if (lineageDetails.getColumnsLineage() != null
+            && !lineageDetails.getColumnsLineage().isEmpty()) {
+          Property hasColumnLineage =
+              model.createProperty("https://open-metadata.org/ontology/", "hasColumnLineage");
+
+          for (org.openmetadata.schema.type.ColumnLineage colLineage :
+              lineageDetails.getColumnsLineage()) {
+            String colLineageUri = detailsUri + "/columnLineage/" + System.nanoTime();
+            Resource colLineageResource = model.createResource(colLineageUri);
+
+            detailsResource.addProperty(hasColumnLineage, colLineageResource);
+            colLineageResource.addProperty(
+                model.createProperty("http://www.w3.org/1999/02/22-rdf-syntax-ns#", "type"),
+                model.createResource("https://open-metadata.org/ontology/ColumnLineage"));
+
+            if (colLineage.getFromColumns() != null) {
+              Property fromColumnProp =
+                  model.createProperty("https://open-metadata.org/ontology/", "fromColumn");
+              for (String fromCol : colLineage.getFromColumns()) {
+                colLineageResource.addProperty(fromColumnProp, fromCol);
+              }
+            }
+
+            if (colLineage.getToColumn() != null) {
+              colLineageResource.addProperty(
+                  model.createProperty("https://open-metadata.org/ontology/", "toColumn"),
+                  colLineage.getToColumn());
+            }
+
+            if (colLineage.getFunction() != null) {
+              colLineageResource.addProperty(
+                  model.createProperty("https://open-metadata.org/ontology/", "transformFunction"),
+                  colLineage.getFunction());
+            }
+          }
+        }
+
+        if (lineageDetails.getCreatedAt() != null) {
+          detailsResource.addProperty(
+              model.createProperty("http://purl.org/dc/terms/", "created"),
+              model.createTypedLiteral(
+                  lineageDetails.getCreatedAt().toString(),
+                  org.apache.jena.datatypes.xsd.XSDDatatype.XSDlong));
+        }
+        if (lineageDetails.getUpdatedAt() != null) {
+          detailsResource.addProperty(
+              model.createProperty("http://purl.org/dc/terms/", "modified"),
+              model.createTypedLiteral(
+                  lineageDetails.getUpdatedAt().toString(),
+                  org.apache.jena.datatypes.xsd.XSDDatatype.XSDlong));
+        }
+
+        if (lineageDetails.getCreatedBy() != null) {
+          detailsResource.addProperty(
+              model.createProperty("https://open-metadata.org/ontology/", "lineageCreatedBy"),
+              lineageDetails.getCreatedBy());
+        }
+        if (lineageDetails.getUpdatedBy() != null) {
+          detailsResource.addProperty(
+              model.createProperty("https://open-metadata.org/ontology/", "lineageUpdatedBy"),
+              lineageDetails.getUpdatedBy());
+        }
+      }
+
+      // Idempotent delete/insert pattern ensures no duplicate triples
+      java.io.StringWriter writer = new java.io.StringWriter();
+      model.write(writer, "N-TRIPLES");
+      String triples = writer.toString();
+
+      if (!triples.isEmpty()) {
+        String deleteQuery =
+            String.format(
+                "DELETE WHERE { GRAPH <%s> { <%s> <https://open-metadata.org/ontology/UPSTREAM> <%s> . } }; "
+                    + "DELETE WHERE { GRAPH <%s> { <%s> <http://www.w3.org/ns/prov#wasDerivedFrom> <%s> . } }",
+                KNOWLEDGE_GRAPH, fromUri, toUri, KNOWLEDGE_GRAPH, toUri, fromUri);
+
+        storageService.executeSparqlUpdate(deleteQuery);
+
+        StringBuilder insertQuery = new StringBuilder();
+        insertQuery.append("INSERT DATA { GRAPH <").append(KNOWLEDGE_GRAPH).append("> { ");
+        insertQuery.append(triples);
+        insertQuery.append(" } }");
+
+        storageService.executeSparqlUpdate(insertQuery.toString());
+        LOG.debug("Added lineage with details from {}/{} to {}/{}", fromType, fromId, toType, toId);
+      }
+    } catch (Exception e) {
+      LOG.error(
+          "Failed to add lineage with details from {}/{} to {}/{}",
+          fromType,
+          fromId,
+          toType,
+          toId,
+          e);
+    }
+  }
+
   public void removeRelationship(EntityRelationship relationship) {
     if (!isEnabled()) {
       return;
@@ -355,9 +530,41 @@ public String executeSparqlQuery(String query, String format) {
       throw new IllegalStateException("RDF not enabled");
     }
 
+    // Check if inference is enabled by default in configuration
+    if (isInferenceEnabledByDefault()) {
+      String defaultLevel = getDefaultInferenceLevel();
+      return executeSparqlQueryWithInference(query, format, defaultLevel);
+    }
+
     return storageService.executeSparqlQuery(query, format);
   }
 
+  /**
+   * Execute SPARQL query without inference, regardless of configuration. Use this for internal
+   * queries where inference overhead is not needed.
+   */
+  public String executeSparqlQueryDirect(String query, String format) {
+    if (!isEnabled()) {
+      throw new IllegalStateException("RDF not enabled");
+    }
+    return storageService.executeSparqlQuery(query, format);
+  }
+
+  public boolean isInferenceEnabledByDefault() {
+    return config.getInferenceEnabled() != null && config.getInferenceEnabled();
+  }
+
+  public String getDefaultInferenceLevel() {
+    if (config.getDefaultInferenceLevel() != null) {
+      return config.getDefaultInferenceLevel().value();
+    }
+    return "NONE";
+  }
+
+  public RdfConfiguration getConfig() {
+    return config;
+  }
+
   public List<Map<String, String>> executeSparqlQueryAsJson(String query) {
     String result = executeSparqlQuery(query, "json");
     return parseSparqlJsonResults(result);