Delex (DAG multi-strategy blocking)¶

DAG-based multi-strategy blocking. madmatcher_pro.delex has an empty __init__; import these symbols from their submodules, as shown below. You build a BlockingProgram from rules and predicates and run it with PlanExecutor. The advanced base classes at the bottom are only needed to write a custom predicate or rule.

Running a blocking program¶

class madmatcher_pro.delex.execution.plan_executor.PlanExecutor(*, index_table, search_table, build_parallelism=4, index_table_id_col='_id', ram_size_in_bytes=None, cost_est=None, optimize, estimate_cost, index_sample_sizes=(25000, 50000, 100000), filter_sample_size=10000, sample_seed=None)¶

Bases: GraphExecutor

Compiles a delex BlockingProgram into a plan and runs it over Spark.

Construct with the index_table / search_table (from GraphExecutor) plus optimize and estimate_cost flags, then call execute() with the compiled program to produce a candidate table (id2, id1_list, ...) – the shared blocking-output contract. execute also exposes the opt-in crash-recovery and exclude_self (dedup) flags. The cost-estimation sampling knobs (index_sample_sizes / filter_sample_size / sample_seed) tune planning only and are used when estimate_cost is set.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

index_table (DataFrame)
search_table (DataFrame)
build_parallelism (Annotated[int, Gt(gt=0)])
index_table_id_col (str)
ram_size_in_bytes (Annotated[int, Gt(gt=0)] | None)
cost_est (CostEstimator | None)
optimize (bool)
estimate_cost (bool)
index_sample_sizes (tuple[int, ...])
filter_sample_size (int)
sample_seed (int | None)

execute(prog, search_table_id_col, projection=None, chunk_size=2000, *, enable_crash_recovery=False, checkpoint_dir=None, n_groups=10, rows_per_group=None, show_progress=False, validate_input=True, dashboard=None, dashboard_port=None, open_browser=None, exclude_self=False)¶

Compile prog into a plan and execute it.

Parameters:

prog (BlockingProgram) – the blocking program to compile into a plan and run
search_table_id_col (str) – the id column of the search table
projection (list of str, optional) – search-table columns to carry into the output alongside the candidate lists; None (default) keeps only the search id column
chunk_size (int, default 2000) – target number of records per partition when repartitioning the tables for execution

Returns:

The second return value differs by mode. Default (enable_crash_recovery=False) returns (candidate_df, PlanExecutionStats) with exec/optimize/cost times and total_time. enable_crash_recovery=True returns (union_df, ResumableBlockingStats), a different stats type (n_groups, groups_committed, groups_run_this_session, optimize/cost times, checkpoint_dir) with no graph_exec_stats or total_time, since a resumable run spans groups and processes.

Return type:

tuple

Notes

enable_crash_recovery (keyword-only, default off) makes the blocking resumable: search records are split into n_groups hash-groups, each run through the full plan and committed with an atomic marker; a re-run with the same checkpoint_dir skips committed groups. It then requires checkpoint_dir. rows_per_group, if set, derives n_groups = ceil(num_records / rows). validate_input (default True) records the search row count and refuses a resume whose count changed (count-only: a same-count change isn’t caught); pass False only for a known-immutable input.

dashboard / dashboard_port / open_browser default to None, meaning “defer to the active MadMatcherSession, else the env/default (on)”; pass an explicit value to override. The global opt-out MADMATCHER_DASHBOARD=0 turns tracking off regardless. Any failure to start the dashboard is logged and swallowed, so it never blocks the blocking job.

Blocking programs and rules¶

class madmatcher_pro.delex.lang.program.BlockingProgram(keep_rules, drop_rules)¶

Bases: object

A blocking program: a set of keep rules (unioned) minus a set of drop rules.

A candidate pair is kept if it satisfies ANY keep rule, unless it also satisfies a drop rule. Compile and run it with execute():

from madmatcher_pro.delex.lang import BlockingProgram, KeepRule
from madmatcher_pro.delex.lang.predicate import BM25TopkPredicate

prog = BlockingProgram(
    keep_rules=[
        KeepRule([BM25TopkPredicate("name", "name", "standard", 20)]),
        KeepRule([BM25TopkPredicate("description", "description", "standard", 20)]),
    ],
    drop_rules=[],
)

