55 lines
2.4 KiB
Python
55 lines
2.4 KiB
Python
from pathlib import Path
|
|
|
|
from tht.cli.schema_cmd import physical_path
|
|
|
|
|
|
def _extract_lsh_values(dwh, physical, annotations, limit):
|
|
from tht.db.sampling import SkippedColumn, TruncatedColumn, is_text_type
|
|
from tht.mschema.eligibility import effective_eligibility
|
|
|
|
values, skipped, truncated = {}, [], []
|
|
for table_name, table in physical.tables.items():
|
|
table_ann = annotations.tables.get(table_name)
|
|
for column_name, column in table.columns.items():
|
|
ann_col = table_ann.columns.get(column_name) if table_ann else None
|
|
if not is_text_type(column.type) or not effective_eligibility(column, ann_col)[0]:
|
|
continue
|
|
try:
|
|
distinct = dwh.distinct_values(table_name, column_name, limit=limit)
|
|
except Exception as exc:
|
|
skipped.append(SkippedColumn(table_name, column_name, f"errore: {exc}"))
|
|
continue
|
|
vals = [str(value) for value in distinct.values if value not in (None, "")]
|
|
if vals:
|
|
values.setdefault(table_name, {})[column_name] = vals
|
|
if distinct.truncated:
|
|
truncated.append(TruncatedColumn(table_name, column_name, len(vals)))
|
|
return values, skipped, truncated
|
|
|
|
|
|
def build_lsh_artifacts(
|
|
cfg, *, dwh=None, verbose: bool = False, physical_file: Path | None = None,
|
|
output_dir: Path | None = None,
|
|
):
|
|
"""Run the existing LSH extraction/build algorithm and persist its outputs."""
|
|
from tht.adapters.factory import build_dwh
|
|
from tht.cli.schema_cmd import annotations_path
|
|
from tht.lshindex import build_index, save_index
|
|
from tht.mschema.models import Annotations, PhysicalSchema
|
|
|
|
phys_file = physical_file or physical_path(cfg)
|
|
if not phys_file.exists():
|
|
raise FileNotFoundError("physical catalog is missing; run schema introspect first")
|
|
physical = PhysicalSchema.from_yaml(phys_file)
|
|
annotations = Annotations.from_yaml(annotations_path(cfg))
|
|
target = dwh if dwh is not None else build_dwh(cfg)
|
|
values, skipped, truncated = _extract_lsh_values(
|
|
target, physical, annotations, cfg.lsh.max_values_per_column
|
|
)
|
|
lsh, minhashes = build_index(values, cfg.lsh, verbose=verbose)
|
|
save_index(
|
|
lsh, minhashes, cfg.lsh, output_dir or (cfg.paths.indexes / "lsh"),
|
|
name=cfg.database.db_schema,
|
|
)
|
|
return minhashes, skipped, truncated, values
|