Infer a Starting Contract with Dry-Run Mode

Write a Contract walks through authoring one by hand, field by field. If you don’t have a contract yet and just want to see what your transformation actually reads and writes, dry-run mode does that for you: run your job with no contract at all, and Invaract prints one back — built from the job’s real inputs and outputs — that you can copy into a file, review, and use with the enforcement rule from then on.

On Apache Beam

Dry-run mode works on Beam too: --invaractDryRun=true (Java) or --invaract_dry_run=true (Python), and with check-only mode it reports the contract a pipeline needs without running it. See the Java and Python guides. This page shows the Spark side.

A starting point, not a finished contract

Dry-run mode infers structure — locations, formats, save mode, field names/types/ nullability — never business rules. It has no way to know your MERGE must always match on customer_id, or that a DELETE should never run unconditionally; add those by hand once you have a contract to add them to. Every inferred field is also marked required: true, since it genuinely was present in this run — review that against what you actually intend before treating it as a finished document.

Try it against the demo job

1. Run the demo with no contract supplied

   ./dev/dry-run

   This builds every module the same way ./dev/test does, then runs the demo job with --dry-run and no contract argument at all.

2. Read the inferred contract

   The run prints something like:

   Dry-run mode: no contract was supplied, so nothing was enforced.
   Inferred contract, from this run's actual inputs/outputs — copy it into a file, review it,
   and pass it as the contract argument to use the normal (enforced) mode:
   
   id: inferred_contract
   version: 0.1.0
   status: draft
   inputs:
   - name: input
     location: /home/user/.../demo/input/sample.csv
     schema:
       fields:
       - name: id
         type: integer
         required: true
         nullable: true
       - name: value
         type: integer
         required: true
         nullable: true
   outputs:
   - name: output
     location: /home/user/.../demo/output/dry_run_result.parquet
     format: parquet
     schema:
       fields:
       - name: id
         type: integer
         required: true
         nullable: true
       - name: value
         type: integer
         required: true
         nullable: true
       - name: value_squared
         type: integer
         required: true
         nullable: true
       - name: value_tier
         type: string
         required: true
         nullable: false
     saveMode: overwrite
     derivedFrom:
     - input

   derivedFrom names exactly the inputs this write was observed reading, so when you later merge several inferred drafts into one multi-output contract, each output keeps naming only the inputs that feed it (see Map Outputs to Their Inputs). A write that read nothing gets derivedFrom: []. Inferred input names are positional per write (input, input_1, input_2, …), so they mean different things in different drafts — merge them with MergeDraftsCli rather than by hand.

   location is the absolute file:-scheme-stripped path Spark actually reported for this run (see Contract Format’s “Location matching”) — replace it with a portable, relative path (demo/input/sample.csv) once you copy this into a real contract file. value_tier is inferred nullable: false because both branches of its CASE WHEN produce a non-null literal.

   Save that to a file (id, version, and status are placeholders — replace them once you know what this contract should really be called), and you have a real starting point for Write a Contract to refine.

Using it in your own job

Dry-run mode installs the same way as real enforcement — a SparkSessionExtensions check rule, at session construction time — just with no contract to enforce. Attach it the same way installing the enforcement rule itself does:

spark-submit \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.dryRun=true \
  --jars invaract-spark-adapter-0.11.0.jar \
  my-job.jar

spark.invaract.contract is ignored entirely in this mode — there’s no job code here to hand an inferred Contract to, so InvaractSparkSessionExtension logs it at WARN instead, the same information printed below:

WARN InvaractSparkSessionExtension: Invaract dry-run mode (spark.invaract.dryRun=true):
inferred contract from this session's real inputs/outputs - copy it into a file, review
it, and set spark.invaract.contract to it to switch to enforced mode:
id: inferred_contract
...

Report dry-run results to a sink, to roll Invaract out across many jobs

Logging a draft is enough for one job you’re looking at. To onboard a fleet — collect every job’s inferred contract in one place, and learn how safe it is to turn enforcement on for each — point dry-run mode at a notification sink by adding spark.invaract.notifyConfig, exactly as for enforcement. Nothing in the job changes:

