Method mergeDelta

engine/seminaivebottomup.go:419–492

mergeDelta updates e.store with facts from e.deltaStore. For facts with custom lattice join operations, replaces facts instead of adding.

()

// mergeDelta updates e.store with facts from e.deltaStore.
// For facts with custom lattice join operations, replaces facts instead of adding.
func (e *engine) mergeDelta() error {
	err := factstore.GetAllFacts(e.deltaStore, func(fact ast.Atom) error {
		pred := fact.Predicate
		fundep, mergePred, ok := e.hasMergePredicate(pred)
		if !ok {
			// Default case: just add the new fact.
			e.store.Add(fact)
			return nil
		}

		// Merge-predicate case: add fact or replace existing fact.
		// TODO: Generalize to merge predicate with n * 3 columns.
		if len(fundep.Target) != 1 {
			return fmt.Errorf("merging with |target vars| != 1 not implemented: %v", fundep.Target)
		}
		targetColumn := fundep.Target[0]

		// Query existing facts whose columns agree on fundep.Source values.
		queryArgs := make([]ast.BaseTerm, pred.Arity, pred.Arity)
		for i := 0; i < pred.Arity; i++ {
			queryArgs[i] = ast.Variable{"_"}
		}
		for i := range fundep.Source {
			queryArgs[i] = fact.Args[i]
		}
		queryExisting := ast.Atom{pred, queryArgs}
		existing := false
		e.store.GetFacts(queryExisting, func(existingFact ast.Atom) error {
			existing = true
			if fact.Equals(existingFact) {
				return nil // nothing to do.
			}

			// Evaluate merge predicate (top-down) to construct replacement fact.
			merged := false

			// Prepare top-down query with merge predicate.
			mergeQuery := ast.Atom{Predicate: mergePred, Args: []ast.BaseTerm{
				existingFact.Args[targetColumn],

				fact.Args[targetColumn],
				ast.Variable{"_"},
			}}
			err := e.newContext().EvalQuery(mergeQuery, mergePredMode, unionfind.New(), func(mergeFact ast.Atom) error {
				merged = true
				value := mergeFact.Args[2]
				fact.Args[fundep.Target[0]] = value
				if !existingFact.Equals(fact) {
					if storeWithRemove, ok := e.store.(factstore.FactStoreWithRemove); ok {
						storeWithRemove.Remove(existingFact)
					}
					e.store.Add(fact)
					return errBreak
				}
				return nil
			})
			if err == errBreak {
				return nil // Already added.