RDF, cleanup relations and remove unnecessary bindings, add distributed mode for RDF reindex

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

@@ -84,3 +84,82 @@ SELECT ue.id, re.id, 'user', 'role', 10
 FROM user_entity ue, role_entity re
 WHERE ue.name = 'mcpapplicationbot'
   AND re.name = 'ApplicationBotImpersonationRole';
+
+-- RDF distributed indexing state tables
+CREATE TABLE IF NOT EXISTS rdf_index_job (
+    id VARCHAR(36) NOT NULL,
+    status VARCHAR(32) NOT NULL,
+    jobConfiguration JSON NOT NULL,
+    totalRecords BIGINT NOT NULL DEFAULT 0,
+    processedRecords BIGINT NOT NULL DEFAULT 0,
+    successRecords BIGINT NOT NULL DEFAULT 0,
+    failedRecords BIGINT NOT NULL DEFAULT 0,
+    stats JSON,
+    createdBy VARCHAR(256) NOT NULL,
+    createdAt BIGINT NOT NULL,
+    startedAt BIGINT,
+    completedAt BIGINT,
+    updatedAt BIGINT NOT NULL,
+    errorMessage TEXT,
+    PRIMARY KEY (id),
+    INDEX idx_rdf_index_job_status (status),
+    INDEX idx_rdf_index_job_created (createdAt DESC)
+);
+
+CREATE TABLE IF NOT EXISTS rdf_index_partition (
+    id VARCHAR(36) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    entityType VARCHAR(128) NOT NULL,
+    partitionIndex INT NOT NULL,
+    rangeStart BIGINT NOT NULL,
+    rangeEnd BIGINT NOT NULL,
+    estimatedCount BIGINT NOT NULL,
+    workUnits BIGINT NOT NULL,
+    priority INT NOT NULL DEFAULT 50,
+    status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
+    processingCursor BIGINT NOT NULL DEFAULT 0,
+    processedCount BIGINT NOT NULL DEFAULT 0,
+    successCount BIGINT NOT NULL DEFAULT 0,
+    failedCount BIGINT NOT NULL DEFAULT 0,
+    assignedServer VARCHAR(255),
+    claimedAt BIGINT,
+    startedAt BIGINT,
+    completedAt BIGINT,
+    lastUpdateAt BIGINT,
+    lastError TEXT,
+    retryCount INT NOT NULL DEFAULT 0,
+    claimableAt BIGINT NOT NULL DEFAULT 0,
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_rdf_partition_job_entity_idx (jobId, entityType, partitionIndex),
+    INDEX idx_rdf_partition_job (jobId),
+    INDEX idx_rdf_partition_status_priority (status, priority DESC),
+    INDEX idx_rdf_partition_claimable (jobId, status, claimableAt),
+    INDEX idx_rdf_partition_assigned_server (jobId, assignedServer),
+    CONSTRAINT fk_rdf_partition_job FOREIGN KEY (jobId) REFERENCES rdf_index_job(id) ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS rdf_reindex_lock (
+    lockKey VARCHAR(64) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    serverId VARCHAR(255) NOT NULL,
+    acquiredAt BIGINT NOT NULL,
+    lastHeartbeat BIGINT NOT NULL,
+    expiresAt BIGINT NOT NULL,
+    PRIMARY KEY (lockKey)
+);
+
+CREATE TABLE IF NOT EXISTS rdf_index_server_stats (
+    id VARCHAR(36) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    serverId VARCHAR(256) NOT NULL,
+    entityType VARCHAR(128) NOT NULL,
+    processedRecords BIGINT DEFAULT 0,
+    successRecords BIGINT DEFAULT 0,
+    failedRecords BIGINT DEFAULT 0,
+    partitionsCompleted INT DEFAULT 0,
+    partitionsFailed INT DEFAULT 0,
+    lastUpdatedAt BIGINT NOT NULL,
+    PRIMARY KEY (id),
+    UNIQUE INDEX idx_rdf_index_server_stats_job_server_entity (jobId, serverId, entityType),
+    INDEX idx_rdf_index_server_stats_job_id (jobId)
+);

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

@@ -96,3 +96,85 @@ FROM user_entity ue, role_entity re
 WHERE ue.name = 'mcpapplicationbot'
   AND re.name = 'ApplicationBotImpersonationRole'
 ON CONFLICT DO NOTHING;
+
+-- RDF distributed indexing state tables
+CREATE TABLE IF NOT EXISTS rdf_index_job (
+    id VARCHAR(36) NOT NULL,
+    status VARCHAR(32) NOT NULL,
+    jobConfiguration JSONB NOT NULL,
+    totalRecords BIGINT NOT NULL DEFAULT 0,
+    processedRecords BIGINT NOT NULL DEFAULT 0,
+    successRecords BIGINT NOT NULL DEFAULT 0,
+    failedRecords BIGINT NOT NULL DEFAULT 0,
+    stats JSONB,
+    createdBy VARCHAR(256) NOT NULL,
+    createdAt BIGINT NOT NULL,
+    startedAt BIGINT,
+    completedAt BIGINT,
+    updatedAt BIGINT NOT NULL,
+    errorMessage TEXT,
+    PRIMARY KEY (id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_rdf_index_job_status ON rdf_index_job(status);
+CREATE INDEX IF NOT EXISTS idx_rdf_index_job_created ON rdf_index_job(createdAt DESC);
+
+CREATE TABLE IF NOT EXISTS rdf_index_partition (
+    id VARCHAR(36) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    entityType VARCHAR(128) NOT NULL,
+    partitionIndex INT NOT NULL,
+    rangeStart BIGINT NOT NULL,
+    rangeEnd BIGINT NOT NULL,
+    estimatedCount BIGINT NOT NULL,
+    workUnits BIGINT NOT NULL,
+    priority INT NOT NULL DEFAULT 50,
+    status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
+    processingCursor BIGINT NOT NULL DEFAULT 0,
+    processedCount BIGINT NOT NULL DEFAULT 0,
+    successCount BIGINT NOT NULL DEFAULT 0,
+    failedCount BIGINT NOT NULL DEFAULT 0,
+    assignedServer VARCHAR(255),
+    claimedAt BIGINT,
+    startedAt BIGINT,
+    completedAt BIGINT,
+    lastUpdateAt BIGINT,
+    lastError TEXT,
+    retryCount INT NOT NULL DEFAULT 0,
+    claimableAt BIGINT NOT NULL DEFAULT 0,
+    PRIMARY KEY (id),
+    UNIQUE (jobId, entityType, partitionIndex),
+    CONSTRAINT fk_rdf_partition_job FOREIGN KEY (jobId) REFERENCES rdf_index_job(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_rdf_partition_job ON rdf_index_partition(jobId);
+CREATE INDEX IF NOT EXISTS idx_rdf_partition_status_priority ON rdf_index_partition(status, priority DESC);
+CREATE INDEX IF NOT EXISTS idx_rdf_partition_claimable ON rdf_index_partition(jobId, status, claimableAt);
+CREATE INDEX IF NOT EXISTS idx_rdf_partition_assigned_server ON rdf_index_partition(jobId, assignedServer);
+
+CREATE TABLE IF NOT EXISTS rdf_reindex_lock (
+    lockKey VARCHAR(64) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    serverId VARCHAR(255) NOT NULL,
+    acquiredAt BIGINT NOT NULL,
+    lastHeartbeat BIGINT NOT NULL,
+    expiresAt BIGINT NOT NULL,
+    PRIMARY KEY (lockKey)
+);
+
+CREATE TABLE IF NOT EXISTS rdf_index_server_stats (
+    id VARCHAR(36) NOT NULL,
+    jobId VARCHAR(36) NOT NULL,
+    serverId VARCHAR(256) NOT NULL,
+    entityType VARCHAR(128) NOT NULL,
+    processedRecords BIGINT DEFAULT 0,
+    successRecords BIGINT DEFAULT 0,
+    failedRecords BIGINT DEFAULT 0,
+    partitionsCompleted INT DEFAULT 0,
+    partitionsFailed INT DEFAULT 0,
+    lastUpdatedAt BIGINT NOT NULL,
+    PRIMARY KEY (id),
+    UNIQUE (jobId, serverId, entityType)
+);
+
+CREATE INDEX IF NOT EXISTS idx_rdf_index_server_stats_job_id ON rdf_index_server_stats(jobId);

docker/development/docker-compose-fuseki.yml

@@ -1,12 +1,44 @@
 version: "3.9"
 
 services:
+  execute-migrate-all:
+    environment:
+      RDF_ENABLED: ${RDF_ENABLED:-true}
+      RDF_STORAGE_TYPE: ${RDF_STORAGE_TYPE:-FUSEKI}
+      RDF_ENDPOINT: ${RDF_ENDPOINT:-http://fuseki:3030/openmetadata}
+      RDF_REMOTE_USERNAME: ${RDF_REMOTE_USERNAME:-admin}
+      RDF_REMOTE_PASSWORD: ${RDF_REMOTE_PASSWORD:-admin}
+      RDF_BASE_URI: ${RDF_BASE_URI:-https://open-metadata.org/}
+      RDF_JSONLD_ENABLED: ${RDF_JSONLD_ENABLED:-true}
+      RDF_SPARQL_ENABLED: ${RDF_SPARQL_ENABLED:-true}
+      RDF_DATASET: ${RDF_DATASET:-openmetadata}
+    depends_on:
+      fuseki:
+        condition: service_healthy
+
+  openmetadata-server:
+    environment:
+      RDF_ENABLED: ${RDF_ENABLED:-true}
+      RDF_STORAGE_TYPE: ${RDF_STORAGE_TYPE:-FUSEKI}
+      RDF_ENDPOINT: ${RDF_ENDPOINT:-http://fuseki:3030/openmetadata}
+      RDF_REMOTE_USERNAME: ${RDF_REMOTE_USERNAME:-admin}
+      RDF_REMOTE_PASSWORD: ${RDF_REMOTE_PASSWORD:-admin}
+      RDF_BASE_URI: ${RDF_BASE_URI:-https://open-metadata.org/}
+      RDF_JSONLD_ENABLED: ${RDF_JSONLD_ENABLED:-true}
+      RDF_SPARQL_ENABLED: ${RDF_SPARQL_ENABLED:-true}
+      RDF_DATASET: ${RDF_DATASET:-openmetadata}
+    depends_on:
+      fuseki:
+        condition: service_healthy
+
   fuseki:
     image: stain/jena-fuseki:5.0.0
     container_name: openmetadata-fuseki
     hostname: fuseki
     ports:
       - "3030:3030"
+    networks:
+      - local_app_net
     environment:
       - ADMIN_PASSWORD=admin
       - JVM_ARGS=-Xmx4g -Xms2g
@@ -19,6 +51,11 @@ services:
           memory: 4G
         reservations:
           memory: 2G
+    healthcheck:
+      test: "curl -s -f http://localhost:3030/\\$/ping > /dev/null || exit 1"
+      interval: 15s
+      timeout: 10s
+      retries: 20
     # Create the database directory before starting Fuseki
     entrypoint: /bin/sh -c "mkdir -p /fuseki/databases/openmetadata && exec /docker-entrypoint.sh /jena-fuseki/fuseki-server --update --loc=/fuseki/databases/openmetadata /openmetadata"
 

docker/run_local_docker.sh

@@ -57,6 +57,8 @@ cd ../
 echo "Stopping any previous Local Docker Containers"
 docker compose -f docker/development/docker-compose-postgres.yml down --remove-orphans
 docker compose -f docker/development/docker-compose.yml down --remove-orphans
+docker compose -f docker/development/docker-compose-postgres.yml -f docker/development/docker-compose-fuseki.yml down --remove-orphans
+docker compose -f docker/development/docker-compose.yml -f docker/development/docker-compose-fuseki.yml down --remove-orphans
 
 if [[ $skipMaven == "false" ]]; then
     if [[ $mode == "no-ui" ]]; then
@@ -80,6 +82,14 @@ if [[ $debugOM == "true" ]]; then
  export OPENMETADATA_DEBUG=true
 fi
 
+export RDF_ENABLED=true
+export RDF_STORAGE_TYPE=FUSEKI
+export RDF_ENDPOINT="${RDF_ENDPOINT:-http://fuseki:3030/openmetadata}"
+export RDF_REMOTE_USERNAME="${RDF_REMOTE_USERNAME:-admin}"
+export RDF_REMOTE_PASSWORD="${RDF_REMOTE_PASSWORD:-admin}"
+export RDF_BASE_URI="${RDF_BASE_URI:-https://open-metadata.org/}"
+export RDF_DATASET="${RDF_DATASET:-openmetadata}"
+
 if [[ $cleanDbVolumes == "true" ]]
 then
   if [[ -d "$PWD/docker/development/docker-volume/" ]]
@@ -116,13 +126,16 @@ else
     exit 1
 fi
 
+RDF_COMPOSE_FILE="docker/development/docker-compose-fuseki.yml"
+COMPOSE_ARGS=(-f "$COMPOSE_FILE" -f "$RDF_COMPOSE_FILE")
+
 if [[ $includeIngestion == "true" ]]; then
     echo "Building all services including ingestion (dependency: ${INGESTION_DEPENDENCY:-all})"
-    docker compose -f $COMPOSE_FILE build --build-arg INGESTION_DEPENDENCY="${INGESTION_DEPENDENCY:-all}" && docker compose -f $COMPOSE_FILE up -d
+    docker compose "${COMPOSE_ARGS[@]}" build --build-arg INGESTION_DEPENDENCY="${INGESTION_DEPENDENCY:-all}" && docker compose "${COMPOSE_ARGS[@]}" up -d
 else
     echo "Building services without ingestion"
-    docker compose -f $COMPOSE_FILE build $SEARCH_SERVICE $DB_SERVICE execute-migrate-all openmetadata-server && \
-    docker compose -f $COMPOSE_FILE up -d $SEARCH_SERVICE $DB_SERVICE execute-migrate-all openmetadata-server
+    docker compose "${COMPOSE_ARGS[@]}" build $SEARCH_SERVICE $DB_SERVICE execute-migrate-all openmetadata-server && \
+    docker compose "${COMPOSE_ARGS[@]}" up -d fuseki $SEARCH_SERVICE $DB_SERVICE execute-migrate-all openmetadata-server
 fi
 
 RESULT=$?
@@ -136,6 +149,11 @@ until curl -s -f "http://localhost:9200/_cat/indices/openmetadata_team_search_in
   sleep 5
 done
 
+until curl -s -f "http://localhost:3030/\$/ping" > /dev/null 2>&1; do
+  echo 'Checking if Fuseki is reachable...\n'
+  sleep 5
+done
+
 if [[ $includeIngestion == "true" ]]; then
   # Function to get OAuth access token for Airflow API
   get_airflow_token() {
@@ -288,6 +306,22 @@ curl --location --request POST 'http://localhost:8585/api/v1/apps/trigger/Search
 --header 'Authorization: Bearer eyJraWQiOiJHYjM4OWEtOWY3Ni1nZGpzLWE5MmotMDI0MmJrOTQzNTYiLCJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiIsImlzQm90IjpmYWxzZSwiaXNzIjoib3Blbi1tZXRhZGF0YS5vcmciLCJpYXQiOjE2NjM5Mzg0NjIsImVtYWlsIjoiYWRtaW5Ab3Blbm1ldGFkYXRhLm9yZyJ9.tS8um_5DKu7HgzGBzS1VTA5uUjKWOCU0B_j08WXBiEC0mr0zNREkqVfwFDD-d24HlNEbrqioLsBuFRiwIWKc1m_ZlVQbG7P36RUxhuv2vbSp80FKyNM-Tj93FDzq91jsyNmsQhyNv_fNr3TXfzzSPjHt8Go0FMMP66weoKMgW2PbXlhVKwEuXUHyakLLzewm9UMeQaEiRzhiTMU3UkLXcKbYEJJvfNFcLwSl9W8JCO_l0Yj3ud-qt_nQYEZwqW6u5nfdQllN133iikV4fM5QZsMCnm8Rq1mvLR0y9bmJiD7fwM1tmJ791TUWqmKaTnP49U493VanKpUAfzIiOiIbhg'
 
 sleep 60 # Sleep for 60 seconds to make sure the elasticsearch reindexing from UI finishes
+
+echo "✔running RDF reindexing"
+curl --location --request POST 'http://localhost:8585/api/v1/apps/trigger/RdfIndexApp' \
+--header 'Authorization: Bearer eyJraWQiOiJHYjM4OWEtOWY3Ni1nZGpzLWE5MmotMDI0MmJrOTQzNTYiLCJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiIsImlzQm90IjpmYWxzZSwiaXNzIjoib3Blbi1tZXRhZGF0YS5vcmciLCJpYXQiOjE2NjM5Mzg0NjIsImVtYWlsIjoiYWRtaW5Ab3Blbm1ldGFkYXRhLm9yZyJ9.tS8um_5DKu7HgzGBzS1VTA5uUjKWOCU0B_j08WXBiEC0mr0zNREkqVfwFDD-d24HlNEbrqioLsBuFRiwIWKc1m_ZlVQbG7P36RUxhuv2vbSp80FKyNM-Tj93FDzq91jsyNmsQhyNv_fNr3TXfzzSPjHt8Go0FMMP66weoKMgW2PbXlhVKwEuXUHyakLLzewm9UMeQaEiRzhiTMU3UkLXcKbYEJJvfNFcLwSl9W8JCO_l0Yj3ud-qt_nQYEZwqW6u5nfdQllN133iikV4fM5QZsMCnm8Rq1mvLR0y9bmJiD7fwM1tmJ791TUWqmKaTnP49U493VanKpUAfzIiOiIbhg' \
+--header 'Content-Type: application/json' \
+--data-raw '{
+    "entities": ["all"],
+    "recreateIndex": true,
+    "batchSize": 100,
+    "useDistributedIndexing": true,
+    "partitionSize": 10000
+}'
+
+sleep 30
 tput setaf 2
 echo "✔ OpenMetadata is up and running"
-
+echo "✔ RDF/Knowledge Graph support is enabled"
+echo "  - Fuseki UI: http://localhost:3030"
+echo "  - SPARQL endpoint: http://localhost:3030/openmetadata/sparql"