Parameters:

keep_rules (list of KeepRule) – rules whose outputs are unioned; must contain at least one keep rule
drop_rules (list of DropRule) – rules whose matches are removed from the union (may be empty)

validate()¶: Validate the program. A blocking program must contain at least one keep rule; raises ValueError if it does not. Called automatically after construction.

pretty_str()¶

create a pretty string of the entire blocking program

Return type:: str

class madmatcher_pro.delex.lang.rule.KeepRule(predicates)¶

Bases: Rule

A conjunctive blocking rule: keep candidate pairs that satisfy all its predicates.

A BlockingProgram is a set of KeepRule s whose outputs are unioned. Each rule ANDs its predicates (e.g. a BM25 top-k on name AND an exact match on zip), and must contain at least one indexable predicate (checked by validate()) so the rule can be driven from an index rather than a full cross product.

Parameters:: predicates (list of Predicate) – the predicates ANDed together; must include at least one indexable predicate

validate()¶: check that this rule has at least one indexable predicate if not raise RuntimeError

class madmatcher_pro.delex.lang.rule.DropRule(predicates)¶

Bases: Rule

A drop rule: remove candidate pairs that satisfy all its predicates.

Parameters:: predicates (list of Predicate) – the predicates ANDed together; all must be streamable

validate()¶: check that all the predicates in this rule are streamable if not raise RuntimeError

Predicates¶

The concrete predicates a blocking rule is built from. Import them from madmatcher_pro.delex.lang.predicate.

class madmatcher_pro.delex.lang.predicate.BM25TopkPredicate(index_col, search_col, tokenizer, k)¶

Bases: Predicate

Parameters:

index_col (str) – the column in the indexed table that the BM25 index is built on
search_col (str) – the column in the search table used to query the index
tokenizer (str) – name of the tokenizer that splits each string into terms
k (int, > 0) – number of top-scoring candidates kept per search record

class madmatcher_pro.delex.lang.predicate.ExactMatchPredicate(index_col, search_col, invert, lowercase=False)¶

Bases: ThresholdPredicate

an exact match predicate, i.e. if x == y return 1.0 else 0.0

index_colstr: the column to be indexed
search_colstr: the column that will be used for search
invertbool: change predicate from index_col == search_col to index_col != search_col
lowercasebool: lowercase the strings before comparing them

Parameters:

index_col (str)
search_col (str)
invert (bool)
lowercase (bool)

class madmatcher_pro.delex.lang.predicate.JaccardPredicate(index_col, search_col, tokenizer, op, val)¶

Bases: SetSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
tokenizer (Tokenizer) – tokenizer that splits each string into the token set the similarity is computed over
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.CosinePredicate(index_col, search_col, tokenizer, op, val)¶

Bases: SetSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
tokenizer (Tokenizer) – tokenizer that splits each string into the token set the similarity is computed over
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.OverlapCoeffPredicate(index_col, search_col, tokenizer, op, val)¶

Bases: SetSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
tokenizer (Tokenizer) – tokenizer that splits each string into the token set the similarity is computed over
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.EditDistancePredicate(index_col, search_col, op, val)¶

Bases: StringSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.JaroPredicate(index_col, search_col, op, val)¶

Bases: StringSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.JaroWinklerPredicate(index_col, search_col, op, val, prefix_weight=0.1)¶

Bases: StringSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when jaro_winkler(index_col, search_col) op val holds
val (float) – the similarity threshold compared against
prefix_weight (float, default 0.1) – weight of the common-prefix boost in the Jaro-Winkler score

class madmatcher_pro.delex.lang.predicate.SmithWatermanPredicate(index_col, search_col, op, val, gap_cost=1.0)¶

Bases: StringSimPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when smith_waterman(index_col, search_col) op val holds
val (float) – the similarity threshold compared against
gap_cost (float, default 1.0) – the gap cost used by the Smith-Waterman alignment

Table validation¶

madmatcher_pro.delex.utils.checks.check_tables(table_a, id_col_table_a, table_b, id_col_table_b)¶

Check that table_a and table_b have valid id columns.

