use std::collections::HashMap;
use serde_json::{json, Value};
use crate::datasource::{DataSource, DataSourceConfig};
use crate::pipeline::RelAst;
use crate::schema::{Location, Registry};
pub const MAX_LINK_COMBINATIONS: usize = 4096;
pub(super) type Candidates = HashMap<String, Vec<Location>>;
type Graph = (Vec<String>, Vec<(String, String)>);
pub(super) fn compatible(a: &Location, b: &Location, ds_cfg: &DataSourceConfig) -> bool {
if a.source != b.source {
return false;
}
let ka = ds_cfg.resolve(Some(&a.source)).ok();
let kb = ds_cfg.resolve(Some(&b.source)).ok();
match (ka, kb) {
(Some(DataSource::Sql(_)), Some(DataSource::Sql(_))) => true,
_ => a.database == b.database && a.schema == b.schema,
}
}
pub(super) fn collect_graph(
root: &str,
root_relations: &[(String, RelAst)],
registry: &Registry,
) -> Result<Graph, String> {
fn push_unique(v: &mut Vec<String>, name: &str) {
if !v.iter().any(|x| x == name) {
v.push(name.to_string());
}
}
fn dfs(
parent: &str,
relations: &[(String, RelAst)],
registry: &Registry,
schemas: &mut Vec<String>,
edges: &mut Vec<(String, String)>,
) -> Result<(), String> {
let parent_schema = registry.get(parent)?;
for (name, rel_ast) in relations {
let Some(rel_def) = parent_schema.relations.get(name) else {
continue;
};
if rel_def.model.is_empty() {
continue;
}
let child = rel_def.model.clone();
push_unique(schemas, &child);
edges.push((parent.to_string(), child.clone()));
dfs(&child, &rel_ast.relations, registry, schemas, edges)?;
}
Ok(())
}
let mut schemas: Vec<String> = Vec::new();
let mut edges: Vec<(String, String)> = Vec::new();
push_unique(&mut schemas, root);
dfs(root, root_relations, registry, &mut schemas, &mut edges)?;
Ok((schemas, edges))
}
pub(super) fn candidates_of(registry: &Registry, name: &str) -> Result<Vec<Location>, String> {
let links = registry.links_of(name)?;
if links.is_empty() {
Ok(vec![registry.primary_location(name)?.clone()])
} else {
Ok(links.to_vec())
}
}
pub(super) fn filter_reachable(cands: &mut Candidates, reachable: Option<&[String]>) {
let Some(allowed) = reachable else {
return;
};
for locs in cands.values_mut() {
locs.retain(|l| allowed.iter().any(|s| s == &l.source));
}
}
#[derive(Debug, Clone)]
pub(super) struct Selection {
pub(super) locs: HashMap<String, Location>,
#[allow(dead_code)] pub(super) groups: usize,
#[allow(dead_code)] pub(super) internal_edges: usize,
pub(super) approximate: bool,
}
struct Dsu {
parent: Vec<usize>,
}
impl Dsu {
fn new(n: usize) -> Self {
Self {
parent: (0..n).collect(),
}
}
fn find(&mut self, x: usize) -> usize {
let mut root = x;
while self.parent[root] != root {
root = self.parent[root];
}
let mut cur = x;
while self.parent[cur] != root {
let next = self.parent[cur];
self.parent[cur] = root;
cur = next;
}
root
}
fn union(&mut self, a: usize, b: usize) {
let ra = self.find(a);
let rb = self.find(b);
if ra != rb {
self.parent[ra] = rb;
}
}
fn components(&mut self) -> usize {
(0..self.parent.len())
.filter(|&i| self.find(i) == i)
.count()
}
}
fn evaluate(
chosen: &[Location],
idx: &HashMap<&str, usize>,
edges: &[(String, String)],
ds_cfg: &DataSourceConfig,
) -> (usize, usize) {
let mut dsu = Dsu::new(chosen.len());
let mut internal = 0usize;
for (u, v) in edges {
let (iu, iv) = match (idx.get(u.as_str()), idx.get(v.as_str())) {
(Some(&a), Some(&b)) => (a, b),
_ => continue,
};
if compatible(&chosen[iu], &chosen[iv], ds_cfg) {
dsu.union(iu, iv);
internal += 1;
}
}
(dsu.components(), internal)
}
type CanonEntry = (String, String, Option<String>, Option<String>);
fn canon_key(schemas: &[String], chosen: &[Location]) -> Vec<CanonEntry> {
let mut v: Vec<CanonEntry> = schemas
.iter()
.zip(chosen.iter())
.map(|(s, l)| {
(
s.clone(),
l.source.clone(),
l.database.clone(),
l.schema.clone(),
)
})
.collect();
v.sort();
v
}
fn better(
new: &(Vec<Location>, usize, usize),
best: &Option<(Vec<Location>, usize, usize)>,
schemas: &[String],
) -> bool {
match best {
None => true,
Some(b) => {
if new.1 != b.1 {
return new.1 < b.1;
}
if new.2 != b.2 {
return new.2 > b.2;
}
canon_key(schemas, &new.0) < canon_key(schemas, &b.0)
}
}
}
pub(super) fn select_links(
schemas: &[String],
edges: &[(String, String)],
cands: &Candidates,
ds_cfg: &DataSourceConfig,
) -> Result<Selection, String> {
for s in schemas {
if cands.get(s).map(|v| v.is_empty()).unwrap_or(true) {
return Err(format!(
"ERR_FEDERATION_NO_REACHABLE_LINK: {} 无可用落点(候选链路被上下文/权限过滤后为空)",
s
));
}
}
let combos: u128 = schemas.iter().map(|s| cands[s].len() as u128).product();
if combos > MAX_LINK_COMBINATIONS as u128 {
return Ok(select_links_approx(schemas, edges, cands, ds_cfg));
}
let idx: HashMap<&str, usize> = schemas
.iter()
.enumerate()
.map(|(i, s)| (s.as_str(), i))
.collect();
let n = schemas.len();
let mut indices = vec![0usize; n];
let mut best: Option<(Vec<Location>, usize, usize)> = None;
loop {
let chosen: Vec<Location> = (0..n)
.map(|i| cands[&schemas[i]][indices[i]].clone())
.collect();
let (groups, internal) = evaluate(&chosen, &idx, edges, ds_cfg);
let cand = (chosen, groups, internal);
if better(&cand, &best, schemas) {
best = Some(cand);
}
let mut k = 0usize;
while k < n {
indices[k] += 1;
if indices[k] < cands[&schemas[k]].len() {
break;
}
indices[k] = 0;
k += 1;
}
if k >= n {
break;
}
}
let (locs, groups, internal_edges) = best.expect("非空候选笛卡尔积至少有一个组合");
Ok(Selection {
locs: schemas.iter().cloned().zip(locs).collect(),
groups,
internal_edges,
approximate: false,
})
}
fn select_links_approx(
schemas: &[String],
edges: &[(String, String)],
cands: &Candidates,
ds_cfg: &DataSourceConfig,
) -> Selection {
let idx: HashMap<&str, usize> = schemas
.iter()
.enumerate()
.map(|(i, s)| (s.as_str(), i))
.collect();
let n = schemas.len();
let base: Vec<Location> = (0..n).map(|i| cands[&schemas[i]][0].clone()).collect();
let (g, e) = evaluate(&base, &idx, edges, ds_cfg);
let mut best: (Vec<Location>, usize, usize) = (base.clone(), g, e);
for i in 0..n {
for alt in cands[&schemas[i]].iter().skip(1) {
let mut trial = base.clone();
trial[i] = alt.clone();
let (g, e) = evaluate(&trial, &idx, edges, ds_cfg);
let cand = (trial, g, e);
if better(&cand, &Some(best.clone()), schemas) {
best = cand;
}
}
}
Selection {
locs: schemas.iter().cloned().zip(best.0).collect(),
groups: best.1,
internal_edges: best.2,
approximate: true,
}
}
pub(super) fn degraded_approx() -> Value {
json!({
"code": "linkSelectionApprox",
"layer": "federation",
"message": format!(
"涉及 schema 数 × 候选链路数超过 {},落点择优走确定性近似",
MAX_LINK_COMBINATIONS
),
"hint": "收敛本次查询涉及的表/链路数,以恢复精确择优",
})
}