diff --git a/bin/main.py b/bin/main.py
index c3ea8e8c..9d98ae22 100644
--- a/bin/main.py
+++ b/bin/main.py
@@ -105,10 +105,9 @@ def main(_):
     case _:
       raise ValueError(f'Unknown mechanism: {_MECHANISM.value}')
 
-  mechanism = dpsynth.TabularSynthesizer(
-      domains=attribute_domains,
+  mechanism = dpsynth.TabularConfig(
       discrete_mechanism=mechanism_config,
-  ).calibrate(epsilon=_EPSILON.value, delta=_DELTA.value)
+  ).calibrate(attribute_domains, epsilon=_EPSILON.value, delta=_DELTA.value)
   result = mechanism(np.random.default_rng(_SEED.value), df)  # pyrefly: ignore[bad-argument-type]
   result.synthetic_data.to_csv(_OUTPUT_PATH.value, index=False)
 
diff --git a/docs/in_memory_api.md b/docs/in_memory_api.md
index 5a16d3e1..efcad728 100644
--- a/docs/in_memory_api.md
+++ b/docs/in_memory_api.md
@@ -14,8 +14,9 @@ within a single machine's RAM.
 ## Python API: `dpsynth.TabularConfig`
 
 The primary entry point for in-memory synthesis is
-`dpsynth.TabularConfig`. It accepts a dictionary of attribute domains and
-mechanism options, is calibrated with a privacy budget to produce a
+`dpsynth.TabularConfig`. It configures the algorithm hyperparameters (e.g.
+discrete mechanism, numerical bin count, budget allocation), is calibrated
+with a dataset `dpsynth.Schema` and privacy budget to produce a
 `dpsynth.TabularMechanism`, and generates a fully synthetic, differentially
 private DataFrame matching the exact schema and data types of your input.
 
@@ -24,14 +25,24 @@ private DataFrame matching the exact schema and data types of your input.
 ```python
 import dpsynth
 from dpsynth import discrete_mechanisms
+from dpsynth import domain
 import numpy as np
 import pandas as pd
 
+# Define schema (or load from domain.from_yaml_file)
+schema = dpsynth.Schema({
+    "age": domain.NumericalAttribute(min_value=18, max_value=90),
+    "workclass": domain.CategoricalAttribute(possible_values=["Private", "Gov", "Other"]),
+})
+
+# Reusable algorithm preset
 config = dpsynth.TabularConfig(
-    domains=domains,
     discrete_mechanism=discrete_mechanisms.MSTConfig(),
+    numerical_bins=32,
 )
-mechanism = config.calibrate(epsilon=1.0, delta=1e-6)
+
+# Calibrate with schema and privacy budget
+mechanism = config.calibrate(schema, epsilon=1.0, delta=1e-6)
 result = mechanism(np.random.default_rng(), sensitive_df)
 synthetic_df = result.synthetic_data
 ```
@@ -40,22 +51,19 @@ synthetic_df = result.synthetic_data
 
 When initializing `dpsynth.TabularConfig`:
 
-*   `domains`: Mapping of column names to domain specifications
-    ([`CategoricalAttribute`, `NumericalAttribute`, or `OpenSetCategoricalAttribute`](data_and_terminology.md)).
-    Every key must exist in `data.columns`.
 *   `discrete_mechanism`: Configuration object specifying which DP synthesis