Parameters:

table_a (sql.DataFrame) – The table A to be indexed.
id_col_table_a (str) – The column name of the id column in table A.
table_b (sql.DataFrame) – The table B to be searched.
id_col_table_b (str) – The column name of the id column in table B.

Raises:

ValueError – If table_a or table_b do not have a valid id column.

Advanced: extension base classes¶

Subclass these only to write a custom predicate or rule. Most users compose the concrete predicates above and do not need them.

class madmatcher_pro.delex.lang.predicate.Predicate¶

Bases: ABC

abstract base class for all Predicates to be used in writing blocking programs

abstract property streamable¶: True if the predicate can be evaluated over a single partition of the indexed table, otherwise False

abstract property indexable¶: True if the predicate can be efficiently indexed

abstract property sim¶: The simiarlity used by the predicate

abstract property is_topk: bool¶: True if the self is Topk based, else False

abstractmethod build(for_search, index_table, index_id_col='_id', cache=None)¶

build the Predicate over index_table using index_id_col as a unique id, optionally using cache to get or set the index

Parameters:

for_search (bool) – build the predicate for searching, otherwise streaming / filtering
index_table (pyspark.sql.DataFrame) – the dataframe that will be preprocessed / indexed
index_id_col (str) – the name of the unique id column in index_table
cache (Optional[BuildCache] = None) – the cache for built indexes and hash tables

abstractmethod contains(other)¶

True if the set output by self is a superset (non-strict) of other

Return type:: bool

abstractmethod search_batch(queries)¶

perform search with queries return a dataframe with schema (id1_list array<long>, scores array<float>, time float)

Parameters:: queries (Series)
Return type:: DataFrame

search(itr)¶

perform search_batch for each batch in itr

Parameters:: itr (Iterator[Series])
Return type:: Iterator[DataFrame]

abstractmethod filter_batch(queries, id1_lists)¶

filter each id_list in id1_lists using this predicate. This is, for each query, id_list pair in zip(queries, id1_lists), return only the ids which satisfy predicate(query, id) for id in id_list. Return a dataframe with schema (id1_list array<long>, scores array<float>, time float)

Parameters:

queries (Series)
id1_lists (Series)

Return type:

DataFrame

filter(itr)¶

perform filter_batch for each batch in itr

Parameters:: itr (Iterator[Tuple[Series, Series]])
Return type:: Iterator[DataFrame]

abstractmethod init()¶: initialize the predicate for searching or filtering

abstractmethod deinit()¶: release the resources acquired by self.init()

abstractmethod index_size_in_bytes()¶

return the total size in bytes of all the files associated with this predicate

Return type:: int

abstractmethod index_component_sizes(for_search)¶

return a dictionary of file sizes for each data structure used by this predicate, if the predicate hasn’t been built yet, the sizes are None

Parameters:: for_search (bool) – return the sizes for searching or for filtering
Return type:: dict[Any, int | None]

class madmatcher_pro.delex.lang.predicate.ThresholdPredicate(index_col, search_col, op, val)¶

Bases: Predicate, ABC

Parameters:: val (float)

class madmatcher_pro.delex.lang.predicate.SetSimPredicate(index_col, search_col, tokenizer, op, val)¶

Bases: ThresholdPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
tokenizer (Tokenizer) – tokenizer that splits each string into the token set the similarity is computed over
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.predicate.StringSimPredicate(index_col, search_col, op, val)¶

Bases: ThresholdPredicate

Parameters:

index_col (str) – the column in the indexed table to compare
search_col (str) – the column in the search table to compare against
op (operator) – comparison operator from the operator module (ge, gt, le, lt); a pair is kept when similarity(index_col, search_col) op val holds
val (float) – the similarity threshold compared against

class madmatcher_pro.delex.lang.rule.Rule(predicates)¶

Bases: object

Base class for a DropRule or KeepRule.

Parameters:: predicates (list of Predicate) – the predicates ANDed together to form the rule (a pair satisfies the rule only if it satisfies every predicate)

contains(other)¶: return True if self logically contains other else False. That is, for any given set of record pairs C, self(C) is a superset of other(C)

pretty_str()¶

format the rule into a pretty string

Return type:: str