SPARQL查询中无法简化为属性路径的重复模式处理咨询

纯SPARQL递归查询方案 #

如果你的SPARQL端点支持SPARQL 1.1递归查询（主流端点如Blazegraph、Virtuoso均支持），可以用WITH RECURSIVE替代重复的UNION块，自动处理任意长度的Biosample到ProcessedSample转换路径，同时保留每一步的流程类型：

PREFIX nmdc: <https://w3id.org/nmdc/>
PREFIX dcterms: <http://purl.org/dc/terms/>
PREFIX skos: <http://www.w3.org/2004/02/skos/core#>

WITH RECURSIVE
  path(?start, ?end, ?process_types) AS (
    # 基础情况：直接一步转换
    {
      ?process a ?pt ;
               nmdc:has_input ?start ;
               nmdc:has_output ?end .
      BIND(?pt AS ?process_types)
    }
    UNION
    # 递归情况：多步转换，拼接流程类型
    {
      ?process a ?pt ;
               nmdc:has_input ?start ;
               nmdc:has_output ?mid .
      path(?mid, ?end, ?rest_types)
      BIND(CONCAT(STR(?pt), "|", ?rest_types) AS ?process_types)
    }
  )
SELECT DISTINCT ?Biosample ?process_types ?ProcessedSample
WHERE {
  ?Biosample a nmdc:Biosample .
  ?ProcessedSample a nmdc:ProcessedSample .
  path(?Biosample, ?ProcessedSample, ?process_types)
}

若需将流程类型拆分为单独列（如?process_type_1、?process_type_2），可根据预期最大路径长度用SPLIT和BIND提取，但递归查询的核心优势是无需预先定义路径长度上限。

用Python结合SPARQL Burger生成动态UNION查询 #

如果必须保留原有的多列结构（每个流程类型单独一列），可以用SPARQL Burger的Python库动态生成对应长度的UNION块，避免手动编写重复代码：

from sparql_burger import SparqlQueryBuilder, Join, Union, Triple, Variable, IRI

# 定义前缀
prefixes = {
    "nmdc": "https://w3id.org/nmdc/",
    "dcterms": "http://purl.org/dc/terms/",
    "skos": "http://www.w3.org/2004/02/skos/core#"
}

# 定义基础三元组
base_triples = [
    Triple(Variable("ProcessedSample"), IRI("a"), IRI("nmdc:ProcessedSample")),
    Triple(Variable("Biosample"), IRI("a"), IRI("nmdc:Biosample"))
]

# 生成UNION块，设置最大路径长度为4（可按需调整）
max_steps = 4
union_blocks = []

for step_count in range(1, max_steps + 1):
    triples = []
    if step_count == 1:
        # 一步转换逻辑
        triples.append(Triple(Variable("Process1"), IRI("a"), Variable("process_type_1")))
        triples.append(Triple(Variable("Process1"), IRI("nmdc:has_input"), Variable("Biosample")))
        triples.append(Triple(Variable("Process1"), IRI("nmdc:has_output"), Variable("ProcessedSample")))
    else:
        # 多步转换逻辑
        prev_output = Variable("Biosample")
        for i in range(1, step_count + 1):
            process_var = Variable(f"Process{i}")
            type_var = Variable(f"process_type_{i}")
            triples.append(Triple(process_var, IRI("a"), type_var))
            triples.append(Triple(process_var, IRI("nmdc:has_input"), prev_output))
            
            if i == step_count:
                triples.append(Triple(process_var, IRI("nmdc:has_output"), Variable("ProcessedSample")))
            else:
                mid_var = Variable(f"intermediate_{i}")
                triples.append(Triple(process_var, IRI("nmdc:has_output"), mid_var))
                prev_output = mid_var
    union_blocks.append(Join(triples))

# 构建并生成查询
query_builder = SparqlQueryBuilder(
    prefixes=prefixes,
    select=[Variable("Biosample")] + [Variable(f"process_type_{i}") for i in range(1, max_steps + 1)] + [Variable("ProcessedSample")],
    distinct=True,
    where=Union([Join(base_triples)] + union_blocks)
)

sparql_query = query_builder.build()
print(sparql_query)

这段代码会根据max_steps参数自动生成对应数量的UNION块，结构与手动编写的查询完全一致，调整max_steps即可快速扩展路径长度。

内容的提问来源于stack exchange，提问作者Mark Miller