-    mechanism to run (e.g., `MSTConfig()`, `AIMConfig()`,
+    mechanism to run (e.g., `MSTConfig()`, `AIMConfig()`, `SWIFTConfig()`,
     `IndependentConfig()`).
 *   `numerical_bins`: Number of equal-frequency quantile buckets used to
     discretize continuous numerical columns (default: `32`).
 *   `init_budget_fraction`: Fraction of total `(epsilon, delta)` budget
     allocated for per-column initialization such as bounds computation and
     partition selection (default: `0.1`).
-*   `cross_attribute_constraints`: Optional sequence of constraints to enforce
-    on generated data.
 
-When calling `config.calibrate(...)`:
+When calling `config.calibrate(schema, ...)` or `config.configure(schema, ...)`:
 
+*   `schema`: The `dpsynth.Schema` (or mapping of column names to attribute
+    domains) defining the dataset columns and optional constraints.
 *   `epsilon`, `delta`: Total differential privacy budget parameters. Returns a
     runnable `TabularMechanism`.
 
@@ -64,7 +72,7 @@ When calling `config.calibrate(...)`:
 ## Standalone End-to-End Python Example
 
 Here is a complete, self-contained Python script demonstrating how to specify a
-domain, set up a `TabularConfig`, calibrate the mechanism with a privacy budget,
+schema, set up a `TabularConfig`, calibrate the mechanism with a privacy budget,
 load sensitive data, synthesize records, and print the first few rows.
 
 ```python
@@ -74,26 +82,25 @@ from dpsynth import domain
 import numpy as np
 import pandas as pd
 
-# 1. Domain Specification: Define the schema of the tabular dataset
-attribute_domains = {
-    "age": domain.NumericalAttribute(lower_bound=18, upper_bound=90),
+# 1. Schema Specification: Define the schema of the tabular dataset
+schema = dpsynth.Schema({
+    "age": domain.NumericalAttribute(min_value=18, max_value=90),
     "workclass": domain.CategoricalAttribute(
-        allowed_values=["Private", "Self-emp", "Gov", "Other"]
+        possible_values=["Private", "Self-emp", "Gov", "Other"]
     ),
     "education": domain.CategoricalAttribute(
-        allowed_values=["HS-grad", "Bachelors", "Masters", "PhD"]
+        possible_values=["HS-grad", "Bachelors", "Masters", "PhD"]
     ),
-}
+})
 
-# 2. Setup Config: Configure synthesizer with domain and mechanism choices
+# 2. Setup Config: Configure synthesizer hyperparameter preset
 config = dpsynth.TabularConfig(
-    domains=attribute_domains,
     discrete_mechanism=discrete_mechanisms.MSTConfig(),
     numerical_bins=16,
 )
 
-# 3. Calibrate Mechanism: Allocate privacy budget to get runnable mechanism
-mechanism = config.calibrate(epsilon=1.0, delta=1e-5)
+# 3. Calibrate Mechanism: Allocate privacy budget with schema to get runnable mechanism
+mechanism = config.calibrate(schema, epsilon=1.0, delta=1e-5)
 
 # 4. Load Data: Create sensitive input DataFrame matching the domain schema
 sensitive_df = pd.DataFrame({
diff --git a/dpsynth/__init__.py b/dpsynth/__init__.py
index dac7c8d2..5e2531a9 100644
--- a/dpsynth/__init__.py
+++ b/dpsynth/__init__.py
@@ -21,6 +21,9 @@
 from dpsynth import discrete_mechanisms
 from dpsynth import domain
 from dpsynth import relational
+from dpsynth.api import CalibratedMechanism
+from dpsynth.api import MechanismConfig
+from dpsynth.constraints import Constraint
 from dpsynth.data_generation_v3 import TabularConfig
 from dpsynth.data_generation_v3 import TabularMechanism
 from dpsynth.data_generation_v3 import TabularSynthesizer
@@ -30,24 +33,35 @@
 from dpsynth.domain import FreeFormTextAttribute
 from dpsynth.domain import NumericalAttribute
 from dpsynth.domain import OpenSetCategoricalAttribute
+from dpsynth.domain import Schema
 
 ForeignKeyRelation = relational.ForeignKeyRelation
 MultiDataGenerationResult = relational.MultiDataGenerationResult
 MultiTableConfig = relational.MultiTableConfig
 MultiTableMechanism = relational.MultiTableMechanism
+RelationalSchema = relational.RelationalSchema
 
 __all__ = [
+    'CalibratedMechanism',
     'CategoricalAttribute',
+    'Constraint',
+    'DiscreteConfig',
+    'DiscreteMechanism',
     'ForeignKeyRelation',
+    'FreeFormTextAttribute',
+    'MechanismConfig',
     'MultiDataGenerationResult',
     'MultiTableConfig',
     'MultiTableMechanism',
     'NumericalAttribute',
     'OpenSetCategoricalAttribute',
+    'RelationalSchema',
+    'Schema',
     'TabularConfig',
     'TabularMechanism',
     'TabularSynthesizer',
     'api',
+    'constraints',
     'discrete_mechanisms',
     'domain',
     'relational',
diff --git a/dpsynth/adapters/beam.py b/dpsynth/adapters/beam.py
index 0e01a3aa..4fb0874c 100644
--- a/dpsynth/adapters/beam.py
+++ b/dpsynth/adapters/beam.py
@@ -24,7 +24,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 import dataclasses
 import io
 import math
@@ -74,12 +74,14 @@ def __init__(self, initializers: dict[str, Initializer]):
     for column, init in initializers.items():
       if isinstance(init, initialization.NumericalInitializerConfig):
         attr = init.attribute
-        lower, upper, gs = init.grid_spec
+        assert attr is not None
+        lower, upper, gs = init.grid_spec(attr)
         delta = (upper - lower) / (gs - 1)
         meta = dict(attribute=attr, lower=lower, upper=upper, delta=delta)
         self._specs.append((column, 'numerical', meta))
 
       elif isinstance(init, initialization.CategoricalInitializerConfig):
+        assert init.attribute is not None
         meta = {
             'lookup': init.attribute.lookup,
             'default': init.attribute.out_of_domain_index,
@@ -211,10 +213,10 @@ def run_from_summary(
   for column, init in initializers.items():
     sparse = sparse_stats[column]
     if isinstance(init, initialization.NumericalInitializer):
-      counts = _sparse_to_dense_numerical(sparse, init.config.grid_spec[2])
+      counts = _sparse_to_dense_numerical(sparse, init.grid_size)
       results[column] = init.from_summary(rng, counts)
     elif isinstance(init, initialization.CategoricalInitializer):
-      counts = _sparse_to_dense_categorical(sparse, init.config.attribute.size)
+      counts = _sparse_to_dense_categorical(sparse, init.attribute.size)
       results[column] = init.from_summary(rng, counts)
     elif isinstance(init, initialization.OpenSetInitializer):
       unique_values, value_counts = _sparse_to_openset(sparse)
@@ -530,13 +532,13 @@ class BeamTabularConfig(api.MechanismConfig):
 
   Usage::
 
-      config = data_generation_v3.TabularConfig(domains=domains)
-      beam_synth = BeamTabularConfig(config).configure(zcdp_rho=1.0)
+      config = data_generation_v3.TabularConfig()
+      beam_synth = BeamTabularConfig(config).configure(schema, zcdp_rho=1.0)
       result = beam_synth(rng, create_rows_fn)
 
   Attributes:
-    synthesizer: The wrapped local-mode TabularConfig. Supplies the domain,
-      sub-mechanisms, constraints, and privacy calibration.
+    synthesizer: The wrapped local-mode TabularConfig. Supplies the
+      sub-mechanisms and initialization parameters.
     temp_location: Directory used to shuttle small singleton results between the
       pipeline and the driver. Must be readable and writable by all workers --
       i.e. a shared distributed filesystem for distributed runners. Defaults to
@@ -544,7 +546,9 @@ class BeamTabularConfig(api.MechanismConfig):
     pipeline_options: Optional Beam pipeline options applied to both passes.
   """
 
-  synthesizer: data_generation_v3.TabularConfig
+  synthesizer: data_generation_v3.TabularConfig = dataclasses.field(
+      default_factory=data_generation_v3.TabularConfig
+  )
   temp_location: str | None = None
   pipeline_options: beam.options.pipeline_options.PipelineOptions | None = None
 
@@ -555,11 +559,25 @@ def __post_init__(self):
           ' method.'
       )
 
+  @property
+  def domains(
+      self,
+  ) -> domain.Schema | Mapping[str, domain.AttributeType] | None:
+    return self.synthesizer.domains
+
   def configure(
-      self, *, zcdp_rho, delta=0, max_records_per_user=1
+      self,
+      schema: domain.Schema | Mapping[str, domain.AttributeType] | None = None,
+      *,
+      zcdp_rho: float,
+      delta: float = 0.0,
+      max_records_per_user: int = 1,
   ) -> BeamTabularMechanism:
     """Returns a copy whose synthesizer is configured with the given budget."""
+    if schema is None:
+      schema = self.domains
     synthesizer = self.synthesizer.configure(
+        schema,
         zcdp_rho=zcdp_rho,
         delta=delta,
         max_records_per_user=max_records_per_user,
diff --git a/dpsynth/api.py b/dpsynth/api.py
index 70ce4ce3..79249a0f 100644
--- a/dpsynth/api.py
+++ b/dpsynth/api.py
@@ -119,7 +119,12 @@ class MechanismConfig(abc.ABC):
 
   @abc.abstractmethod
   def configure(
-      self, *, zcdp_rho, delta=0, max_records_per_user=1
+      self,
+      domain: Any = None,
+      *,
+      zcdp_rho: float,
+      delta: float = 0.0,
+      max_records_per_user: int = 1,
   ) -> CalibratedMechanism:
     """Returns a calibrated mechanism for the given zCDP budget.
 
@@ -133,6 +138,8 @@ def configure(
     not provided (i.e., is 0).
 
     Args:
+      domain: Mechanism-specific domain or schema specification (e.g.
+        ``domain.Schema``, ``domain.AttributeType``, or ``mbi.Domain``).
       zcdp_rho: The zCDP privacy budget (rho).
       delta: Approximate DP delta consumed by the mechanism itself (e.g., for
         thresholding). Defaults to 0 (pure zCDP). Mechanisms that need delta
@@ -203,6 +210,7 @@ def _find_optimal_rho(
 
   def calibrate(
       self,
+      domain: Any = None,
       *,
       epsilon: float | None = None,
       delta: float | None = None,
@@ -221,6 +229,7 @@ def calibrate(
       ``configure(zcdp_rho=...)`` directly instead.
 
     Args:
+      domain: Mechanism-specific domain or schema specification.
       epsilon: Target epsilon for (epsilon, delta)-DP.
       delta: Target delta for (epsilon, delta)-DP.
       zcdp_rho: Deprecated. Direct zCDP budget. Use ``configure()`` instead.
@@ -252,6 +261,7 @@ def calibrate(
           stacklevel=2,
       )
       return self.configure(
+          domain,
           zcdp_rho=zcdp_rho,
           max_records_per_user=max_records_per_user,
       )
@@ -261,6 +271,7 @@ def calibrate(
 
     def make_event_fn(rho: float) -> dp_accounting.DpEvent:
       base = self.configure(
+          domain,
           zcdp_rho=rho,
           delta=delta,
           max_records_per_user=max_records_per_user,
@@ -274,6 +285,7 @@ def make_event_fn(rho: float) -> dp_accounting.DpEvent:
         target_delta=delta,
     )
     return self.configure(
+        domain,
         zcdp_rho=optimal_rho,
         delta=delta,
         max_records_per_user=max_records_per_user,
diff --git a/dpsynth/data_generation_v2.py b/dpsynth/data_generation_v2.py
index 081276b0..db7cb24f 100644
--- a/dpsynth/data_generation_v2.py
+++ b/dpsynth/data_generation_v2.py
@@ -23,6 +23,7 @@
 from dpsynth import constraints
 from dpsynth import discrete_mechanisms
 from dpsynth import domain
+from dpsynth.data_generation_v3 import TabularConfig
 from dpsynth.data_generation_v3 import TabularSynthesizer
 import numpy as np
 import pandas as pd
@@ -51,14 +52,17 @@ def generate(
       stacklevel=2,
   )
   del skip_compression  # Not supported by TabularSynthesizer.
-  synth = TabularSynthesizer(
-      domains=domains,
+  config = TabularConfig(
       discrete_mechanism=discrete_config,  # pyrefly: ignore[bad-argument-type]
-      cross_attribute_constraints=cross_attribute_constraints,
       numerical_bins=numerical_bins,
       init_budget_fraction=one_way_marginal_budget_fraction,
   )
-  result = synth.calibrate(
+  schema = domain.Schema(
+      attributes=domains,
+      constraints=cross_attribute_constraints or (),
+  )
+  result = config.calibrate(
+      schema,
       epsilon=epsilon,
       delta=delta,
   )(np.random.default_rng(), data)
diff --git a/dpsynth/data_generation_v3.py b/dpsynth/data_generation_v3.py
index 43b0f548..12893f58 100644
--- a/dpsynth/data_generation_v3.py
+++ b/dpsynth/data_generation_v3.py
@@ -18,6 +18,7 @@
 
 from collections.abc import Mapping, Sequence
 import dataclasses
+from typing import Any
 import warnings
 
 from absl import logging
@@ -55,18 +56,15 @@ def create_initializers(
   for col, attr in domains.items():
     if isinstance(attr, domain.NumericalAttribute):
       initializers[col] = initialization.NumericalInitializerConfig(
-          name=col,
           num_partitions=numerical_bins,
           attribute=attr,
       )
     elif isinstance(attr, domain.CategoricalAttribute):
       initializers[col] = initialization.CategoricalInitializerConfig(
-          name=col,
           attribute=attr,
       )
     elif isinstance(attr, domain.OpenSetCategoricalAttribute):
       initializers[col] = initialization.OpenSetInitializerConfig(
-          name=col,
           attribute=attr,
       )
     else:
@@ -130,7 +128,12 @@ def from_measurements(
       domains: Mapping[str, domain.AttributeType],
   ) -> TabularCodec:
     """Builds a codec from initialization results and the original domains."""
-    columns = {col: ColumnCodec(m, domains[col]) for col, m in results.items()}
+    columns = {}
+    for col, m in results.items():
+      if m.measurement is not None and not m.measurement.clique:
+        measurement = dataclasses.replace(m.measurement, clique=(col,))
+        m = dataclasses.replace(m, measurement=measurement)
+      columns[col] = ColumnCodec(m, domains[col])
     return cls(columns=columns)
 
   @property
@@ -188,23 +191,42 @@ class TabularMechanism(api.CalibratedMechanism):
   """End-to-end DP synthetic tabular data generation, calibrated and runnable.
 
   Attributes:
-    domains: Mapping from column names to attribute domain specifications.
+    config: The preset configuration used to create this mechanism.
+    schema: The attribute domain schema or mapping.
     base_mechanism: The calibrated discrete mechanism.
     initializers: Per-column calibrated initializers.
     total_count_sigma: Sigma for the total-count mechanism.
-    cross_attribute_constraints: Constraints to enforce on generated data.
     max_records_per_user: Assumed upper bound on the number of records a single
       user contributes.
   """
 
   config: TabularConfig
-  domains: Mapping[str, domain.AttributeType]
+  schema: domain.Schema
   base_mechanism: discrete_mechanisms.CalibratedMechanism
   initializers: dict[str, api.CalibratedMechanism]
   total_count_sigma: float = dataclasses.field(repr=False)
-  cross_attribute_constraints: Sequence[constraints.Constraint] = ()
   max_records_per_user: int = 1
 
+  @property
+  def domains(self) -> Mapping[str, domain.AttributeType]:
+    return self.schema.attributes
+
+  @property
+  def cross_attribute_constraints(self) -> Sequence[Any]:
+    return self.schema.constraints
+
+  @property
+  def discrete_mechanism(self) -> api.MechanismConfig:
+    return self.config.discrete_mechanism
+
+  @property
+  def numerical_bins(self) -> int:
+    return self.config.numerical_bins
+
+  @property
+  def init_budget_fraction(self) -> float:
+    return self.config.init_budget_fraction
+
   @property
   def dp_event(self) -> dp_accounting.DpEvent:
     """Returns the composed DpEvent for all sub-mechanisms."""
@@ -248,7 +270,7 @@ def __call__(
             f'{col=} not found in dataset. Available: {list(data.columns)}'
         )
     if not cross_attribute_constraints:
-      cross_attribute_constraints = self.config.cross_attribute_constraints
+      cross_attribute_constraints = self.schema.constraints
 
     mbi_constraints = tuple(c.to_mbi() for c in cross_attribute_constraints)
 
@@ -307,44 +329,39 @@ def __call__(
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
 class TabularConfig(api.MechanismConfig):
-  """Configures end-to-end DP synthetic data generation.
-
-  This config encodes input categorical and numerical data into a discrete
-  domain using local mode primitives, runs a discrete mechanism on the
-  discretized data, and converts the synthetic output back to the original
-  domain.
-
-  Usage::
+  """Preset configuration for the tabular synthesizer.
 
-      config = TabularConfig(domains=domains)
-      calibrated = config.configure(zcdp_rho=1.0)
-      result = calibrated(rng, df)
-      synthetic_df = result.synthetic_data
+  ``TabularConfig`` defines reusable hyperparameters (such as bin counts,
+  budget allocation fractions, and discrete mechanism choice) independent of
+  any specific dataset schema. It produces a calibrated ``TabularMechanism``
+  when ``configure(schema, ...)`` or ``calibrate(schema, ...)`` is called.
 
   Attributes:
-    domains: Mapping from column names to attribute domain specifications.
     discrete_mechanism: The mechanism to run on the discretized data.
     numerical_bins: Number of bins for numerical attribute discretization.
     init_budget_fraction: Fraction of total zCDP budget allocated to per-column
       initialization (the rest goes to the discrete mechanism).
+    domains: Optional mapping from column names to attribute domain
+      specifications (for backwards compatibility).
     cross_attribute_constraints: Constraints to enforce on generated data.
+    initializers: Optional pre-configured column initializers.
   """
 
-  domains: Mapping[str, domain.AttributeType]
-  discrete_mechanism: api.MechanismConfig = discrete_mechanisms.MSTConfig()
+  discrete_mechanism: api.MechanismConfig = dataclasses.field(
+      default_factory=discrete_mechanisms.MSTConfig
+  )
   numerical_bins: int = 32
   init_budget_fraction: float = 0.1
+  domains: domain.Schema | Mapping[str, domain.AttributeType] | None = None
+  cross_attribute_constraints: Sequence[Any] = ()
   initializers: dict[str, api.MechanismConfig] | None = None
-  cross_attribute_constraints: Sequence[constraints.Constraint] = ()
 
-  def _compute_per_col_deltas(self, delta):
-    # Split delta across open-set columns, analogous to splitting zcdp_rho.
-    # Under calibrate(), any delta not consumed here is automatically
-    # available for the zCDP-to-(epsilon, delta) conversion, so this
-    # simple additive split is tight.
+  def _compute_per_col_deltas(
+      self, schema: Mapping[str, domain.AttributeType], delta: float
+  ) -> dict[str, float]:
     num_open_set = sum(
         isinstance(attr, domain.OpenSetCategoricalAttribute)
-        for attr in self.domains.values()
+        for attr in schema.values()
     )
     if num_open_set > 0 and delta <= 0:
       raise ValueError(
@@ -355,8 +372,8 @@ def _compute_per_col_deltas(self, delta):
     thresholding_delta = self.init_budget_fraction * delta
 
     per_col_deltas = {}
-    for col in self.domains:
-      if isinstance(self.domains[col], domain.OpenSetCategoricalAttribute):
+    for col in schema:
+      if isinstance(schema[col], domain.OpenSetCategoricalAttribute):
         per_col_deltas[col] = thresholding_delta / num_open_set
       else:
         per_col_deltas[col] = 0.0
@@ -364,6 +381,7 @@ def _compute_per_col_deltas(self, delta):
 
   def configure(
       self,
+      schema: domain.Schema | Mapping[str, domain.AttributeType] | None = None,
       *,
       zcdp_rho: float,
       delta: float = 0.0,
@@ -386,6 +404,9 @@ def configure(
     ensures the overall (epsilon, delta) guarantee is tight.
 
     Args:
+      schema: The attribute domain schema or mapping of column names to
+        attribute domain specifications. If omitted, falls back to
+        ``self.domains``.
       zcdp_rho: The zCDP privacy budget.
       delta: Overall approximate DP delta for the mechanism. A fraction
         (``init_budget_fraction``) is allocated to partition selection for
@@ -402,15 +423,28 @@ def configure(
       A calibrated TabularMechanism ready to be run on tabular data.
 
     Raises:
-      ValueError: If open-set attributes exist but delta is 0.
+      ValueError: If open-set attributes exist but delta is 0, or if schema is
+        not provided and self.domains is None.
     """
     api.validate_max_records_per_user(max_records_per_user)
-    per_col_deltas = self._compute_per_col_deltas(delta)
+    if schema is None:
+      schema = self.domains
+    if schema is None:
+      raise ValueError('TabularConfig requires schema.')
+    if not isinstance(schema, domain.Schema):
+      constraints_to_use = (
+          schema.constraints
+          if hasattr(schema, 'constraints')
+          else self.cross_attribute_constraints
+      )
+      schema = domain.Schema(schema, constraints=constraints_to_use)
 
+    attr_schema = schema.attributes
+    per_col_deltas = self._compute_per_col_deltas(attr_schema, delta)
     inits = (
         self.initializers
         if self.initializers is not None
-        else create_initializers(self.domains, self.numerical_bins)
+        else create_initializers(attr_schema, self.numerical_bins)
     )
     init_rho = self.init_budget_fraction * zcdp_rho
     # +1 for the DPGaussianCount that always measures the total.
@@ -418,10 +452,9 @@ def configure(
     discrete_rho = (1 - self.init_budget_fraction) * zcdp_rho
     total_count_sigma = (0.5 / per_col_rho) ** 0.5
 
-    calibrated_inits: dict[str, api.CalibratedMechanism]
-
     calibrated_inits = {
         col: init.configure(
+            attr_schema[col],
             zcdp_rho=per_col_rho,
             delta=per_col_deltas[col],
             max_records_per_user=max_records_per_user,
@@ -430,13 +463,13 @@ def configure(
     }
 
     calibrated_discrete = self.discrete_mechanism.configure(
-        max_records_per_user=max_records_per_user,
         zcdp_rho=discrete_rho,
+        max_records_per_user=max_records_per_user,
     )
 
     return TabularMechanism(
         config=self,
-        domains=self.domains,
+        schema=schema,
         base_mechanism=calibrated_discrete,
         initializers=calibrated_inits,
         total_count_sigma=total_count_sigma,
@@ -448,6 +481,9 @@ def configure(
 class TabularSynthesizer(TabularConfig):
   """Deprecated. Use TabularConfig and TabularMechanism instead."""
 
+  domains: Mapping[str, domain.AttributeType] | None = None
+  cross_attribute_constraints: Sequence[Any] = ()
+
   def __post_init__(self):
     warnings.warn(
         'TabularSynthesizer is deprecated. Use TabularConfig for configuration '
diff --git a/dpsynth/discrete_mechanisms/aim.py b/dpsynth/discrete_mechanisms/aim.py
index 243df32c..c5e765f1 100644
--- a/dpsynth/discrete_mechanisms/aim.py
+++ b/dpsynth/discrete_mechanisms/aim.py
@@ -130,7 +130,7 @@ def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
         domain, self.workload, self.max_marginal_size
     )
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     api.validate_max_records_per_user(max_records_per_user)
     return AIM(
         config=self,
diff --git a/dpsynth/discrete_mechanisms/aim_gdp.py b/dpsynth/discrete_mechanisms/aim_gdp.py
index 79f91bf6..fdda96a3 100644
--- a/dpsynth/discrete_mechanisms/aim_gdp.py
+++ b/dpsynth/discrete_mechanisms/aim_gdp.py
@@ -195,7 +195,7 @@ def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
         domain, self.workload, self.max_marginal_size
     )
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     api.validate_max_records_per_user(max_records_per_user)
     return AIMGDP(
         config=self,
diff --git a/dpsynth/discrete_mechanisms/direct.py b/dpsynth/discrete_mechanisms/direct.py
index d693f5b2..37f6292c 100644
--- a/dpsynth/discrete_mechanisms/direct.py
+++ b/dpsynth/discrete_mechanisms/direct.py
@@ -29,7 +29,7 @@
 class DirectConfig(api.MechanismConfig):
   """Config for the direct mechanism that measures prespecified marginals."""
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     api.validate_max_records_per_user(max_records_per_user)
     return Direct(
         config=self,
diff --git a/dpsynth/discrete_mechanisms/discrete.py b/dpsynth/discrete_mechanisms/discrete.py
index 924d2ea0..74704217 100644
--- a/dpsynth/discrete_mechanisms/discrete.py
+++ b/dpsynth/discrete_mechanisms/discrete.py
@@ -64,13 +64,14 @@ def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
       raise ValueError('Inner mechanism does not support supporting_cliques.')
     return self.mechanism.supporting_cliques(domain)
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     """Configures the synthesizer with a zCDP budget."""
     api.validate_max_records_per_user(max_records_per_user)
 
     one_way_rho = zcdp_rho * self.one_way_budget_fraction
     remaining_rho = zcdp_rho - one_way_rho
     inner = self.mechanism.configure(
+        _,
         zcdp_rho=remaining_rho,
         delta=delta,
         max_records_per_user=max_records_per_user,
diff --git a/dpsynth/discrete_mechanisms/independent.py b/dpsynth/discrete_mechanisms/independent.py
index ce154267..4ea289b3 100644
--- a/dpsynth/discrete_mechanisms/independent.py
+++ b/dpsynth/discrete_mechanisms/independent.py
@@ -29,7 +29,7 @@ class IndependentConfig(api.MechanismConfig):
 
   pgm_iters: int = 5000
 
-  def configure(self, *, zcdp_rho, delta=0.0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0.0, max_records_per_user=1):
     return Independent(config=self)
 
   def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
diff --git a/dpsynth/discrete_mechanisms/mst.py b/dpsynth/discrete_mechanisms/mst.py
index 037dd91e..48bf0fdb 100644
--- a/dpsynth/discrete_mechanisms/mst.py
+++ b/dpsynth/discrete_mechanisms/mst.py
@@ -185,7 +185,7 @@ def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
         self.maximum_marginal_size,
     )
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     api.validate_max_records_per_user(max_records_per_user)
     return MST(
         config=self,
diff --git a/dpsynth/discrete_mechanisms/swift.py b/dpsynth/discrete_mechanisms/swift.py
index 4d94a8ad..7bcdf6a9 100644
--- a/dpsynth/discrete_mechanisms/swift.py
+++ b/dpsynth/discrete_mechanisms/swift.py
@@ -74,7 +74,7 @@ def supporting_cliques(self, domain: mbi.Domain) -> list[mbi.Clique]:
         domain, self.workload, self.max_marginal_size
     )
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(self, _=None, *, zcdp_rho, delta=0, max_records_per_user=1):
     api.validate_max_records_per_user(max_records_per_user)
     return SWIFT(
         config=self,
diff --git a/dpsynth/domain.py b/dpsynth/domain.py
index d485cd1d..bffa93e5 100644
--- a/dpsynth/domain.py
+++ b/dpsynth/domain.py
@@ -43,6 +43,8 @@
 values when none should exist.
 """
 
+from __future__ import annotations
+
 from collections.abc import Mapping, Sequence
 import dataclasses
 import functools
@@ -51,7 +53,6 @@
 
 from typing import Any, Literal, TypeAlias
 
-from absl import logging
 import numpy as np
 import yaml
 
@@ -304,11 +305,54 @@ class FreeFormTextAttribute:
     | FreeFormTextAttribute
 )
 
-Schema: TypeAlias = Mapping[str, AttributeType]
 
+@dataclasses.dataclass(frozen=True, eq=False)
+class Schema(Mapping[str, AttributeType]):
+  """Schema defining attribute domains and optional cross-attribute constraints.
+
+  Implements ``collections.abc.Mapping[str, AttributeType]`` so it can be
+  indexed like a dictionary (e.g. ``schema['col']``, ``'col' in schema``,
+  ``len(schema)``, ``for col in schema``).
+
+  Attributes:
+    attributes: Mapping from column names to attribute domain specifications.
+    constraints: Cross-attribute constraints associated with this schema.
+  """
+
+  attributes: Mapping[str, AttributeType]
+  constraints: Sequence[Any] = ()
+
+  def __post_init__(self):
+    if isinstance(self.constraints, list):
+      object.__setattr__(self, 'constraints', tuple(self.constraints))
+
+  def __getitem__(self, key: str) -> AttributeType:
+    return self.attributes[key]
+
+  def __contains__(self, key: object) -> bool:
+    return key in self.attributes
+
+  def __iter__(self) -> Any:
+    return iter(self.attributes)
+
+  def __len__(self) -> int:
+    return len(self.attributes)
+
+  def __eq__(self, other: object) -> bool:
+    if isinstance(other, Schema):
+      return (
+          self.attributes == other.attributes
+          and self.constraints == other.constraints
+      )
+    if isinstance(other, Mapping):
+      return not self.constraints and self.attributes == other
+    return False
 
-def to_yaml_file(domain: Mapping[str, AttributeType], filepath: str | PathType):
-  """Writes a dictionary of Attribute objects to a YAML file."""
+
+def to_yaml_file(
+    domain: Schema | Mapping[str, AttributeType], filepath: str | PathType
+) -> None:
+  """Writes a Schema or dictionary of Attribute objects to a YAML file."""
   yaml_data = {}
   for name, attr_obj in domain.items():
     attr_data = dataclasses.asdict(attr_obj)
@@ -318,28 +362,25 @@ def to_yaml_file(domain: Mapping[str, AttributeType], filepath: str | PathType):
     yaml.dump(yaml_data, f, default_flow_style=False)
 
 
-def from_yaml_file(filepath: str | PathType) -> Mapping[str, AttributeType]:
-  """Reads a dictionary of Attribute objects from a YAML file."""
+def from_yaml_file(filepath: str | PathType) -> Schema:
+  """Reads a Schema from a YAML file."""
   with open(filepath, 'r') as f:
     yaml_data = yaml.safe_load(f)
-  domain = {}
-
+  attrs = {}
   for name, attr_data in yaml_data.items():
     attr_type = attr_data.pop('type', None)
-    if attr_type is None:
-      logging.warning(
-          'Field "type" missing in domain YAML; re-save using `to_yaml_file`.'
-          'In the future, missing this field will raise an error.'
-      )
-    if 'possible_values' in attr_data:
-      domain[name] = CategoricalAttribute(**attr_data)
-    elif 'min_value' in attr_data:
-      domain[name] = NumericalAttribute(**attr_data)
-    elif 'max_tokens' in attr_data:
-      domain[name] = FreeFormTextAttribute(**attr_data)
-    elif 'default_value' in attr_data or not attr_data:
-      domain[name] = OpenSetCategoricalAttribute(**attr_data)
+    if attr_type == 'CategoricalAttribute' or 'possible_values' in attr_data:
+      attrs[name] = CategoricalAttribute(**attr_data)
+    elif attr_type == 'NumericalAttribute' or 'min_value' in attr_data:
+      attrs[name] = NumericalAttribute(**attr_data)
+    elif attr_type == 'FreeFormTextAttribute' or 'max_tokens' in attr_data:
+      attrs[name] = FreeFormTextAttribute(**attr_data)
+    elif (
+        attr_type == 'OpenSetCategoricalAttribute'
+        or 'default_value' in attr_data
+        or not attr_data
+    ):
+      attrs[name] = OpenSetCategoricalAttribute(**attr_data)
     else:
       raise ValueError(f'Invalid YAML data for attribute: {name}')
-
-  return domain
+  return Schema(attrs)
diff --git a/dpsynth/local_mode/initialization.py b/dpsynth/local_mode/initialization.py
index 19187a3a..61cfe2bb 100644
--- a/dpsynth/local_mode/initialization.py
+++ b/dpsynth/local_mode/initialization.py
@@ -75,11 +75,10 @@ class ColumnMeasurement:
 class NumericalInitializerConfig(api.MechanismConfig):
   """Configuration for initializing numerical attributes."""
 
-  name: str
   num_partitions: int
-  attribute: domain.NumericalAttribute
   max_grid_size: int = 10_000_000
   epsilon_ratio: float = 2.0
+  attribute: domain.NumericalAttribute | None = None
 
   def __post_init__(self):
     if self.max_grid_size < 2:
@@ -87,27 +86,33 @@ def __post_init__(self):
     if self.num_partitions >= self.max_grid_size:
       raise ValueError(f'{self.num_partitions=} >= {self.max_grid_size=}')
 
-  @property
-  def grid_spec(self) -> tuple[float, float, int]:
+  def grid_spec(
+      self, attribute: domain.NumericalAttribute
+  ) -> tuple[float, float, int]:
     """Returns (lower, upper, grid_size) for the quantile candidate grid."""
-    attr = self.attribute
-    min_value = float(attr.min_value)
-    if attr.dtype == 'int':
+    min_value = float(attribute.min_value)
+    if attribute.dtype == 'int':
       m = _quantiles.jitter_factor(self.num_partitions)
       budget = max(2, self.max_grid_size // m)
-      int_range = int(attr.max_value - attr.min_value + 1)
+      int_range = int(attribute.max_value - attribute.min_value + 1)
       step = max(1, math.ceil(int_range / budget))
       gs = math.ceil(int_range / step)
       return min_value, min_value + (gs - 1) * step, gs
 
-    return min_value, float(attr.exclusive_max_value), self.max_grid_size
+    return min_value, float(attribute.exclusive_max_value), self.max_grid_size
 
-  @property
-  def grid_size(self) -> int:
-    return self.grid_spec[2]
-
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(
+      self,
+      attribute: domain.NumericalAttribute | None = None,
+      *,
+      zcdp_rho: float,
+      delta: float = 0.0,
+      max_records_per_user: int = 1,
+  ) -> NumericalInitializer:
     api.validate_max_records_per_user(max_records_per_user)
+    attr = attribute if attribute is not None else self.attribute
+    if attr is None:
+      raise ValueError('NumericalInitializerConfig requires attribute.')
 
     levels = int(np.log2(self.num_partitions))
     if 2**levels != self.num_partitions:
@@ -117,8 +122,14 @@ def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
     budget_weights = rho_ratio ** np.arange(levels)[::-1]
     rho_levels = zcdp_rho * budget_weights / budget_weights.sum()
     eps = np.sqrt(8.0 * rho_levels)
+    bound_config = (
+        self
+        if self.attribute is attr
+        else dataclasses.replace(self, attribute=attr)
+    )
     return NumericalInitializer(
-        config=self,
+        config=bound_config,
+        attribute=attr,
         epsilon_levels=tuple(eps.tolist()),
         max_records_per_user=max_records_per_user,
     )
@@ -129,6 +140,7 @@ class NumericalInitializer(api.CalibratedMechanism):
   """Calibrated mechanism for initializing numerical attributes."""
 
   config: NumericalInitializerConfig
+  attribute: domain.NumericalAttribute
   epsilon_levels: tuple[float, ...]
   max_records_per_user: int = 1
 
@@ -140,9 +152,13 @@ def __post_init__(self):
   def _num_levels(self) -> int:
     return int(np.log2(self.config.num_partitions))
 
+  @property
+  def grid_spec(self) -> tuple[float, float, int]:
+    return self.config.grid_spec(self.attribute)
+
   @property
   def grid_size(self) -> int:
-    return self.config.grid_size
+    return self.grid_spec[2]
 
   @property
   def zcdp_rho(self) -> float:
@@ -170,9 +186,9 @@ def __call__(
   def _grid_histogram(self, data):
     """Returns the quantile candidate-grid histogram (length grid_size)."""
     # Applies NumericalAttribute.standardize semantics in a vectorized manner.
-    lower, upper, gs = self.config.grid_spec
+    lower, upper, gs = self.grid_spec
     delta = (upper - lower) / (gs - 1)
-    attr = self.config.attribute
+    attr = self.attribute
     values = np.asarray(data, dtype=float)
     if attr.clip_to_range:
       values = np.where(np.isnan(values), attr.min_value, values)
@@ -192,9 +208,7 @@ def from_summary(
       estimated_total: float | None = None,
   ) -> ColumnMeasurement:
     """Returns a ColumnMeasurement from pre-aggregated histogram counts."""
-    jitter_strategy = (
-        'refine' if self.config.attribute.dtype == 'int' else 'symmetric'
-    )
+    jitter_strategy = 'refine' if self.attribute.dtype == 'int' else 'symmetric'
     indices = _quantiles.quantiles_from_histogram(
         rng,
         counts,
@@ -202,14 +216,13 @@ def from_summary(
         jitter_strategy=jitter_strategy,
         max_records_per_user=self.max_records_per_user,
     )
-    lower, upper, _ = self.config.grid_spec
+    lower, upper, _ = self.grid_spec
     delta = (upper - lower) / max(1, np.asarray(counts).size - 1)
     raw_edges = [lower + i * delta for i in indices]
 
     return edges_to_column_measurement(
         raw_edges=raw_edges,
-        attribute=self.config.attribute,
-        name=self.config.name,
+        attribute=self.attribute,
         zcdp_rho=self.zcdp_rho,
         estimated_total=estimated_total,
         max_records_per_user=self.max_records_per_user,
@@ -219,7 +232,6 @@ def from_summary(
 def edges_to_column_measurement(
     raw_edges,
     attribute,
-    name,
     zcdp_rho,
     estimated_total=None,
     max_records_per_user=1,
@@ -234,7 +246,6 @@ def edges_to_column_measurement(
   Args:
     raw_edges: Quantile edge values (unsorted duplicates are fine).
     attribute: The ``NumericalAttribute`` defining the data domain.
-    name: Attribute name used as the clique key in any measurement.
     zcdp_rho: Total zCDP rho consumed by the quantile mechanism.
     estimated_total: If provided, a heuristic one-way measurement is included.
     max_records_per_user: Assumed upper bound on the number of records a single
@@ -269,7 +280,7 @@ def edges_to_column_measurement(
     stddev = max_records_per_user / np.sqrt(zcdp_rho)
     measurement = mbi.LinearMeasurement(
         counts,
-        (name,),
+        (),
         stddev=stddev,
         query=mbi.DatavectorQuery(use_for_total_estimation=False),
     )
@@ -281,13 +292,28 @@ def edges_to_column_measurement(
 class CategoricalInitializerConfig(api.MechanismConfig):
   """Configuration for initializing categorical attributes."""
 
-  name: str
-  attribute: domain.CategoricalAttribute
+  attribute: domain.CategoricalAttribute | None = None
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(
+      self,
+      attribute: domain.CategoricalAttribute | None = None,
+      *,
+      zcdp_rho: float,
+      delta: float = 0.0,
+      max_records_per_user: int = 1,
+  ) -> CategoricalInitializer:
     api.validate_max_records_per_user(max_records_per_user)
+    attr = attribute if attribute is not None else self.attribute
+    if attr is None:
+      raise ValueError('CategoricalInitializerConfig requires attribute.')
+    bound_config = (
+        self
+        if self.attribute is attr
+        else dataclasses.replace(self, attribute=attr)
+    )
     return CategoricalInitializer(
-        config=self,
+        config=bound_config,
+        attribute=attr,
         sigma=math.sqrt(0.5 / zcdp_rho),
         max_records_per_user=max_records_per_user,
     )
@@ -298,6 +324,7 @@ class CategoricalInitializer(api.CalibratedMechanism):
   """Calibrated mechanism for initializing categorical attributes."""
 
   config: CategoricalInitializerConfig
+  attribute: domain.CategoricalAttribute
   sigma: float
   max_records_per_user: int = 1
 
@@ -310,8 +337,8 @@ def __call__(
       self, rng: np.random.Generator, data: np.ndarray
   ) -> ColumnMeasurement:
     """Returns a ColumnMeasurement with the noisy histogram."""
-    encoded = vtx.discrete_encode(data, self.config.attribute)
-    counts = np.bincount(encoded, minlength=self.config.attribute.size)
+    encoded = vtx.discrete_encode(data, self.attribute)
+    counts = np.bincount(encoded, minlength=self.attribute.size)
     return self.from_summary(rng, counts)
 
   def from_summary(
@@ -324,24 +351,39 @@ def from_summary(
     noisy_counts = np.asarray(noisy)
     measurement = mbi.LinearMeasurement(
         noisy_counts,
-        (self.config.name,),
+        (),
         stddev=self.max_records_per_user * self.sigma,
     )
-    return ColumnMeasurement(self.config.attribute, measurement=measurement)
+    return ColumnMeasurement(self.attribute, measurement=measurement)
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
 class OpenSetInitializerConfig(api.MechanismConfig):
   """Configuration for initializing open-set categorical attributes."""
 
-  name: str
-  attribute: domain.OpenSetCategoricalAttribute
   min_count: int = 1
+  attribute: domain.OpenSetCategoricalAttribute | None = None
 
-  def configure(self, *, zcdp_rho, delta=0, max_records_per_user=1):
+  def configure(
+      self,
+      attribute: domain.OpenSetCategoricalAttribute | None = None,
+      *,
+      zcdp_rho: float,
+      delta: float = 0.0,
+      max_records_per_user: int = 1,
+  ) -> OpenSetInitializer:
     api.validate_max_records_per_user(max_records_per_user)
+    attr = attribute if attribute is not None else self.attribute
+    if attr is None:
+      raise ValueError('OpenSetInitializerConfig requires attribute.')
+    bound_config = (
+        self
+        if self.attribute is attr
+        else dataclasses.replace(self, attribute=attr)
+    )
     return OpenSetInitializer(
-        config=self,
+        config=bound_config,
+        attribute=attr,
         max_records_per_user=max_records_per_user,
         sigma=math.sqrt(0.5 / zcdp_rho),
         delta=delta,
@@ -353,6 +395,7 @@ class OpenSetInitializer(api.CalibratedMechanism):
   """Calibrated mechanism for initializing open-set categorical attributes."""
 
   config: OpenSetInitializerConfig
+  attribute: domain.OpenSetCategoricalAttribute
   max_records_per_user: int = 1
   sigma: float
   delta: float
@@ -400,8 +443,8 @@ def from_summary(
         [str(v) for v in unique_values[selected_partitions]]
     )
 
-    if self.config.attribute.public_possible_values:
-      pub = np.array(self.config.attribute.public_possible_values)
+    if self.attribute.public_possible_values:
+      pub = np.array(self.attribute.public_possible_values)
       selected_values, estimated_counts = primitives.ensure_public_partitions(
           rng,
           selected_values,
@@ -411,7 +454,7 @@ def from_summary(
       )
 
     # Build the discovered domain: default first, then selected values.
-    default = self.config.attribute.default_value
+    default = self.attribute.default_value
     possible_values = [default] + selected_values.tolist()
     cat_attr = domain.CategoricalAttribute(possible_values)
 
@@ -419,7 +462,7 @@ def from_summary(
     # not the unmeasured default at index 0.
     measurement = mbi.LinearMeasurement(
         estimated_counts,  # pyrefly: ignore[bad-argument-type]
-        (self.config.name,),
+        (),
         stddev=stddev,
         query=mbi.SlicedQuery(start=1),
     )
diff --git a/dpsynth/relational/__init__.py b/dpsynth/relational/__init__.py
index efe45d29..2358b9a2 100644
--- a/dpsynth/relational/__init__.py
+++ b/dpsynth/relational/__init__.py
@@ -17,6 +17,7 @@
 # pylint: disable=g-importing-member
 
 from dpsynth.relational.domain import ForeignKeyRelation
+from dpsynth.relational.domain import RelationalSchema
 from dpsynth.relational.synthesizer import MultiDataGenerationResult
 from dpsynth.relational.synthesizer import MultiTableConfig
 from dpsynth.relational.synthesizer import MultiTableMechanism
@@ -26,4 +27,5 @@
     'MultiDataGenerationResult',
     'MultiTableConfig',
     'MultiTableMechanism',
+    'RelationalSchema',
 ]
diff --git a/dpsynth/relational/domain.py b/dpsynth/relational/domain.py
index e4235ccc..95e326d5 100644
--- a/dpsynth/relational/domain.py
+++ b/dpsynth/relational/domain.py
@@ -59,6 +59,47 @@ def __post_init__(self):
       )
 
 
+@dataclasses.dataclass(frozen=True)
+class RelationalSchema:
+  """Schema defining relational tables and foreign key relationships.
+
+  Attributes:
+    tables: Mapping from table name to table Schema or AttributeType mapping.
+    foreign_keys: Sequence of foreign key relationships between tables.
+  """
+
+  tables: Mapping[str, domain.Schema]
+  foreign_keys: Sequence[ForeignKeyRelation] = ()
+
+  def __post_init__(self):
+    if isinstance(self.foreign_keys, list):
+      object.__setattr__(self, 'foreign_keys', tuple(self.foreign_keys))
+    normalized_tables = {}
+    for table_name, table_schema in self.tables.items():
+      if isinstance(table_schema, domain.Schema):
+        normalized_tables[table_name] = table_schema
+      elif isinstance(table_schema, Mapping):
+        normalized_tables[table_name] = domain.Schema(table_schema)
+      else:
+        raise TypeError(
+            f'Table {table_name!r} schema must be a Schema or Mapping, got'
+            f' {type(table_schema).__name__}.'
+        )
+    object.__setattr__(self, 'tables', normalized_tables)
+
+  def __getitem__(self, key: str) -> domain.Schema:
+    return self.tables[key]
+
+  def __contains__(self, key: object) -> bool:
+    return key in self.tables
+
+  def __iter__(self) -> Any:
+    return iter(self.tables)
+
+  def __len__(self) -> int:
+    return len(self.tables)
+
+
 def topological_sort_hierarchy(
     tables: Sequence[str],
     foreign_keys: Sequence[ForeignKeyRelation],
@@ -166,14 +207,14 @@ def _parse_attribute(
 
 def from_dict(
     config: Mapping[str, Any],
-) -> tuple[dict[str, domain.Schema], list[ForeignKeyRelation]]:
+) -> RelationalSchema:
   """Parses multi-table schema and foreign keys from a dictionary.
 
   Args:
     config: Dictionary with 'tables' and optional 'foreign_keys' blocks.
 
   Returns:
-    A tuple of (table_domains, foreign_keys).
+    A RelationalSchema instance.
 
   Raises:
     ValueError: If configuration format or attribute specifications are invalid.
@@ -189,10 +230,10 @@ def from_dict(
   for table_name, table_schema in config['tables'].items():
     if not isinstance(table_schema, Mapping):
       raise ValueError(f'Table schema for {table_name!r} must be a mapping.')
-    table_domains[table_name] = {
+    table_domains[table_name] = domain.Schema({
         col_name: _parse_attribute(table_name, col_name, spec)
         for col_name, spec in table_schema.items()
-    }
+    })
 
   foreign_keys: list[ForeignKeyRelation] = []
   for fk in config.get('foreign_keys', []):
@@ -208,19 +249,19 @@ def from_dict(
       len(table_domains),
       len(foreign_keys),
   )
-  return table_domains, foreign_keys
+  return RelationalSchema(tables=table_domains, foreign_keys=foreign_keys)
 
 
 def from_yaml_file(
     filepath: str | PathType,
-) -> tuple[dict[str, domain.Schema], list[ForeignKeyRelation]]:
+) -> RelationalSchema:
   """Reads multi-table schema and foreign keys from a YAML file.
 
   Args:
     filepath: Path to the YAML schema file.
 
   Returns:
-    A tuple of (table_domains, foreign_keys).
+    A RelationalSchema instance.
   """
   logging.info('Loading relational domain schema from YAML file: %s', filepath)
   path = epath.Path(filepath)
@@ -232,52 +273,59 @@ def from_yaml_file(
 
 
 def to_dict(
-    table_domains: Mapping[str, domain.Schema],
+    schema: RelationalSchema | Mapping[str, domain.Schema],
     foreign_keys: Sequence[ForeignKeyRelation] = (),
 ) -> dict[str, Any]:
   """Converts multi-table schemas and foreign keys to a dictionary.
 
   Args:
-    table_domains: Mapping from table name to per-column AttributeType schemas.
-    foreign_keys: Optional sequence of ForeignKeyRelation objects.
+    schema: RelationalSchema or mapping from table name to table schemas.
+    foreign_keys: Optional sequence of ForeignKeyRelation objects (if schema is
+      a mapping).
 
   Returns:
     A dictionary with 'tables' and optional 'foreign_keys' blocks.
   """
+  if isinstance(schema, RelationalSchema):
+    table_domains = schema.tables
+    fks = schema.foreign_keys
+  else:
+    table_domains = schema
+    fks = foreign_keys
+
   tables_dict: dict[str, dict[str, Any]] = {}
-  for table_name, schema in table_domains.items():
+  for table_name, table_schema in table_domains.items():
     table_dict: dict[str, Any] = {}
-    for col_name, attr in schema.items():
+    for col_name, attr in table_schema.items():
       attr_dict = dataclasses.asdict(attr)
       attr_dict['type'] = attr.__class__.__name__
       table_dict[col_name] = attr_dict
     tables_dict[table_name] = table_dict
 
   result: dict[str, Any] = {'tables': tables_dict}
-  if foreign_keys:
-    result['foreign_keys'] = [dataclasses.asdict(fk) for fk in foreign_keys]
+  if fks:
+    result['foreign_keys'] = [dataclasses.asdict(fk) for fk in fks]
   return result
 
 
 def to_yaml_file(
-    table_domains: Mapping[str, domain.Schema],
-    foreign_keys: Sequence[ForeignKeyRelation],
+    schema: RelationalSchema | Mapping[str, domain.Schema],
     filepath: str | PathType,
+    foreign_keys: Sequence[ForeignKeyRelation] = (),
 ) -> None:
   """Writes multi-table schema and foreign keys to a YAML file.
 
   Args:
-    table_domains: Mapping from table name to per-column AttributeType schemas.
-    foreign_keys: Sequence of ForeignKeyRelation objects.
+    schema: RelationalSchema or mapping from table name to table schemas.
     filepath: Destination path for the YAML schema file.
+    foreign_keys: Optional sequence of ForeignKeyRelation objects (if schema is
+      a mapping).
   """
   logging.info(
-      'Saving relational domain schema (%d tables, %d foreign keys) to: %s',
-      len(table_domains),
-      len(foreign_keys),
+      'Saving relational domain schema to: %s',
       filepath,
   )
-  data = to_dict(table_domains, foreign_keys)
+  data = to_dict(schema, foreign_keys=foreign_keys)
   path = epath.Path(filepath)
   with path.open('w') as f:
     yaml.dump(data, f, default_flow_style=False, sort_keys=False)
diff --git a/dpsynth/relational/synthesizer.py b/dpsynth/relational/synthesizer.py
index 8ecbfb1a..9db3c12b 100644
--- a/dpsynth/relational/synthesizer.py
+++ b/dpsynth/relational/synthesizer.py
@@ -19,6 +19,7 @@
 from collections.abc import Collection, Hashable, Mapping, Sequence
 import dataclasses
 import math
+import warnings
 from typing import Any, Literal
 
 from absl import logging
@@ -192,7 +193,7 @@ def _run_single_col_initializer(
     ValueError: If init is not a supported initializer type.
   """
   if isinstance(init, initialization.NumericalInitializer):
-    attr = init.config.attribute
+    attr = init.attribute
     values = np.asarray(data, dtype=float)
     if attr.clip_to_range:
       values = np.where(np.isnan(values), attr.min_value, values)
@@ -201,16 +202,16 @@ def _run_single_col_initializer(
       values, weights = values[in_domain], weights[in_domain]
     if attr.dtype == 'int':
       values = np.round(values)
-    lower, upper, gs = init.config.grid_spec
+    lower, upper, gs = init.grid_spec
     delta = (upper - lower) / (gs - 1)
     indices = initialization.encode_to_grid(values, lower, upper, delta)
     counts = np.bincount(indices, weights=weights, minlength=gs)
     return init.from_summary(rng, counts, estimated_total=estimated_total)
 
   if isinstance(init, initialization.CategoricalInitializer):
-    encoded = vtx.discrete_encode(data, init.config.attribute)
+    encoded = vtx.discrete_encode(data, init.attribute)
     counts = np.bincount(
-        encoded, weights=weights, minlength=init.config.attribute.size
+        encoded, weights=weights, minlength=init.attribute.size
     )
     return init.from_summary(rng, counts)
 
@@ -937,30 +938,42 @@ class MultiTableMechanism(api.CalibratedMechanism):
   """Calibrated, runnable multi-table relational differential privacy mechanism.
 
   Attributes:
-    domains: Mapping from table name to per-column attribute specifications.
-    foreign_keys: Sequence of foreign key relationships defining the hierarchy.
+    config: MultiTableConfig hyperparameter preset.
+    schema: RelationalSchema containing table schemas and foreign keys.
     calibrated_discrete_mechanisms: Mapping from link names to calibrated
       discrete mechanisms.
     calibrated_initializers: Mapping from table and column to calibrated
       initializers.
     total_count_sigma: Sigma for the root table total-count mechanism.
-    num_permutation_slots: Permutation exploration slot count (o), default 2.
-    exploration_strategy: Exploration strategy ('empty_token' or 'size_sliced').
     max_records_per_user: Assumed upper bound on records a single user
-      contributes to the root table. Essentially the sensitivitiy at the root.
+      contributes to the root table. Essentially the sensitivity at the root.
 
-  Note: For simplicity, user-defined contraints are not supported yet.
+  Note: For simplicity, user-defined constraints are not supported yet.
   """
 
-  domains: Mapping[str, domain.Schema]
-  foreign_keys: Sequence[rel_domain.ForeignKeyRelation]
+  config: MultiTableConfig
+  schema: rel_domain.RelationalSchema
   calibrated_discrete_mechanisms: Mapping[str, api.CalibratedMechanism]
   calibrated_initializers: Mapping[str, Mapping[str, api.CalibratedMechanism]]
   total_count_sigma: float = dataclasses.field(repr=False)
-  num_permutation_slots: int = 2
-  exploration_strategy: Literal['empty_token', 'size_sliced'] = 'empty_token'
   max_records_per_user: int = 1
 
+  @property
+  def domains(self) -> Mapping[str, domain.Schema]:
+    return self.schema.tables
+
+  @property
+  def foreign_keys(self) -> Sequence[rel_domain.ForeignKeyRelation]:
+    return self.schema.foreign_keys
+
+  @property
+  def num_permutation_slots(self) -> int:
+    return self.config.num_permutation_slots
+
+  @property
+  def exploration_strategy(self) -> Literal['empty_token', 'size_sliced']:
+    return self.config.exploration_strategy
+
   @property
   def dp_event(self) -> dp_accounting.DpEvent:
     """Returns the composed DpEvent combining all relational sub-mechanisms.
@@ -1049,14 +1062,11 @@ def __call__(
     )
 
 
-@dataclasses.dataclass
+@dataclasses.dataclass(frozen=True, kw_only=True)
 class MultiTableConfig(api.MechanismConfig):
   """Configuration recipe for multi-table relational differential privacy synthesis.
 
   Attributes:
-    domains: Mapping from table name to per-column attribute domain
-      specifications.
-    foreign_keys: Sequence of foreign key relationships defining the hierarchy.
     discrete_mechanism: Discrete mechanism config (e.g. AIM, MST) for relational
       links.
     numerical_bins: Number of bins for numerical attribute discretization.
@@ -1066,29 +1076,15 @@ class MultiTableConfig(api.MechanismConfig):
     exploration_strategy: Exploration strategy ('empty_token' or 'size_sliced').
   """
 
-  domains: Mapping[str, domain.Schema]
-  foreign_keys: Sequence[rel_domain.ForeignKeyRelation]
   discrete_mechanism: api.MechanismConfig = dataclasses.field(
       default_factory=discrete_mechanisms.AIMConfig
   )
   numerical_bins: int = 32
   init_budget_fraction: float = 0.1
-  initializers: Mapping[str, Mapping[str, api.MechanismConfig]] | None = None
   num_permutation_slots: int = 2
   exploration_strategy: Literal['empty_token', 'size_sliced'] = 'empty_token'
 
   def __post_init__(self):
-    if len(self.domains) < 2:
-      raise ValueError(
-          'MultiTableConfig requires at least two tables in domains, got'
-          f' {len(self.domains)}. For single-table synthesis, use'
-          ' TabularConfig.'
-      )
-    if not self.foreign_keys:
-      raise ValueError(
-          'MultiTableConfig requires at least one foreign key relationship in'
-          ' foreign_keys. For single-table synthesis, use TabularConfig.'
-      )
     if not (0.0 < self.init_budget_fraction < 1.0):
       raise ValueError(
           'init_budget_fraction must be strictly in (0.0, 1.0), got'
@@ -1113,17 +1109,30 @@ def __post_init__(self):
           f' {type(self.discrete_mechanism).__name__}.'
       )
 
+  def _validate_schema(self, schema: rel_domain.RelationalSchema) -> None:
+    if len(schema.tables) < 2:
+      raise ValueError(
+          'MultiTableConfig requires at least two tables in schema, got'
+          f' {len(schema.tables)}. For single-table synthesis, use'
+          ' TabularConfig.'
+      )
+    if not schema.foreign_keys:
+      raise ValueError(
+          'MultiTableConfig requires at least one foreign key relationship in'
+          ' foreign_keys. For single-table synthesis, use TabularConfig.'
+      )
+
     # 1. Validate table names, column names, and attribute types.
-    for table_name, schema in self.domains.items():
+    for table_name, table_schema in schema.tables.items():
       if '.' in table_name:
         raise ValueError(
             f"Table name {table_name!r} must not contain '.' characters."
         )
-      if not schema:
+      if not table_schema:
         raise ValueError(
             f'Table {table_name!r} schema in domains cannot be empty.'
         )
-      for col_name, attr in schema.items():
+      for col_name, attr in table_schema.items():
         if '.' in col_name:
           raise ValueError(
               f"Table {table_name!r} column {col_name!r} must not contain '.'"
@@ -1155,7 +1164,7 @@ def __post_init__(self):
     # 2. Validate DAG hierarchy (acyclicity, known tables,
     # in-degree <= 1, single root).
     hierarchy = rel_domain.topological_sort_hierarchy(
-        list(self.domains.keys()), self.foreign_keys
+        list(schema.tables.keys()), schema.foreign_keys
     )
     roots = [t for _, t, fk in hierarchy if fk is None]
     if len(roots) > 1:
@@ -1166,41 +1175,23 @@ def __post_init__(self):
       )
 
     # 3. Ensure PK and FK columns are not present in domain schemas.
-    for fk in self.foreign_keys:
-      if fk.parent_primary_key in self.domains[fk.parent_table]:
+    for fk in schema.foreign_keys:
+      if fk.parent_primary_key in schema.tables[fk.parent_table]:
         raise ValueError(
             f'Primary key column {fk.parent_primary_key!r} of table'
-            f' {fk.parent_table!r} must not be in domains[{fk.parent_table!r}].'
+            f' {fk.parent_table!r} must not be in'
+            f' schema.tables[{fk.parent_table!r}].'
         )
-      if fk.child_foreign_key in self.domains[fk.child_table]:
+      if fk.child_foreign_key in schema.tables[fk.child_table]:
         raise ValueError(
             f'Foreign key column {fk.child_foreign_key!r} of table'
-            f' {fk.child_table!r} must not be in domains[{fk.child_table!r}].'
+            f' {fk.child_table!r} must not be in'
+            f' schema.tables[{fk.child_table!r}].'
         )
 
-    # 4. Validate custom initializers structure if provided.
-    if self.initializers is not None:
-      if set(self.initializers.keys()) != set(self.domains.keys()):
-        raise ValueError(
-            f'Custom initializers tables {set(self.initializers.keys())} do not'
-            f' match domains tables {set(self.domains.keys())}.'
-        )
-      for table_name, table_inits in self.initializers.items():
-        if set(table_inits.keys()) != set(self.domains[table_name].keys()):
-          raise ValueError(
-              f'Custom initializers for table {table_name!r}'
-              f' columns {set(table_inits.keys())} do not match'
-              f' domains columns {set(self.domains[table_name].keys())}.'
-          )
-        for col_name, init_cfg in table_inits.items():
-          if not isinstance(init_cfg, api.MechanismConfig):
-            raise ValueError(
-                f'Custom initializer for {table_name}.{col_name} must be an'
-                f' api.MechanismConfig, got {type(init_cfg).__name__}.'
-            )
-
   def configure(
       self,
+      schema: rel_domain.RelationalSchema | None = None,
       *,
       zcdp_rho: float,
       delta: float = 0.0,
@@ -1227,6 +1218,7 @@ def configure(
         guaranteeing root parent differential privacy without Cartesian joins.
 
     Args:
+      schema: RelationalSchema defining tables and foreign key relationships.
       zcdp_rho: The total zCDP privacy budget (rho > 0).
       delta: Approximate DP delta for open-set Gaussian partition selection.
       max_records_per_user: Upper bound on root entity contributions (>= 1).
@@ -1238,28 +1230,31 @@ def configure(
       ValueError: If configuration hyperparameters or budgets are invalid.
     """
     api.validate_max_records_per_user(max_records_per_user)
+    if schema is None:
+      raise ValueError('MultiTableConfig requires schema.')
     if zcdp_rho <= 0:
       raise ValueError(f'zcdp_rho must be positive, got {zcdp_rho}.')
 
+    if not isinstance(schema, rel_domain.RelationalSchema):
+      schema = rel_domain.RelationalSchema(schema)
+
+    self._validate_schema(schema)
+
     hierarchy = rel_domain.topological_sort_hierarchy(
-        list(self.domains.keys()), self.foreign_keys
+        list(schema.tables.keys()), schema.foreign_keys
     )
     link_sensitivities = _compute_link_sensitivities(
         hierarchy, max_records_per_user=max_records_per_user
     )
 
     per_col_deltas = _compute_table_col_deltas(
-        self.domains,
+        schema.tables,
         delta=delta,
         init_budget_fraction=self.init_budget_fraction,
     )
-    inits = (
-        self.initializers
-        if self.initializers is not None
-        else _create_table_initializers(self.domains, self.numerical_bins)
-    )
+    inits = _create_table_initializers(schema.tables, self.numerical_bins)
 
-    total_cols = sum(len(schema) for schema in self.domains.values())
+    total_cols = sum(len(s) for s in schema.tables.values())
     init_rho = self.init_budget_fraction * zcdp_rho
     per_col_rho = init_rho / (total_cols + 1)  # +1 for root table total count.
     total_count_rho = per_col_rho
@@ -1290,12 +1285,24 @@ def configure(
     }
 
     return MultiTableMechanism(
-        domains=self.domains,
-        foreign_keys=self.foreign_keys,
+        config=self,
+        schema=schema,
         calibrated_discrete_mechanisms=calibrated_discrete,
         calibrated_initializers=calibrated_inits,
         total_count_sigma=total_count_sigma,
-        num_permutation_slots=self.num_permutation_slots,
-        exploration_strategy=self.exploration_strategy,
         max_records_per_user=max_records_per_user,
     )
+
+
+class MultiTableSynthesizer(MultiTableConfig):
+  """Deprecated. Use MultiTableConfig and MultiTableMechanism instead."""
+
+  def __init__(self, *args: Any, **kwargs: Any):
+    warnings.warn(
+        'MultiTableSynthesizer is deprecated. Use MultiTableConfig for'
+        ' configuration and MultiTableMechanism for the calibrated runnable'
+        ' mechanism.',
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    super().__init__(*args, **kwargs)
diff --git a/examples/relational/california_census/example_california_census.py b/examples/relational/california_census/example_california_census.py
index 1381db8e..4ea86f5e 100644
--- a/examples/relational/california_census/example_california_census.py
+++ b/examples/relational/california_census/example_california_census.py
@@ -160,14 +160,15 @@ def create_calibrated_mechanism(
       num_permutation_slots,
       exploration_strategy,
   )
+  schema = rel_domain.RelationalSchema(
+      tables=table_domains, foreign_keys=foreign_keys
+  )
   config = rel_synth.MultiTableConfig(
-      domains=table_domains,
-      foreign_keys=foreign_keys,
       discrete_mechanism=discrete_mechanisms.MSTConfig(pgm_iters=PGM_ITERS),
       num_permutation_slots=num_permutation_slots,
       exploration_strategy=exploration_strategy,
   )
-  mechanism = config.calibrate(epsilon=epsilon, delta=delta)
+  mechanism = config.calibrate(schema, epsilon=epsilon, delta=delta)
   assert isinstance(mechanism, rel_synth.MultiTableMechanism)
   logging.info('Mechanism calibrated successfully.')
   return mechanism
diff --git a/tests/adapters/beam_test.py b/tests/adapters/beam_test.py
index 5e9dbed6..ba9f79e6 100644
--- a/tests/adapters/beam_test.py
+++ b/tests/adapters/beam_test.py
@@ -61,7 +61,6 @@ class NumericalHistogramTest(absltest.TestCase):
 
   def _run(self, rows, attr, max_grid_size=101, num_partitions=4):
     init = initialization.NumericalInitializerConfig(
-        name='x',
         num_partitions=num_partitions,
         attribute=attr,
         max_grid_size=max_grid_size,
@@ -79,7 +78,6 @@ def _run(self, rows, attr, max_grid_size=101, num_partitions=4):
   def _ref_counts(self, values, attr, max_grid_size=101, num_partitions=4):
     """In-memory grid histogram as an {index: count} dict."""
     init = initialization.NumericalInitializerConfig(
-        name='x',
         num_partitions=num_partitions,
         attribute=attr,
         max_grid_size=max_grid_size,
@@ -165,7 +163,6 @@ def test_basic_counts(self):
         out_of_domain_index=0,
     )
     init = initialization.CategoricalInitializerConfig(
-        name='col',
         attribute=attr,
     )
     rows = [
@@ -197,9 +194,7 @@ class OpenSetCountsTest(absltest.TestCase):
 
   def test_basic_counts(self):
     attr = domain.OpenSetCategoricalAttribute(default_value='<OOD>')
-    init = initialization.OpenSetInitializerConfig(
-        name='col', attribute=attr, min_count=1
-    )
+    init = initialization.OpenSetInitializerConfig(attribute=attr, min_count=1)
     rows = [
         {'col': 'apple'},
         {'col': 'apple'},
@@ -232,13 +227,13 @@ def test_end_to_end_mixed(self):
 
     initializers = {
         'score': initialization.NumericalInitializerConfig(
-            name='score', num_partitions=4, attribute=num_attr
+            num_partitions=4, attribute=num_attr
         ),
         'grade': initialization.CategoricalInitializerConfig(
-            name='grade', attribute=cat_attr
+            attribute=cat_attr
         ),
         'tag': initialization.OpenSetInitializerConfig(
-            name='tag', attribute=open_attr, min_count=1
+            attribute=open_attr, min_count=1
         ),
     }
 
@@ -275,11 +270,9 @@ def test_marginals_match_manual_counts(self):
     cat_attr = domain.CategoricalAttribute(possible_values=['a', 'b', 'c'])
     num_attr = domain.NumericalAttribute(min_value=0, max_value=10)
     cat_init = initialization.CategoricalInitializerConfig(
-        name='color',
         attribute=cat_attr,
     )
     num_init = initialization.NumericalInitializerConfig(
-        name='size',
         num_partitions=4,
         attribute=num_attr,
         max_grid_size=11,
@@ -351,8 +344,9 @@ def _domains(self):
     }
 
   def test_end_to_end_generates_synthetic_data(self):
-    synth = data_generation_v3.TabularConfig(domains=self._domains())
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
+    beam_synth = beam_adapter.BeamTabularConfig().configure(
+        self._domains(), zcdp_rho=100.0
+    )
     rows = [
         {'color': 'r', 'size': 's'},
         {'color': 'g', 'size': 'm'},
@@ -371,8 +365,9 @@ def test_end_to_end_mixed_types(self):
         'age': domain.NumericalAttribute(min_value=0, max_value=100),
         'grade': domain.CategoricalAttribute(possible_values=['a', 'b', 'c']),
     }
-    synth = data_generation_v3.TabularConfig(domains=domains)
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
+    beam_synth = beam_adapter.BeamTabularConfig().configure(
+        domains, zcdp_rho=100.0
+    )
     rng_data = np.random.default_rng(0)
     rows = [
         {
@@ -408,10 +403,10 @@ def test_runs_across_mechanisms(self, mechanism):
         'a': domain.CategoricalAttribute(possible_values=['x', 'y']),
         'b': domain.CategoricalAttribute(possible_values=['p', 'q', 'r']),
     }
-    synth = data_generation_v3.TabularConfig(
-        domains=domains, discrete_mechanism=mechanism
+    synth = data_generation_v3.TabularConfig(discrete_mechanism=mechanism)
+    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(
+        domains, zcdp_rho=100.0
     )
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
     rows = [
         {'a': 'x', 'b': 'p'},
         {'a': 'y', 'b': 'q'},
@@ -427,8 +422,9 @@ def test_runs_across_mechanisms(self, mechanism):
   def test_total_count_matches_input_under_high_budget(self):
     """With negligible noise, synthetic row count matches the input (F2)."""
     domains = {'a': domain.CategoricalAttribute(possible_values=['x', 'y'])}
-    synth = data_generation_v3.TabularConfig(domains=domains)
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=1e8)
+    beam_synth = beam_adapter.BeamTabularConfig().configure(
+        domains, zcdp_rho=1e8
+    )
     rows = [{'a': 'x'}, {'a': 'y'}] * 150  # 300 rows.
 
     result = beam_synth(np.random.default_rng(0), _rows_fn(rows))
@@ -445,10 +441,10 @@ def test_respects_impossible_combinations(self):
         attribute_domains=(a_attr, b_attr),
         impossible_combinations=[('a0', 'b1')],
     )
-    synth = data_generation_v3.TabularConfig(
-        domains=domains, cross_attribute_constraints=(constraint,)
+    schema = domain.Schema(attributes=domains, constraints=(constraint,))
+    beam_synth = beam_adapter.BeamTabularConfig().configure(
+        schema, zcdp_rho=100.0
     )
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
     # The data never contains (a0, b1). Without enforcement, independent
     # (a, b) marginals would put ~25% of mass on that cell; forwarding the
     # constraint suppresses it to a few percent (mbi's constrained sampling
@@ -472,8 +468,9 @@ def test_preserves_domain_column_order(self):
         'm': domain.CategoricalAttribute(possible_values=['c', 'd']),
         'a': domain.CategoricalAttribute(possible_values=['e', 'f']),
     }
-    synth = data_generation_v3.TabularConfig(domains=domains)
-    beam_synth = beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
+    beam_synth = beam_adapter.BeamTabularConfig().configure(
+        domains, zcdp_rho=100.0
+    )
     rows = [
         {'z': 'a', 'm': 'c', 'a': 'e'},
         {'z': 'b', 'm': 'd', 'a': 'f'},
@@ -484,11 +481,9 @@ def test_preserves_domain_column_order(self):
     self.assertEqual(list(result.synthetic_data.columns), ['z', 'm', 'a'])
 
   def test_configure_returns_calibrated_wrapper(self):
-    beam_synth = beam_adapter.BeamTabularConfig(
-        data_generation_v3.TabularConfig(domains=self._domains())
-    )
+    beam_synth = beam_adapter.BeamTabularConfig()
 
-    configured = beam_synth.configure(zcdp_rho=1.0)
+    configured = beam_synth.configure(self._domains(), zcdp_rho=1.0)
 
     self.assertIsInstance(configured, beam_adapter.BeamTabularMechanism)
     # dp_event is delegated to the wrapped, now-calibrated synthesizer.
@@ -500,30 +495,26 @@ def test_configure_returns_calibrated_wrapper(self):
   def test_inherited_calibrate_produces_calibrated_wrapper(self):
     # calibrate is inherited from DPMechanism; it binary-searches a zCDP budget
     # by repeatedly calling our configure (which delegates to the synthesizer).
-    beam_synth = beam_adapter.BeamTabularConfig(
-        data_generation_v3.TabularConfig(domains=self._domains())
-    )
+    beam_synth = beam_adapter.BeamTabularConfig()
 
-    calibrated = beam_synth.calibrate(epsilon=1.0, delta=1e-6)
+    calibrated = beam_synth.calibrate(self._domains(), epsilon=1.0, delta=1e-6)
 
     self.assertIsInstance(calibrated, beam_adapter.BeamTabularMechanism)
     self.assertIsNotNone(calibrated.dp_event)
 
   def test_uncalibrated_call_raises(self):
-    beam_synth = beam_adapter.BeamTabularConfig(
-        data_generation_v3.TabularConfig(domains=self._domains())
-    )
+    beam_synth = beam_adapter.BeamTabularConfig()
     with self.assertRaises(Exception):
       beam_synth(np.random.default_rng(0), lambda p: p)
 
   def test_honors_temp_location(self):
-    synth = data_generation_v3.TabularConfig(
-        domains={'a': domain.CategoricalAttribute(possible_values=['x', 'y'])}
-    )
     temp_dir = self.create_tempdir().full_path
     beam_synth = beam_adapter.BeamTabularConfig(
-        synth, temp_location=temp_dir
-    ).configure(zcdp_rho=100.0)
+        temp_location=temp_dir
+    ).configure(
+        {'a': domain.CategoricalAttribute(possible_values=['x', 'y'])},
+        zcdp_rho=100.0,
+    )
     rows = [{'a': 'x'}, {'a': 'y'}] * 50
 
     result = beam_synth(np.random.default_rng(0), _rows_fn(rows))
@@ -532,10 +523,10 @@ def test_honors_temp_location(self):
     self.assertTrue(os.path.exists(os.path.join(temp_dir, 'clique_vector.bin')))
 
   def _single_col_synth(self):
-    synth = data_generation_v3.TabularConfig(
-        domains={'a': domain.CategoricalAttribute(possible_values=['x', 'y'])}
+    return beam_adapter.BeamTabularConfig().configure(
+        {'a': domain.CategoricalAttribute(possible_values=['x', 'y'])},
+        zcdp_rho=100.0,
     )
-    return beam_adapter.BeamTabularConfig(synth).configure(zcdp_rho=100.0)
 
   def _spy_mkdtemp(self):
     """Returns (created_paths_list, patched_mkdtemp) recording our temp dirs.
@@ -580,13 +571,13 @@ def failing_rows_fn(_):
     self.assertFalse(os.path.exists(created[0]))
 
   def test_forwards_pipeline_options_to_both_passes(self):
-    synth = data_generation_v3.TabularConfig(
-        domains={'a': domain.CategoricalAttribute(possible_values=['x', 'y'])}
-    )
     options = pipeline_options.PipelineOptions(flags=['--runner=DirectRunner'])
     beam_synth = beam_adapter.BeamTabularConfig(
-        synth, pipeline_options=options
-    ).configure(zcdp_rho=100.0)
+        pipeline_options=options
+    ).configure(
+        {'a': domain.CategoricalAttribute(possible_values=['x', 'y'])},
+        zcdp_rho=100.0,
+    )
     rows = [{'a': 'x'}, {'a': 'y'}] * 50
     seen_options = []
     real_pipeline = beam.Pipeline
diff --git a/tests/adapters/pydantic_api_test.py b/tests/adapters/pydantic_api_test.py
index aa254893..adf6fd9a 100644
--- a/tests/adapters/pydantic_api_test.py
+++ b/tests/adapters/pydantic_api_test.py
@@ -256,9 +256,9 @@ def test_dp_synthetic_data_generation_with_supported_model(self):
 
     domains = pydantic_api.infer_domain_from_model(SupportedModel)
     df = pydantic_api.models_to_dataframe(real_data, domains)
-    synth = data_generation_v3.TabularSynthesizer(
-        domains=domains,
-    ).calibrate(epsilon=epsilon, delta=delta)
+    synth = data_generation_v3.TabularConfig().calibrate(
+        domains, epsilon=epsilon, delta=delta
+    )
     synthetic_df = synth(np.random.default_rng(), df).synthetic_data
     synthetic_records = pydantic_api.dataframe_to_models(
         synthetic_df, SupportedModel, domains
@@ -294,9 +294,9 @@ def test_dp_synthetic_data_generation_with_numerical_model(self):
 
     domains = pydantic_api.infer_domain_from_model(ModelForNumericalDefaults)
     df = pydantic_api.models_to_dataframe(real_data, domains)
-    synth = data_generation_v3.TabularSynthesizer(
-        domains=domains,
-    ).calibrate(epsilon=epsilon, delta=delta)
+    synth = data_generation_v3.TabularConfig().calibrate(
+        domains, epsilon=epsilon, delta=delta
+    )
     synthetic_df = synth(np.random.default_rng(), df).synthetic_data
     synthetic_records = pydantic_api.dataframe_to_models(
         synthetic_df, ModelForNumericalDefaults, domains
@@ -334,9 +334,9 @@ def test_dp_synthetic_data_generation_with_categorical_model(self):
 
     domains = pydantic_api.infer_domain_from_model(ModelForCategorical)
     df = pydantic_api.models_to_dataframe(real_data, domains)
-    synth = data_generation_v3.TabularSynthesizer(
-        domains=domains,
-    ).calibrate(epsilon=epsilon, delta=delta)
+    synth = data_generation_v3.TabularConfig().calibrate(
+        domains, epsilon=epsilon, delta=delta
+    )
     synthetic_df = synth(np.random.default_rng(), df).synthetic_data
     synthetic_records = pydantic_api.dataframe_to_models(
         synthetic_df, ModelForCategorical, domains
diff --git a/tests/data_generation_v3_test.py b/tests/data_generation_v3_test.py
index 9a3c3dc7..d2c76cb5 100644
--- a/tests/data_generation_v3_test.py
+++ b/tests/data_generation_v3_test.py
@@ -374,30 +374,39 @@ def _categorical_domains(self):
 
   def test_configure_propagates_k_to_submechanisms(self):
     k = 5
-    config = TabularConfig(domains=self._categorical_domains())
-    calibrated = config.configure(zcdp_rho=100.0, max_records_per_user=k)
+    config = TabularConfig()
+    calibrated = config.configure(
+        self._categorical_domains(), zcdp_rho=100.0, max_records_per_user=k
+    )
     self.assertEqual(calibrated.max_records_per_user, k)
     self.assertEqual(calibrated.base_mechanism.max_records_per_user, k)
 
   def test_dp_event_invariant_to_k(self):
-    config = TabularConfig(domains=self._categorical_domains())
-    calibrated1 = config.configure(zcdp_rho=100.0)
-    calibrated2 = config.configure(zcdp_rho=100.0, max_records_per_user=5)
+    config = TabularConfig()
+    domains = self._categorical_domains()
+    calibrated1 = config.configure(domains, zcdp_rho=100.0)
+    calibrated2 = config.configure(
+        domains, zcdp_rho=100.0, max_records_per_user=5
+    )
     self.assertEqual(repr(calibrated1.dp_event), repr(calibrated2.dp_event))
 
   def test_end_to_end_with_k(self):
     df = pd.DataFrame({'A': ['a', 'b', 'c'], 'B': [1.0, 5.0, 10.0]})
-    config = TabularConfig(domains=self._categorical_domains())
-    calibrated = config.configure(zcdp_rho=100.0, max_records_per_user=3)
+    config = TabularConfig()
+    calibrated = config.configure(
+        self._categorical_domains(), zcdp_rho=100.0, max_records_per_user=3
+    )
     synthetic_df = calibrated(np.random.default_rng(0), df).synthetic_data
     self.assertListEqual(synthetic_df.columns.tolist(), ['A', 'B'])
 
   def test_open_set_with_k_supported(self):
     df = pd.DataFrame({'A': ['a', 'b', 'c', 'a', 'b', 'a'] * 5})
     domains = {'A': domain.OpenSetCategoricalAttribute()}
-    base = TabularConfig(domains=domains).configure(zcdp_rho=100.0, delta=1e-5)
-    config = TabularConfig(domains=domains)
-    mech = config.configure(zcdp_rho=100.0, delta=1e-5, max_records_per_user=3)
+    base = TabularConfig().configure(domains, zcdp_rho=100.0, delta=1e-5)
+    config = TabularConfig()
+    mech = config.configure(
+        domains, zcdp_rho=100.0, delta=1e-5, max_records_per_user=3
+    )
     # Accounting is byte-identical across k; only the injected noise scales.
     self.assertEqual(repr(mech.dp_event), repr(base.dp_event))
     synthetic_df = mech(np.random.default_rng(0), df).synthetic_data
@@ -406,16 +415,20 @@ def test_open_set_with_k_supported(self):
   def test_custom_initializers_inherit_k(self):
     domains = self._categorical_domains()
     inits = data_generation_v3.create_initializers(domains, 32)
-    config = TabularConfig(domains=domains, initializers=inits)
-    calibrated = config.configure(zcdp_rho=100.0, max_records_per_user=2)
-    for init in calibrated.initializers.values():
+    calibrated = {
+        col: init.configure(domains[col], zcdp_rho=50.0, max_records_per_user=2)
+        for col, init in inits.items()
+    }
+    for init in calibrated.values():
       self.assertEqual(init.max_records_per_user, 2)
 
   @parameterized.named_parameters(('zero', 0), ('negative', -3))
   def test_invalid_k_raises(self, k):
-    config = TabularConfig(domains=self._categorical_domains())
+    config = TabularConfig()
     with self.assertRaises(Exception):
-      _ = config.configure(zcdp_rho=100.0, max_records_per_user=k)
+      _ = config.configure(
+          self._categorical_domains(), zcdp_rho=100.0, max_records_per_user=k
+      )
 
   def test_poisson_calibrate_with_categorical_domains_and_gdp_mech(self):
     domains = {
@@ -423,10 +436,10 @@ def test_poisson_calibrate_with_categorical_domains_and_gdp_mech(self):
         'B': domain.CategoricalAttribute(possible_values=['x', 'y', 'z']),
     }
     config = TabularConfig(
-        domains=domains,
         discrete_mechanism=discrete_mechanisms.IndependentConfig(),
     )
     mechanism = config.calibrate(
+        domains,
         epsilon=1.0,
         delta=1e-6,
         poisson_sampling_prob=0.1,
@@ -439,9 +452,10 @@ def test_poisson_calibrate_with_mixed_domains(self):
         'B': domain.NumericalAttribute(min_value=0, max_value=10),
         'C': domain.OpenSetCategoricalAttribute(),
     }
-    config = TabularConfig(domains=domains)
+    config = TabularConfig()
     with self.assertRaises(dp_accounting.UnsupportedEventError):
       _ = config.calibrate(
+          domains,
           epsilon=1.0,
           delta=1e-6,
           poisson_sampling_prob=0.1,
@@ -457,10 +471,6 @@ def test_configure_infinite_zcdp_rho(self):
     self.assertIsNotNone(mechanism)
     self.assertEqual(mechanism.total_count_sigma, 0.0)
 
-
-if __name__ == '__main__':
-  absltest.main()
-
   def test_tabular_synthesizer_deprecated(self):
     with self.assertWarnsRegex(
         DeprecationWarning,
@@ -468,3 +478,7 @@ def test_tabular_synthesizer_deprecated(self):
         'and TabularMechanism for the calibrated runnable mechanism.',
     ):
       data_generation_v3.TabularSynthesizer(domains={})
+
+
+if __name__ == '__main__':
+  absltest.main()
diff --git a/tests/domain_test.py b/tests/domain_test.py
index 6e98455c..57004d6e 100644
--- a/tests/domain_test.py
+++ b/tests/domain_test.py
@@ -15,6 +15,7 @@
 import math
 
 from absl.testing import absltest
+from dpsynth import constraints
 from dpsynth import domain
 import numpy as np
 
@@ -191,9 +192,9 @@ def test_freeform_text_yaml_roundtrip(self):
     loaded_domain = domain.from_yaml_file(temp_file.full_path)
     self.assertEqual(loaded_domain, original_domain)
 
-  def test_freeform_text_yaml_backward_compatibility(self):
-    """A YAML file written with fewer fields can still be loaded."""
-    yaml_content = 'text:\n  max_tokens: 64\n'
+  def test_freeform_text_yaml_loading(self):
+    """A YAML file written with optional fields omitted can still be loaded."""
+    yaml_content = 'text:\n  type: FreeFormTextAttribute\n  max_tokens: 64\n'
     temp_file = self.create_tempfile('compat.yaml', content=yaml_content)
     loaded = domain.from_yaml_file(temp_file.full_path)
     self.assertEqual(
@@ -211,6 +212,35 @@ def test_open_set_yaml_roundtrip(self):
     loaded_domain = domain.from_yaml_file(temp_file.full_path)
     self.assertEqual(loaded_domain, original_domain)
 
+  def test_schema_mapping_interface(self):
+    attrs = {
+        'age': domain.NumericalAttribute(min_value=0, max_value=100),
+        'gender': domain.CategoricalAttribute(possible_values=['M', 'F']),
+    }
+    schema = domain.Schema(attributes=attrs)
+    self.assertEqual(schema['age'], attrs['age'])
+    self.assertIn('gender', schema)
+    self.assertNotIn('unknown', schema)
+    self.assertLen(schema, 2)
+    self.assertEqual(list(schema), ['age', 'gender'])
+    self.assertEqual(schema.get('age'), attrs['age'])
+    self.assertIsNone(schema.get('unknown'))
+
+  def test_schema_with_constraints(self):
+    attrs = {
+        'age': domain.NumericalAttribute(min_value=0, max_value=100),
+        'gender': domain.CategoricalAttribute(possible_values=['M', 'F']),
+    }
+    c = constraints.Constraint(
+        attribute_names=('gender',),
+        attribute_domains=(attrs['gender'],),
+        impossible_combinations=[('X',)],
+    )
+    schema = domain.Schema(attributes=attrs, constraints=(c,))
+    self.assertEqual(schema.attributes, attrs)
+    self.assertEqual(schema.constraints, (c,))
+    self.assertEqual(schema['gender'], attrs['gender'])
+
 
 if __name__ == '__main__':
   absltest.main()
diff --git a/tests/examples/relational/california_census/domain_test.py b/tests/examples/relational/california_census/domain_test.py
index c10762ee..0f180879 100644
--- a/tests/examples/relational/california_census/domain_test.py
+++ b/tests/examples/relational/california_census/domain_test.py
@@ -51,17 +51,15 @@ def setUp(self):
     self.domain_path = _find_domain_path()
 
   def test_load_california_census_domain(self):
-    table_domains, foreign_keys = rel_domain.from_yaml_file(
-        str(self.domain_path)
-    )
+    schema = rel_domain.from_yaml_file(str(self.domain_path))
 
     # Validate table set
     self.assertCountEqual(
-        list(table_domains.keys()), ['household', 'individual']
+        list(schema.tables.keys()), ['household', 'individual']
     )
 
     # Validate household attributes
-    household_schema = table_domains['household']
+    household_schema = schema.tables['household']
     self.assertLen(household_schema, 10)
     self.assertIsInstance(household_schema['FARM'], domain.CategoricalAttribute)
     self.assertEqual(household_schema['FARM'].size, 2)
@@ -72,7 +70,7 @@ def test_load_california_census_domain(self):
     self.assertEqual(household_schema['PROPINSR'].max_value, 59.0)
 
     # Validate individual attributes
-    individual_schema = table_domains['individual']
+    individual_schema = schema.tables['individual']
     self.assertLen(individual_schema, 15)
     self.assertIsInstance(
         individual_schema['RELATE'], domain.CategoricalAttribute
@@ -83,8 +81,8 @@ def test_load_california_census_domain(self):
     self.assertEqual(individual_schema['AGE'].max_value, 85.0)
 
     # Validate foreign keys
-    self.assertLen(foreign_keys, 1)
-    fk = foreign_keys[0]
+    self.assertLen(schema.foreign_keys, 1)
+    fk = schema.foreign_keys[0]
     self.assertEqual(fk.parent_table, 'household')
     self.assertEqual(fk.parent_primary_key, 'HOUSEHOLD')
     self.assertEqual(fk.child_table, 'individual')
@@ -93,7 +91,7 @@ def test_load_california_census_domain(self):
 
     # Validate topological hierarchy
     hierarchy = rel_domain.topological_sort_hierarchy(
-        list(table_domains.keys()), foreign_keys
+        list(schema.tables.keys()), schema.foreign_keys
     )
     self.assertEqual(
         hierarchy,
diff --git a/tests/examples/relational/california_census/synthesis_integration_test.py b/tests/examples/relational/california_census/synthesis_integration_test.py
index 2cf045d8..e3a82045 100644
--- a/tests/examples/relational/california_census/synthesis_integration_test.py
+++ b/tests/examples/relational/california_census/synthesis_integration_test.py
@@ -109,23 +109,21 @@ class CaliforniaCensusSynthesisIntegrationTest(parameterized.TestCase):
   def setUp(self):
     super().setUp()
     domain_path = _find_domain_path()
-    self.table_domains, self.foreign_keys = rel_domain.from_yaml_file(
-        str(domain_path)
-    )
+    self.schema = rel_domain.from_yaml_file(str(domain_path))
 
   def test_california_census_pipeline_e2e_mst(self):
     rng = np.random.default_rng(12345)
     data = _generate_mock_california_data(num_households=50, rng=rng)
 
     config = rel_synth.MultiTableConfig(
-        domains=self.table_domains,
-        foreign_keys=self.foreign_keys,
         discrete_mechanism=discrete_mechanisms.MSTConfig(pgm_iters=10),
         num_permutation_slots=1,
         exploration_strategy='empty_token',
         numerical_bins=2,
     )
-    calibrated_mechanism = config.calibrate(epsilon=3.2, delta=1e-6)
+    calibrated_mechanism = config.calibrate(
+        self.schema, epsilon=3.2, delta=1e-6
+    )
 
     # Synthesize
     result = calibrated_mechanism(rng=rng, data=data)
diff --git a/tests/local_mode/initialization_test.py b/tests/local_mode/initialization_test.py
index 95818e64..7e498936 100644
--- a/tests/local_mode/initialization_test.py
+++ b/tests/local_mode/initialization_test.py
@@ -27,7 +27,7 @@ class InitializationTest(absltest.TestCase):
   def test_numerical_initializer_dp_event(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     event = initializer.configure(zcdp_rho=1.0).dp_event
     self.assertIsInstance(event, dp_accounting.ComposedDpEvent)
@@ -39,7 +39,7 @@ def test_numerical_initializer_call(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10)
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
 
     data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
@@ -63,7 +63,7 @@ def test_numerical_initializer_deduplicates_bin_edges(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=100)
     rng = np.random.default_rng(42)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=8, attribute=attr
+        num_partitions=8, attribute=attr
     )
     # Data is heavily concentrated at 50.
     data = np.array([50] * 100 + [1, 99])
@@ -84,7 +84,7 @@ def test_numerical_initializer_integer_data(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10, dtype='int')
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=8, attribute=attr
+        num_partitions=8, attribute=attr
     )
     # Only 3 distinct values but 8 partitions requested.
     data = np.array([3, 3, 3, 3, 5, 5, 5, 7])
@@ -103,7 +103,7 @@ def test_numerical_initializer_integer_edges_are_floored(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=100, dtype='int')
     rng = np.random.default_rng(42)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     data = np.arange(100)
     result = initializer.configure(zcdp_rho=100.0)(rng, data)
@@ -118,7 +118,7 @@ def test_numerical_initializer_measurement_with_merged_bins(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=100, dtype='int')
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=8, attribute=attr
+        num_partitions=8, attribute=attr
     )
     # Concentrated data will cause edge collisions.
     data = np.array([50] * 100 + [1, 99])
@@ -136,7 +136,7 @@ def test_max_grid_size_below_two_raises(self):
     for bad in (0, 1):
       with self.assertRaises(ValueError):
         initialization.NumericalInitializerConfig(
-            name='x', num_partitions=10, attribute=attr, max_grid_size=bad
+            num_partitions=10, attribute=attr, max_grid_size=bad
         ).configure(zcdp_rho=1.0).configure(zcdp_rho=1.0).configure(
             zcdp_rho=1.0
         ).configure(
@@ -151,7 +151,7 @@ def test_int_grid_reserves_budget_for_jitter_refinement(self):
     )
     max_grid_size = 100_000
     init = initialization.NumericalInitializerConfig(
-        name='x', num_partitions=64, attribute=attr, max_grid_size=max_grid_size
+        num_partitions=64, attribute=attr, max_grid_size=max_grid_size
     )
     m = _quantiles.jitter_factor(init.num_partitions)
     self.assertLessEqual(
@@ -162,7 +162,7 @@ def test_numerical_initializer_measurement_with_estimated_total(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10)
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='num_col', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
     result = initializer.configure(zcdp_rho=1.0)(
@@ -177,7 +177,7 @@ def test_numerical_initializer_measurement_with_estimated_total(self):
         result.measurement.noisy_measurement,
         np.full(num_bins, expected_count),
     )
-    self.assertEqual(result.measurement.clique, ('num_col',))
+    self.assertEqual(result.measurement.clique, ())
     # stddev should be 1/sqrt(rho) = 1.0 in count space.
     self.assertAlmostEqual(result.measurement.stddev, 1.0)
 
@@ -185,7 +185,7 @@ def test_numerical_initializer_no_measurement_without_estimated_total(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10)
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
     result = initializer.configure(zcdp_rho=1.0)(rng, data)
@@ -196,7 +196,7 @@ def test_integer_edges_at_max_value_absorbed_into_last_bin(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10, dtype='int')
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=8, attribute=attr
+        num_partitions=8, attribute=attr
     )
     # A spread of lower values carrying most of the mass, plus a moderate spike
     # at max_value.  The lower values form genuine interior bins, while the top
@@ -223,7 +223,7 @@ def test_bin_weights_sum_to_num_partitions(self):
     for seed in range(10):
       rng = np.random.default_rng(seed)
       initializer = initialization.NumericalInitializerConfig(
-          name='test', num_partitions=8, attribute=attr
+          num_partitions=8, attribute=attr
       )
       data = np.array([5] * 50 + [15] * 50)
       result = initializer.configure(zcdp_rho=1.0)(
@@ -242,7 +242,7 @@ def test_integer_jitter_prevents_spurious_splits(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=100, dtype='int')
     rng = np.random.default_rng(42)
     initializer = initialization.NumericalInitializerConfig(
-        name='test', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     # Uniform data: with high budget, edges should land at 25, 50, 75.
     data = np.arange(101)
@@ -257,7 +257,7 @@ def test_integer_heterogeneous_data_buckets(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10, dtype='int')
     rng = np.random.default_rng(42)
     initializer = initialization.NumericalInitializerConfig(
-        name='x', num_partitions=4, attribute=attr
+        num_partitions=4, attribute=attr
     )
     # Deliberately lumpy distribution: 45 points across 4 distinct values.
     data = np.array([0] * 10 + [3] * 10 + [5] * 17 + [6] * 8)
@@ -385,7 +385,7 @@ def test_measurement_approximates_true_histogram(
   ):
     rng = np.random.default_rng(0)
     initializer = initialization.NumericalInitializerConfig(
-        name='x', num_partitions=num_partitions, attribute=attr
+        num_partitions=num_partitions, attribute=attr
     )
     result = initializer.configure(zcdp_rho=rho)(
         rng, data, estimated_total=len(data)
@@ -455,7 +455,7 @@ def test_measurement_property_random_configs(self):
         # -- Run initializer --
         rng = np.random.default_rng(trial)
         initializer = initialization.NumericalInitializerConfig(
-            name='x', num_partitions=num_partitions, attribute=attr
+            num_partitions=num_partitions, attribute=attr
         )
         result = initializer.configure(zcdp_rho=rho)(
             rng, data, estimated_total=len(data)
@@ -500,9 +500,7 @@ class CategoricalInitializerTest(absltest.TestCase):
 
   def test_dp_event(self):
     attr = domain.CategoricalAttribute(possible_values=['A', 'B', 'C'])
-    initializer = initialization.CategoricalInitializerConfig(
-        name='test', attribute=attr
-    )
+    initializer = initialization.CategoricalInitializerConfig(attribute=attr)
     event = initializer.configure(zcdp_rho=0.5).dp_event
     self.assertIsInstance(event, dp_accounting.GaussianDpEvent)
     # rho = 0.5 => sigma = 1/sqrt(2*0.5) = 1.0
@@ -511,9 +509,7 @@ def test_dp_event(self):
   def test_call_noiseless(self):
     attr = domain.CategoricalAttribute(possible_values=['A', 'B', 'C'])
     rng = np.random.default_rng(0)
-    initializer = initialization.CategoricalInitializerConfig(
-        name='col', attribute=attr
-    )
+    initializer = initialization.CategoricalInitializerConfig(attribute=attr)
     data = np.array(['A', 'A', 'B', 'C', 'C', 'C'])
     result = initializer.configure(zcdp_rho=np.inf)(rng, data)
 
@@ -523,7 +519,7 @@ def test_call_noiseless(self):
     np.testing.assert_array_equal(
         result.measurement.noisy_measurement, [2, 1, 3]
     )
-    self.assertEqual(result.measurement.clique, ('col',))
+    self.assertEqual(result.measurement.clique, ())
     self.assertEqual(result.measurement.stddev, 0.0)
 
   def test_out_of_domain_values(self):
@@ -531,9 +527,7 @@ def test_out_of_domain_values(self):
         possible_values=['<OOD>', 'X', 'Y'], out_of_domain_index=0
     )
     rng = np.random.default_rng(0)
-    initializer = initialization.CategoricalInitializerConfig(
-        name='col', attribute=attr
-    )
+    initializer = initialization.CategoricalInitializerConfig(attribute=attr)
     data = np.array(['X', 'Y', 'Z', 'W'])
     result = initializer.configure(zcdp_rho=np.inf)(rng, data)
 
@@ -548,7 +542,6 @@ class OpenSetCategoricalInitializerTest(absltest.TestCase):
   def test_dp_event(self):
     attr = domain.OpenSetCategoricalAttribute(default_value='<OOD>')
     initializer = initialization.OpenSetInitializerConfig(
-        name='test',
         attribute=attr,
     )
     event = initializer.configure(zcdp_rho=0.5, delta=1e-5).dp_event
@@ -564,7 +557,6 @@ def test_call_noiseless(self):
     attr = domain.OpenSetCategoricalAttribute(default_value='<OOD>')
     rng = np.random.default_rng(42)
     initializer = initialization.OpenSetInitializerConfig(
-        name='col',
         attribute=attr,
     )
     # 'A' appears 100 times, 'B' 50, 'C' 1 (rare).
@@ -586,7 +578,6 @@ def test_undiscovered_values_map_to_default(self):
     attr = domain.OpenSetCategoricalAttribute(default_value='OTHER')
     rng = np.random.default_rng(0)
     initializer = initialization.OpenSetInitializerConfig(
-        name='col',
         attribute=attr,
     )
     data = np.array(['A'] * 100 + ['B'] * 50)
@@ -605,7 +596,6 @@ def test_empty_data(self):
     attr = domain.OpenSetCategoricalAttribute(default_value='<OOD>')
     rng = np.random.default_rng(0)
     initializer = initialization.OpenSetInitializerConfig(
-        name='col',
         attribute=attr,
     )
     data = np.array([], dtype=str)
@@ -622,7 +612,6 @@ class NumericalInitializerFromSummaryTest(absltest.TestCase):
   def test_calibrate_sets_dp_event(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=100)
     init = initialization.NumericalInitializerConfig(
-        name='age',
         num_partitions=4,
         max_grid_size=10001,
         attribute=attr,
@@ -636,7 +625,6 @@ def test_integer_attribute_snaps_edges(self):
     rng = np.random.default_rng(42)
     attr = domain.NumericalAttribute(min_value=0, max_value=10, dtype='int')
     init = initialization.NumericalInitializerConfig(
-        name='count',
         num_partitions=4,
         attribute=attr,
     ).configure(zcdp_rho=1.0)
@@ -652,7 +640,6 @@ def test_call_and_from_summary_produce_same_structure(self):
     attr = domain.NumericalAttribute(min_value=0.0, max_value=100.0)
     max_grid_size = 10001
     init = initialization.NumericalInitializerConfig(
-        name='x',
         num_partitions=4,
         attribute=attr,
         max_grid_size=max_grid_size,
@@ -677,10 +664,10 @@ def test_categorical_stddev_scales_with_k(self):
     attr = domain.CategoricalAttribute(possible_values=['a', 'b', 'c'])
     data = np.array(['a', 'b', 'c', 'a'])
     base = initialization.CategoricalInitializerConfig(
-        name='x', attribute=attr
+        attribute=attr
     ).configure(zcdp_rho=1.0)
     scaled = initialization.CategoricalInitializerConfig(
-        name='x', attribute=attr
+        attribute=attr
     ).configure(zcdp_rho=1.0, max_records_per_user=4)
     b = base(np.random.default_rng(0), data)
     s = scaled(np.random.default_rng(0), data)
@@ -690,18 +677,16 @@ def test_numerical_raises_with_multiple_records_per_user(self):
     attr = domain.NumericalAttribute(min_value=0, max_value=10)
     with self.assertRaises(NotImplementedError):
       _ = initialization.NumericalInitializerConfig(
-          name='x', num_partitions=4, attribute=attr
+          num_partitions=4, attribute=attr
       ).configure(zcdp_rho=1.0, max_records_per_user=4)
 
   def test_open_set_stddev_scales_with_k(self):
     attr = domain.OpenSetCategoricalAttribute()
     data = np.array(['a'] * 50 + ['b'] * 40 + ['c'] * 30)
     base = initialization.OpenSetInitializerConfig(
-        name='x',
         attribute=attr,
     ).configure(zcdp_rho=1.0, delta=1e-5)
     scaled = initialization.OpenSetInitializerConfig(
-        name='x',
         attribute=attr,
     ).configure(zcdp_rho=1.0, delta=1e-5, max_records_per_user=4)
     b = base(np.random.default_rng(0), data)
@@ -712,9 +697,9 @@ def test_open_set_stddev_scales_with_k(self):
   def test_invalid_k_raises(self, k):
     attr = domain.CategoricalAttribute(possible_values=['a', 'b'])
     with self.assertRaises(ValueError):
-      initialization.CategoricalInitializerConfig(
-          name='x', attribute=attr
-      ).configure(zcdp_rho=0.5, max_records_per_user=k)
+      initialization.CategoricalInitializerConfig(attribute=attr).configure(
+          zcdp_rho=0.5, max_records_per_user=k
+      )
 
   def test_open_set_public_possible_values_retained(self):
     attr = domain.OpenSetCategoricalAttribute(
@@ -724,7 +709,7 @@ def test_open_set_public_possible_values_retained(self):
     # PUB1 and PUB2 have 0 occurrences but should still appear.
     data = np.array(['a'] * 100 + ['b'] * 1)
     init = initialization.OpenSetInitializerConfig(
-        name='x', attribute=attr, min_count=5
+        attribute=attr, min_count=5
     ).configure(zcdp_rho=1.0, delta=1e-6)
     col_meas = init(np.random.default_rng(0), data)
     cat_attr = col_meas.categorical_attribute
diff --git a/tests/relational/domain_test.py b/tests/relational/domain_test.py
index 20aaf3ad..473e8d67 100644
--- a/tests/relational/domain_test.py
+++ b/tests/relational/domain_test.py
@@ -154,22 +154,23 @@ def test_from_dict_valid_3tier_schema(self):
             'max_children_per_parent': 3,
         }],
     }
-    table_domains, fks = domain.from_dict(config)
-    self.assertIn('households', table_domains)
-    self.assertIn('persons', table_domains)
+    schema = domain.from_dict(config)
+    self.assertIsInstance(schema, domain.RelationalSchema)
+    self.assertIn('households', schema.tables)
+    self.assertIn('persons', schema.tables)
     self.assertIsInstance(
-        table_domains['households']['income'], base_domain.NumericalAttribute
+        schema.tables['households']['income'], base_domain.NumericalAttribute
     )
     self.assertIsInstance(
-        table_domains['households']['region'],
+        schema.tables['households']['region'],
         base_domain.CategoricalAttribute,
     )
-    self.assertLen(fks, 1)
-    self.assertEqual(fks[0].parent_table, 'households')
-    self.assertEqual(fks[0].parent_primary_key, 'household_id')
-    self.assertEqual(fks[0].child_table, 'persons')
-    self.assertEqual(fks[0].child_foreign_key, 'household_id')
-    self.assertEqual(fks[0].max_children_per_parent, 3)
+    self.assertLen(schema.foreign_keys, 1)
+    self.assertEqual(schema.foreign_keys[0].parent_table, 'households')
+    self.assertEqual(schema.foreign_keys[0].parent_primary_key, 'household_id')
+    self.assertEqual(schema.foreign_keys[0].child_table, 'persons')
+    self.assertEqual(schema.foreign_keys[0].child_foreign_key, 'household_id')
+    self.assertEqual(schema.foreign_keys[0].max_children_per_parent, 3)
 
   def test_from_dict_missing_tables_block_raises_error(self):
     with self.assertRaisesRegex(ValueError, "'tables' block missing"):
@@ -241,21 +242,22 @@ def test_from_yaml_file_roundtrip(self):
             max_children_per_parent: 4
         """)
     tmp_path = self.create_tempfile(content=yaml_content).full_path
-    table_domains, fks = domain.from_yaml_file(tmp_path)
-    self.assertIn('households', table_domains)
-    self.assertIn('persons', table_domains)
+    schema = domain.from_yaml_file(tmp_path)
+    self.assertIsInstance(schema, domain.RelationalSchema)
+    self.assertIn('households', schema.tables)
+    self.assertIn('persons', schema.tables)
     self.assertIsInstance(
-        table_domains['persons']['age'], base_domain.NumericalAttribute
+        schema.tables['persons']['age'], base_domain.NumericalAttribute
     )
     self.assertIsInstance(
-        table_domains['persons']['gender'], base_domain.CategoricalAttribute
+        schema.tables['persons']['gender'], base_domain.CategoricalAttribute
     )
-    self.assertLen(fks, 1)
-    self.assertEqual(fks[0].parent_table, 'households')
-    self.assertEqual(fks[0].parent_primary_key, 'hid')
-    self.assertEqual(fks[0].child_table, 'persons')
-    self.assertEqual(fks[0].child_foreign_key, 'hid')
-    self.assertEqual(fks[0].max_children_per_parent, 4)
+    self.assertLen(schema.foreign_keys, 1)
+    self.assertEqual(schema.foreign_keys[0].parent_table, 'households')
+    self.assertEqual(schema.foreign_keys[0].parent_primary_key, 'hid')
+    self.assertEqual(schema.foreign_keys[0].child_table, 'persons')
+    self.assertEqual(schema.foreign_keys[0].child_foreign_key, 'hid')
+    self.assertEqual(schema.foreign_keys[0].max_children_per_parent, 4)
 
   def test_to_dict_and_roundtrip(self):
     table_domains = {
@@ -295,16 +297,16 @@ def test_to_dict_and_roundtrip(self):
     )
 
     # Roundtrip verification
-    rt_domains, rt_fks = domain.from_dict(serialized)
+    rt_schema = domain.from_dict(serialized)
     self.assertEqual(
-        rt_domains['households']['income'].min_value,
+        rt_schema.tables['households']['income'].min_value,
         table_domains['households']['income'].min_value,
     )
     self.assertEqual(
-        rt_domains['persons']['gender'].possible_values,
+        rt_schema.tables['persons']['gender'].possible_values,
         table_domains['persons']['gender'].possible_values,
     )
-    self.assertEqual(rt_fks, fks)
+    self.assertEqual(list(rt_schema.foreign_keys), fks)
 
   def test_to_yaml_file_and_roundtrip(self):
     table_domains = {
@@ -332,13 +334,16 @@ def test_to_yaml_file_and_roundtrip(self):
         )
     ]
     tmp_path = self.create_tempfile().full_path
-    domain.to_yaml_file(table_domains, fks, tmp_path)
+    schema = domain.RelationalSchema(tables=table_domains, foreign_keys=fks)
+    domain.to_yaml_file(schema, tmp_path)
 
-    rt_domains, rt_fks = domain.from_yaml_file(tmp_path)
-    self.assertIn('households', rt_domains)
-    self.assertIn('persons', rt_domains)
-    self.assertEqual(rt_domains['households']['income'].max_value, 150000.0)
-    self.assertEqual(rt_fks, fks)
+    rt_schema = domain.from_yaml_file(tmp_path)
+    self.assertIn('households', rt_schema.tables)
+    self.assertIn('persons', rt_schema.tables)
+    self.assertEqual(
+        rt_schema.tables['households']['income'].max_value, 150000.0
+    )
+    self.assertEqual(list(rt_schema.foreign_keys), fks)
 
   def test_to_dict_without_foreign_keys(self):
     table_domains = {
@@ -351,9 +356,9 @@ def test_to_dict_without_foreign_keys(self):
     serialized = domain.to_dict(table_domains)
     self.assertIn('tables', serialized)
     self.assertNotIn('foreign_keys', serialized)
-    rt_domains, rt_fks = domain.from_dict(serialized)
-    self.assertIn('single_table', rt_domains)
-    self.assertEmpty(rt_fks)
+    rt_schema = domain.from_dict(serialized)
+    self.assertIn('single_table', rt_schema.tables)
+    self.assertEmpty(rt_schema.foreign_keys)
 
 
 if __name__ == '__main__':
diff --git a/tests/relational/synthesizer_test.py b/tests/relational/synthesizer_test.py
index 0923bc51..685900c9 100644
--- a/tests/relational/synthesizer_test.py
+++ b/tests/relational/synthesizer_test.py
@@ -136,14 +136,12 @@ def test_dp_event_composition(self):
     household_inits = {
         'income': (
             initialization.NumericalInitializerConfig(
-                name='income',
                 num_partitions=16,
                 attribute=domain.NumericalAttribute(min_value=0, max_value=100),
             ).configure(zcdp_rho=0.01)
         ),
         'region': (
             initialization.CategoricalInitializerConfig(
-                name='region',
                 attribute=domain.CategoricalAttribute(
                     possible_values=['U', 'R']
                 ),
@@ -153,7 +151,6 @@ def test_dp_event_composition(self):
     person_inits = {
         'age': (
             initialization.NumericalInitializerConfig(
-                name='age',
                 num_partitions=16,
                 attribute=domain.NumericalAttribute(min_value=0, max_value=100),
             ).configure(zcdp_rho=0.01)
@@ -173,8 +170,8 @@ def test_dp_event_composition(self):
     }
 
     mech = synthesizer.MultiTableMechanism(
-        domains={},
-        foreign_keys=(),
+        config=synthesizer.MultiTableConfig(),
+        schema=rel_domain.RelationalSchema(tables={}, foreign_keys=()),
         calibrated_discrete_mechanisms=calibrated_discrete_mechanisms,
         calibrated_initializers=calibrated_initializers,
         total_count_sigma=5.0,
@@ -259,25 +256,27 @@ def test_compute_link_sensitivities_no_links(self):
     self.assertEmpty(sensitivities)
 
   def test_configure_single_table_raises(self):
+    schema = rel_domain.RelationalSchema(
+        tables={'Household': {'income': domain.NumericalAttribute(0, 100)}},
+        foreign_keys=(),
+    )
     with self.assertRaisesRegex(
-        ValueError, 'requires at least two tables in domains'
+        ValueError, 'requires at least two tables in schema'
     ):
-      synthesizer.MultiTableConfig(
-          domains={'Household': {'income': domain.NumericalAttribute(0, 100)}},
-          foreign_keys=(),
-      )
+      synthesizer.MultiTableConfig().configure(schema, zcdp_rho=0.5)
 
   def test_configure_empty_foreign_keys_raises(self):
+    schema = rel_domain.RelationalSchema(
+        tables={
+            'Household': {'income': domain.NumericalAttribute(0, 100)},
+            'Person': {'age': domain.NumericalAttribute(0, 100)},
+        },
+        foreign_keys=(),
+    )
     with self.assertRaisesRegex(
         ValueError, 'requires at least one foreign key relationship'
     ):
-      synthesizer.MultiTableConfig(
-          domains={
-              'Household': {'income': domain.NumericalAttribute(0, 100)},
-              'Person': {'age': domain.NumericalAttribute(0, 100)},
-          },
-          foreign_keys=(),
-      )
+      synthesizer.MultiTableConfig().configure(schema, zcdp_rho=0.5)
 
   def test_configure_end_to_end_3_tier(self):
     domains = {
@@ -309,9 +308,10 @@ def test_configure_end_to_end_3_tier(self):
             max_children_per_parent=2,
         ),
     ]
+    schema = rel_domain.RelationalSchema(
+        tables=domains, foreign_keys=foreign_keys
+    )
     config = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=foreign_keys,
         init_budget_fraction=0.1,
     )
 
@@ -319,7 +319,7 @@ def test_configure_end_to_end_3_tier(self):
     # init_rho = 0.1 * 0.6 = 0.06 => per_col_rho = 0.01.
     # total_count_sigma = sqrt(0.5 / 0.01) = sqrt(50).
     # discrete_rho = 0.6 - 0.06 = 0.54 => per_link_rho = 0.27 across 2 links.
-    mech = config.configure(zcdp_rho=0.6, max_records_per_user=1)
+    mech = config.configure(schema, zcdp_rho=0.6, max_records_per_user=1)
 
     self.assertIsInstance(mech, synthesizer.MultiTableMechanism)
     self.assertAlmostEqual(mech.total_count_sigma, math.sqrt(50.0))
@@ -353,12 +353,12 @@ def test_calibrate_end_to_end(self):
             max_children_per_parent=3,
         ),
     ]
-    config = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=foreign_keys,
+    schema = rel_domain.RelationalSchema(
+        tables=domains, foreign_keys=foreign_keys
     )
+    config = synthesizer.MultiTableConfig()
     # Calibrate solves for optimal zcdp_rho using PLD / RDP accountant.
-    mech = config.calibrate(epsilon=1.0, delta=1e-5)
+    mech = config.calibrate(schema, epsilon=1.0, delta=1e-5)
     self.assertIsInstance(mech, synthesizer.MultiTableMechanism)
     self.assertGreater(mech.total_count_sigma, 0.0)
     self.assertLen(mech.calibrated_discrete_mechanisms, 1)
@@ -377,21 +377,20 @@ def test_configure_hyperparameter_validations(self):
             max_children_per_parent=3,
         ),
     ]
+    schema = rel_domain.RelationalSchema(
+        tables=domains, foreign_keys=foreign_keys
+    )
 
     with self.subTest('negative_zcdp_rho'):
-      config = synthesizer.MultiTableConfig(
-          domains=domains, foreign_keys=foreign_keys
-      )
+      config = synthesizer.MultiTableConfig()
       with self.assertRaisesRegex(ValueError, 'zcdp_rho must be positive'):
-        config.configure(zcdp_rho=-0.1)
+        config.configure(schema, zcdp_rho=-0.1)
 
     with self.subTest('invalid_init_budget_fraction_above_one'):
       with self.assertRaisesRegex(
           ValueError, 'init_budget_fraction must be strictly in'
       ):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             init_budget_fraction=1.5,
         )
 
@@ -400,16 +399,12 @@ def test_configure_hyperparameter_validations(self):
           ValueError, 'init_budget_fraction must be strictly in'
       ):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             init_budget_fraction=0.0,
         )
 
     with self.subTest('invalid_numerical_bins'):
       with self.assertRaisesRegex(ValueError, 'numerical_bins must be >= 1'):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             numerical_bins=0,
         )
 
@@ -418,8 +413,6 @@ def test_configure_hyperparameter_validations(self):
           ValueError, 'num_permutation_slots must be >= 1'
       ):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             num_permutation_slots=0,
         )
 
@@ -428,8 +421,6 @@ def test_configure_hyperparameter_validations(self):
           ValueError, 'Unsupported exploration_strategy'
       ):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             exploration_strategy='unsupported_strategy',
         )
 
@@ -439,139 +430,126 @@ def test_configure_hyperparameter_validations(self):
           'discrete_mechanism must be an instance of MechanismConfig',
       ):
         synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
             discrete_mechanism='not_a_config',  # pyrefly: ignore[bad-argument-type]
         )
 
     with self.subTest('dot_in_table_name'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'House.hold': {'income': domain.NumericalAttribute(0, 100)},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=[
+              rel_domain.ForeignKeyRelation(
+                  parent_table='House.hold',
+                  parent_primary_key='hid',
+                  child_table='Person',
+                  child_foreign_key='hid',
+                  max_children_per_parent=2,
+              )
+          ],
+      )
       with self.assertRaisesRegex(ValueError, "must not contain '.'"):
-        synthesizer.MultiTableConfig(
-            domains={
-                'House.hold': {'income': domain.NumericalAttribute(0, 100)},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=[
-                rel_domain.ForeignKeyRelation(
-                    parent_table='House.hold',
-                    parent_primary_key='hid',
-                    child_table='Person',
-                    child_foreign_key='hid',
-                    max_children_per_parent=2,
-                )
-            ],
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('dot_in_column_name'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'inc.ome': domain.NumericalAttribute(0, 100)},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(ValueError, "must not contain '.'"):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'inc.ome': domain.NumericalAttribute(0, 100)},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('reserved_column_name_group_size'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'group_size': domain.NumericalAttribute(0, 100)},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(
           ValueError, 'reserved for relational exploration'
       ):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'group_size': domain.NumericalAttribute(0, 100)},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('reserved_column_name_slot_prefix'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'slot_1': domain.NumericalAttribute(0, 100)},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(ValueError, 'reserved for permutation slots'):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'slot_1': domain.NumericalAttribute(0, 100)},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('empty_table_schema'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(
           ValueError, 'schema in domains cannot be empty'
       ):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('unsupported_attribute_type'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'text': domain.FreeFormTextAttribute()},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(ValueError, 'unsupported attribute type'):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'text': domain.FreeFormTextAttribute()},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('multi_root_forest_raises'):
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'income': domain.NumericalAttribute(0, 100)},
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+              'Unlinked': {'type': domain.CategoricalAttribute(['A', 'B'])},
+          },
+          foreign_keys=foreign_keys,
+      )
       with self.assertRaisesRegex(ValueError, 'expects a single root table'):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'income': domain.NumericalAttribute(0, 100)},
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-                'Unlinked': {'type': domain.CategoricalAttribute(['A', 'B'])},
-            },
-            foreign_keys=foreign_keys,
-        )
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('pk_in_domain_schema_raises'):
-      with self.assertRaisesRegex(ValueError, 'must not be in domains'):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {
-                    'income': domain.NumericalAttribute(0, 100),
-                    'hid': domain.CategoricalAttribute(['H1', 'H2']),
-                },
-                'Person': {'age': domain.NumericalAttribute(0, 100)},
-            },
-            foreign_keys=foreign_keys,
-        )
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {
+                  'income': domain.NumericalAttribute(0, 100),
+                  'hid': domain.CategoricalAttribute(['H1', 'H2']),
+              },
+              'Person': {'age': domain.NumericalAttribute(0, 100)},
+          },
+          foreign_keys=foreign_keys,
+      )
+      with self.assertRaisesRegex(ValueError, 'must not be in'):
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
     with self.subTest('fk_in_domain_schema_raises'):
-      with self.assertRaisesRegex(ValueError, 'must not be in domains'):
-        synthesizer.MultiTableConfig(
-            domains={
-                'Household': {'income': domain.NumericalAttribute(0, 100)},
-                'Person': {
-                    'age': domain.NumericalAttribute(0, 100),
-                    'hid': domain.CategoricalAttribute(['H1', 'H2']),
-                },
-            },
-            foreign_keys=foreign_keys,
-        )
-
-    with self.subTest('custom_initializers_mismatched_tables'):
-      with self.assertRaisesRegex(ValueError, 'do not match domains tables'):
-        synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
-            initializers={'Household': {}},
-        )
-
-    with self.subTest('custom_initializers_mismatched_columns'):
-      mock_cfg = unittest.mock.MagicMock(spec=api.MechanismConfig)
-      with self.assertRaisesRegex(ValueError, 'do not match domains columns'):
-        synthesizer.MultiTableConfig(
-            domains=domains,
-            foreign_keys=foreign_keys,
-            initializers={
-                'Household': {'wrong_col': mock_cfg},
-                'Person': {'age': mock_cfg},
-            },
-        )
+      invalid_schema = rel_domain.RelationalSchema(
+          tables={
+              'Household': {'income': domain.NumericalAttribute(0, 100)},
+              'Person': {
+                  'age': domain.NumericalAttribute(0, 100),
+                  'hid': domain.CategoricalAttribute(['H1', 'H2']),
+              },
+          },
+          foreign_keys=foreign_keys,
+      )
+      with self.assertRaisesRegex(ValueError, 'must not be in'):
+        synthesizer.MultiTableConfig().configure(invalid_schema, zcdp_rho=0.5)
 
   def test_validate_input_table_columns_success(self):
     domains = {
@@ -738,7 +716,6 @@ def test_run_single_col_initializer(self):
 
     # 1. Numerical initializer on weighted data
     num_init = initialization.NumericalInitializerConfig(
-        name='income',
         num_partitions=8,
         attribute=domain.NumericalAttribute(min_value=0.0, max_value=100.0),
     ).configure(zcdp_rho=0.5)
@@ -750,11 +727,10 @@ def test_run_single_col_initializer(self):
     self.assertIsNotNone(res_num.bin_edges)
     self.assertIsNotNone(res_num.categorical_attribute)
     self.assertIsNotNone(res_num.measurement)
-    self.assertEqual(res_num.measurement.clique, ('income',))
+    self.assertEqual(res_num.measurement.clique, ())
 
     # 2. Categorical initializer on weighted data
     cat_init = initialization.CategoricalInitializerConfig(
-        name='gender',
         attribute=domain.CategoricalAttribute(possible_values=['M', 'F']),
     ).configure(zcdp_rho=0.5)
     data_cat = np.array(['M', 'M', 'F', 'F'])
@@ -764,11 +740,10 @@ def test_run_single_col_initializer(self):
     )
     self.assertIsNone(res_cat.bin_edges)
     self.assertEqual(res_cat.categorical_attribute.size, 2)
-    self.assertEqual(res_cat.measurement.clique, ('gender',))
+    self.assertEqual(res_cat.measurement.clique, ())
 
     # 3. Open-set initializer on weighted strings (including mixed-type data)
     open_init = initialization.OpenSetInitializerConfig(
-        name='tags',
         attribute=domain.OpenSetCategoricalAttribute(),
         min_count=1,
     ).configure(zcdp_rho=0.5, delta=1e-3)
@@ -792,7 +767,6 @@ def test_run_table_initializers(self):
         'Household': {
             'income': (
                 initialization.NumericalInitializerConfig(
-                    name='income',
                     num_partitions=8,
                     attribute=domain.NumericalAttribute(
                         min_value=0.0, max_value=100.0
@@ -801,7 +775,6 @@ def test_run_table_initializers(self):
             ),
             'region': (
                 initialization.CategoricalInitializerConfig(
-                    name='region',
                     attribute=domain.CategoricalAttribute(
                         possible_values=['U', 'R']
                     ),
@@ -811,7 +784,6 @@ def test_run_table_initializers(self):
         'Person': {
             'age': (
                 initialization.NumericalInitializerConfig(
-                    name='age',
                     num_partitions=8,
                     attribute=domain.NumericalAttribute(
                         min_value=0, max_value=100
@@ -960,12 +932,13 @@ def test_run_table_preprocessing_end_to_end(self):
             max_children_per_parent=3,
         ),
     ]
+    schema = rel_domain.RelationalSchema(
+        tables=domains, foreign_keys=foreign_keys
+    )
     config = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=foreign_keys,
         init_budget_fraction=0.2,
     )
-    mech = config.configure(zcdp_rho=0.5, max_records_per_user=1)
+    mech = config.configure(schema, zcdp_rho=0.5, max_records_per_user=1)
     rng = np.random.default_rng(42)
 
     data = {
@@ -1216,18 +1189,17 @@ def test_synthesize_relational_hierarchy_3_tier(self):
             max_children_per_parent=2,
         ),
     ]
+    schema = rel_domain.RelationalSchema(tables=domains, foreign_keys=fks)
     hierarchy = rel_domain.topological_sort_hierarchy(
         tables=list(domains.keys()), foreign_keys=fks
     )
     cfg = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=fks,
         discrete_mechanism=discrete_mechanisms.AIMConfig(
             pgm_iters=10, max_rounds=2
         ),
         num_permutation_slots=2,
     )
-    mech = cfg.configure(zcdp_rho=0.5, max_records_per_user=1)
+    mech = cfg.configure(schema, zcdp_rho=0.5, max_records_per_user=1)
 
     preprocessed = synthesizer._run_table_preprocessing(
         mechanism=mech,
@@ -1307,18 +1279,17 @@ def test_synthesize_relational_hierarchy_branching(self):
             max_children_per_parent=2,
         ),
     ]
+    schema = rel_domain.RelationalSchema(tables=domains, foreign_keys=fks)
     hierarchy = rel_domain.topological_sort_hierarchy(
         tables=list(domains.keys()), foreign_keys=fks
     )
     cfg = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=fks,
         discrete_mechanism=discrete_mechanisms.AIMConfig(
             pgm_iters=10, max_rounds=2
         ),
         num_permutation_slots=2,
     )
-    mech = cfg.configure(zcdp_rho=0.5, max_records_per_user=1)
+    mech = cfg.configure(schema, zcdp_rho=0.5, max_records_per_user=1)
 
     preprocessed = synthesizer._run_table_preprocessing(
         mechanism=mech,
@@ -1612,9 +1583,8 @@ def test_multi_table_mechanism_call_end_to_end(self):
         )
     ]
 
+    schema = rel_domain.RelationalSchema(tables=domains, foreign_keys=fks)
     cfg = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=fks,
         discrete_mechanism=discrete_mechanisms.AIMConfig(
             max_rounds=2,
             pgm_iters=50,
@@ -1623,6 +1593,7 @@ def test_multi_table_mechanism_call_end_to_end(self):
         init_budget_fraction=0.2,
     )
     mechanism = cfg.configure(
+        schema,
         zcdp_rho=1.0,
     )
 
@@ -1691,9 +1662,8 @@ def test_multi_table_mechanism_call_size_sliced(self):
         )
     ]
 
+    schema = rel_domain.RelationalSchema(tables=domains, foreign_keys=fks)
     cfg = synthesizer.MultiTableConfig(
-        domains=domains,
-        foreign_keys=fks,
         discrete_mechanism=discrete_mechanisms.AIMConfig(
             max_rounds=2,
             pgm_iters=50,
@@ -1703,6 +1673,7 @@ def test_multi_table_mechanism_call_size_sliced(self):
         exploration_strategy='size_sliced',
     )
     mechanism = cfg.configure(
+        schema,
         zcdp_rho=1.0,
     )