spark-submit \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.dryRun=true \
  --conf spark.invaract.notifyConfig=/etc/invaract/dry-run.properties \
  --conf spark.invaract.jobId=orders_nightly \
  --conf spark.invaract.job.metadata.team=orders \
  --files /etc/invaract/dry-run.properties \
  --jars invaract-spark-adapter-0.11.0.jar \
  my-job.jar

With a sink configured, dry-run mode stops logging the draft and instead publishes two kinds of event:

• CONTRACT_INFERENCE, one per write dry-run mode met — including the writes it couldn’t infer a contract from, which plain logging never mentioned.
• DRY_RUN_SUMMARY, once, when the application ends — how many writes landed in each status, plus every draft from the run merged into one contract.

The statuses

status	Meaning	What to do with it
INFERRED	A contract was inferred and nothing about it is in doubt.	Candidate for your contract repository.
INFERRED_DEGRADED	A contract was inferred, but from a plan Invaract could only partly see: an unresolved .checkpoint()/cache boundary (inputs behind it are missing), a resolution that had to guess, or row-level DML (MERGE/UPDATE/DELETE, where the draft describes the target’s current schema and the operation’s own logic isn’t inferred). diagnostics says which.	Review before trusting.
SKIPPED_UNSUPPORTED	A state-changing procedure CALL. Dry-run mode infers a contract from a write’s output schema, and a procedure call has none.	Nothing to infer; enforcement would verify the resulting table state.
SKIPPED_UNRECOGNIZED	A command that may write data but that Invaract has no translation for. With a contract active, enforcement would reject it as an UNVERIFIABLE_WRITE.	A blocker for turning enforcement on for this job.
INFERENCE_ERROR	Inference itself threw. The job was unaffected.	Report it.

A job that ran but wrote nothing Invaract recognizes reports NO_WRITES_OBSERVED in the summary — the only place that outcome is visible, since there’s no write to attach an event to. Treat it as “not enough information”, not as “safe”.

The self-check

Every inferred draft is verified against the very write it came from, under the options enforcement would use (so your spark.invaract.rejectUndeclaredFields and friends count). The event’s selfCheck is PASSED when turning enforcement on with that contract would not have blocked this write, and FAILED (with selfCheckViolations), INVALID or ERROR otherwise. That is a better readiness signal than the count of successes: a draft that doesn’t pass its own write isn’t something to promote.

What’s in an event

Real events, from a real spark-sql run with the settings above (the file paths are the run’s own; the transformation’s fingerprints are trimmed from the first, see Fingerprint a Transformation for their shape). The first is a write dry-run mode inferred a contract for; its contractYaml is the draft itself, and selfCheck says it would pass enforcement:

{
  "eventType": "CONTRACT_INFERENCE",
  "timestamp": 1791097550747,
  "contractYaml": "id: inferred_contract\nversion: 0.1.0\nstatus: draft\ninputs:\n- name: input\n  location: /tmp/invaract-example/data/orders.csv\n  schema:\n    fields:\n    - name: id\n      type: integer\n      required: true\n      nullable: true\n    - name: value\n      type: integer\n      required: true\n      nullable: true\noutputs:\n- name: output\n  location: /tmp/invaract-example/out/orders_doubled\n  format: parquet\n  schema:\n    fields:\n    - name: id\n      type: integer\n      required: true\n      nullable: true\n    - name: doubled\n      type: integer\n      required: true\n      nullable: true\n  saveMode: overwrite\n  derivedFrom:\n  - input\nextensions:\n  source_system: crm\n  owner: orders-team\n",
  "job": {
    "engineDetails": {
      "master": "local[2]",
      "deployMode": "client"
    },
    "name": "SparkSQL::192.0.2.2",
    "startTimeMs": 1791097540432,
    "runId": "local-1791097541542",
    "engine": "spark",
    "engineVersion": "3.5.7",
    "attributes": {
      "run_id": "orders_daily__2026-10-04T02:00",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "user": "root"
  },
  "contractDigest": "29efd2c8fc03a026e644e4f15b096297618f42dcf8b7310bf21b26b98273b648",
  "selfCheckViolations": [],
  "reason": null,
  "eventId": "0fb8a74380802ca79d89483118bcec8c0a6d2e52972ff4d7a7d1c565a8934fc3",
  "fingerprints": {"...": "the transformation's fingerprint, trimmed here: see Fingerprint a Transformation"},
  "diagnostics": [],
  "status": "INFERRED",
  "selfCheck": "PASSED",
  "writeLocation": "/tmp/invaract-example/out/orders_doubled",
  "metadata": {
    "source_system": "crm",
    "owner": "orders-team"
  },
  "schemaVersion": 1
}

The second is a write it could not infer from. Here the job used INSERT OVERWRITE DIRECTORY, which Invaract has no translation for, so with a contract active enforcement would reject it:

{
  "eventType": "CONTRACT_INFERENCE",
  "timestamp": 1791097550611,
  "contractYaml": null,
  "job": {
    "engineDetails": {
      "master": "local[2]",
      "deployMode": "client"
    },
    "name": "SparkSQL::192.0.2.2",
    "startTimeMs": 1791097540432,
    "runId": "local-1791097541542",
    "engine": "spark",
    "engineVersion": "3.5.7",
    "attributes": {
      "run_id": "orders_daily__2026-10-04T02:00",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "user": "root"
  },
  "contractDigest": null,
  "selfCheckViolations": [],
  "reason": "'InsertIntoDataSourceDirCommand' may write data but Invaract has no translation for it; with a contract active, enforcement would reject it as an UNVERIFIABLE_WRITE",
  "eventId": "3b4f276765243defe0b23188f9be2bc9c65cd6d0413ed2b449cde8187aa1e11c",
  "fingerprints": null,
  "diagnostics": [],
  "status": "SKIPPED_UNRECOGNIZED",
  "selfCheck": null,
  "writeLocation": null,
  "metadata": {
    "source_system": "crm",
    "owner": "orders-team"
  },
  "schemaVersion": 1
}

And the summary published when the application ended:

{
  "mergeStatus": "MERGED",
  "mergeConflicts": [],
  "eventType": "DRY_RUN_SUMMARY",
  "timestamp": 1791097551727,
  "job": {
    "engineDetails": {
      "master": "local[2]",
      "deployMode": "client"
    },
    "name": "SparkSQL::192.0.2.2",
    "startTimeMs": 1791097540432,
    "runId": "local-1791097541542",
    "engine": "spark",
    "engineVersion": "3.5.7",
    "attributes": {
      "run_id": "orders_daily__2026-10-04T02:00",
      "dag": "orders_daily"
    },
    "jobId": "orders_nightly",
    "user": "root"
  },
  "mergedContractYaml": "id: orders_nightly\nversion: 0.1.0\nstatus: draft\ninputs:\n- name: orders\n  location: /tmp/invaract-example/data/orders.csv\n  schema:\n    fields:\n    - name: id\n      type: integer\n      required: true\n      nullable: true\n    - name: value\n      type: integer\n      required: true\n      nullable: true\noutputs:\n- name: orders_doubled\n  location: /tmp/invaract-example/out/orders_doubled\n  format: parquet\n  schema:\n    fields:\n    - name: id\n      type: integer\n      required: true\n      nullable: true\n    - name: doubled\n      type: integer\n      required: true\n      nullable: true\n  saveMode: overwrite\n  derivedFrom:\n  - orders\nextensions:\n  source_system: crm\n  owner: orders-team\n",
  "eventId": "9f21991dcafb1f97bfd8a3303db9efe924b2419e2161913a5be84e9010c97a35",
  "statusCounts": {
    "SKIPPED_UNRECOGNIZED": 1,
    "INFERRED": 1
  },
  "metadata": {
    "source_system": "crm",
    "owner": "orders-team"
  },
  "schemaVersion": 1,
  "mergedContractDigest": "da5d6a750b375826cf983342bc94129a020702974c0173cd3adc385ce6d407eb"
}

Every event also carries schemaVersion, a deterministic eventId to deduplicate on, and (on the first two) a contractDigest: the SHA-256 of the draft, identical for identical drafts. The JSON Schema and a reference receiver are in the notification sinks guide.

job.jobId comes from spark.invaract.jobId — set it. Spark’s own app name is often different on every run under a scheduler, so it can’t tell you “this is the same job as last night”; the fingerprints field identifies the transformation itself, which makes repeat runs recognizable. Every spark.invaract.job.metadata.<key>=<value> conf entry arrives as job.attributes.<key>; nothing else from your Spark configuration is sent.

Tag the draft, and link events back to the run

Two kinds of metadata, kept apart on purpose:

Conf prefix	Describes	Where it lands
spark.invaract.contract.metadata.<key>=<value>	The contract: owner, team, source system, a ticket.	The draft’s extensions: block, and every event’s metadata.
spark.invaract.job.metadata.<key>=<value>	One run of the job: an orchestrator run id, a DAG, a trigger.	job.attributes on every event, dry-run and enforcement alike, and nowhere else.

The run-level facts stay out of the contract because a contract outlives the run: a run id in a promoted contract would be stale the next night. The contract-level ones ride in the draft, so it arrives already tagged with what it is and who owns it.

They also link the dry run to what comes after it. When you promote a draft and enforce it, Invaract copies a contract’s extensions into its own events, so the same keys appear in metadata on every CONTRACT_VALIDATION and WRITE event of the real runs. Whatever you use to find the dry run’s events, a key like owner or a ticket id, finds those too. Both prefixes can be set with spark-submit --conf, with no change to the job:

  --conf spark.invaract.contract.metadata.owner=orders-team \
  --conf spark.invaract.contract.metadata.source_system=crm \
  --conf spark.invaract.job.metadata.run_id=orders_daily__2026-10-04T02:00

That is what produced the examples above: look at metadata and job.attributes in each, and at the extensions: block in contractYaml. Values are strings.

The DRY_RUN_SUMMARY’s mergedContractYaml is the job’s drafts merged exactly as MergeDraftsCli does, named for the job. If two writes disagree about a location’s schema, mergeStatus is CONFLICT and mergeConflicts lists them; no contract is guessed at.

Sending successes and problems to different places

Use the FanOutNotificationSink described in Configure a Notification Sink to send the clean drafts to your contract repository’s endpoint and the rest to a different one:

sink.enabled=true
sink.class=com.invaract.verification.notification.FanOutNotificationSink
sink.property.sinks=repository,triage
sink.property.repository.class=com.invaract.verification.notification.HttpNotificationSink
sink.property.repository.property.url=https://contracts.corp.internal/inferred
sink.property.repository.property.bearerTokenEnv=INVARACT_TOKEN
sink.property.repository.statuses=INFERRED
sink.property.triage.class=com.invaract.verification.notification.HttpNotificationSink
sink.property.triage.property.url=https://triage.corp.internal/invaract
sink.property.triage.statuses=INFERRED_DEGRADED,SKIPPED_UNSUPPORTED,SKIPPED_UNRECOGNIZED,INFERENCE_ERROR

The end-of-run summary always reaches every target, so each destination still learns that the job ended.

Dry-run never gets in the job's way

Inference, the self-check and publishing are each failure-isolated: a sink that is down or a plan shape Invaract hasn’t seen costs a log line, never a query. That includes setting the sink up: a notifyConfig that can’t be read, or names a class that doesn’t exist, logs one WARN and falls back to logging the inferred contract, instead of failing the job (enforcement mode, by contrast, still fails loudly on a bad notifyConfig). Delivery to an HTTP endpoint is asynchronous, and at application end the sink is given up to ten seconds to finish what it already accepted, so a short job doesn’t lose its last events. Each SparkSession a job creates reports on its own; a job that spawns child sessions sends one summary per session.

Migrating a job: from dry-run to enforcement

Dry-run exists to answer one question per job: is it safe to turn enforcement on? Work through the answer like this.

1. Run the job once with dry-run reporting, down every code path that writes (a job that writes differently by date or by parameter needs a run for each).
2. Read its DRY_RUN_SUMMARY. statusCounts says how many writes landed in each status.
3. A job is ready when every write is INFERRED with selfCheck: PASSED, nothing is SKIPPED_UNRECOGNIZED, and mergeStatus is MERGED.
4. If it isn’t, the status says what to do:
   • SKIPPED_UNRECOGNIZED: the write is a shape Invaract can’t verify, so enforcement would reject it. Check Connector Support; change the write to a supported shape if you can (an example).
   • INFERRED_DEGRADED: read diagnostics. An unresolved .checkpoint() means a draft’s inputs are incomplete; row-level DML means the draft describes only the target’s current schema.
   • selfCheck of FAILED, INVALID or ERROR: the draft doesn’t pass its own write, so don’t promote it. selfCheckViolations and diagnostics say why.
   • NO_WRITES_OBSERVED: Invaract saw no write it recognizes. Either that run never wrote, or the write shape isn’t recognized; don’t read it as “safe”.
   • CONFLICT: two writes disagree about one location’s schema. mergeConflicts names it.
5. Review the merged draft (mergedContractYaml): replace the placeholder id, version and status (its extensions already carry the contract metadata you set), make locations portable, and add the rules dry-run can’t infer, as in Write a Contract.
6. Switch to enforcement: replace spark.invaract.dryRun=true with spark.invaract.contract=<your reviewed contract>. The enforcement events carry the same job (engine, jobId, attributes), so the job’s dry run and its enforced runs can be followed together. The strict toggles (rejectUndeclaredInputs, rejectUndeclaredFields) are off by default; turn them on later, once the job has run clean.

Need a custom callback?

Call ContractEnforcementRule.dryRun yourself when you want to do something with the inferred contract besides logging it — write it to a specific file, publish it somewhere, merge it with an existing draft. See Install and Configure Invaract Explicitly in Code.

ContractEnforcementRule.dryRun never throws and never blocks a write, under either mechanism — there’s no contract to violate. It only calls its callback (or, in the conf-driven path, logs) when the analyzed plan is a write Invaract recognizes (see Connector Support); a .count(), a plain read, or an unrecognized write shape produces nothing — unless a sink is configured, which also reports the writes it couldn’t infer from (see above). If your job performs more than one write, the callback fires once per recognized write — capture the one you want by overwriting a mutable cell, the same way the explicit-code example does.

Merging drafts from a job with several writes

A job with several writes prints one draft per write. Save each draft to its own file (a directory of them works too), then merge them into one multi-output contract with the contract module’s command line — no Spark involved:

cd contract
sbt "runMain com.invaract.contract.cli.MergeDraftsCli --id customer_pipeline --version 1.0.0 drafts/"

Given two drafts — one write reading data/orders.csv, another reading data/customers.csv and data/orders.csv (as its input and input_1) — it prints:

inputs:
- name: orders
  location: data/orders.csv
  ...
- name: customers
  location: data/customers.csv
  ...
outputs:
- name: orders_report
  location: out/orders_report.parquet
  ...
  derivedFrom:
  - orders
- name: customer_totals
  location: out/customer_totals.parquet
  ...
  derivedFrom:
  - orders
  - customers

What the merge does, and what it refuses to guess:

• Inputs are one per location. The same location read by several writes becomes one input, named after its location (data/orders.csv → orders; a catalog table main.sales.orders → orders; a clash gets _2, _3, …). Each output’s derivedFrom is rewritten to those names.
• Conflicting schemas are an error, not a pick. If two drafts saw the same location with different schemas (say, one was inferred before the source changed), the command prints every conflict, exits 1, and writes nothing. Re-run the stale write and merge again.
• Outputs stay distinct by location. Two writes to one location merge only if they agree on its schema.
• A write that read nothing keeps derivedFrom: [].
• The merged contract is checked with the same validation as any contract file, and parses back with the normal parser. --output merged.yaml writes it to a file instead of standard output; --id defaults to merged_contract, --version to 0.1.0.

From code, ContractDraftMerger.merge(drafts, id, version) does the same on parsed Contracts and returns either the merged contract or the list of conflicts.

Checkpoints

If your job calls .checkpoint() between reading and writing, dry-run mode sees through it the same way real enforcement does (see MISSING_INPUT in Violation Types): the inferred contract names the inputs read before the checkpoint, not an empty list. The checkpointed Dataset has to have been created while the rule was installed — a checkpoint made before the session had the extension can’t be resolved, and its upstream inputs won’t appear in the draft.

What it doesn’t do

Dry-run mode only observes — it’s the mirror image of SparkAdapterListener, not a replacement for real enforcement. Once you have a reviewed contract, switch back to ContractEnforcementRule.forContract — that’s the mechanism that actually gates a write. Row-level DML (MERGE/UPDATE/DELETE) is inferred only weakly: those operations have no single “new output” to build a dataset schema from, so the draft describes the target’s current schema and none of the operation’s own logic (see Enforce Row-Level DML Rules for how they’re verified once a contract exists). When reporting to a sink, such a draft is flagged INFERRED_DEGRADED.