Dvp Io Schema Identifier

DVP IO Identifier

Overview

Scans the source PySpark code in dvp/01-source/ to identify all data inputs and outputs (files and tables), then infers column schemas for each entry. Produces dvp/04-results/data_io_schema.json — a unified inventory with full schema information that downstream skills use for test data generation and test setup.

This skill combines IO detection and schema extraction in a single pass, producing a complete data_io_schema.json with all fields populated.

Output Format

The skill produces dvp/04-results/data_io_schema.json as an array of IO entries, each with full schema information:

[
  {
    "name": "raw_transactions.csv",
    "full_name": "raw_transactions.csv",
    "source": "workload.py:29",
    "type": "file",
    "format": "csv",
    "detection": "static",
    "role": "input",
    "path": "input/raw_transactions.csv",
    "columns": [
      {"name": "transaction_id", "type": "STRING", "confidence": "evidence"},
      {"name": "customer_id", "type": "STRING", "confidence": "evidence"},
      {"name": "product_id", "type": "STRING", "confidence": "evidence"},
      {"name": "transaction_date", "type": "DATE", "confidence": "pattern"},
      {"name": "transaction_amount", "type": "DECIMAL", "confidence": "pattern"},
      {"name": "return_amount", "type": "DECIMAL", "confidence": "ambiguous"}
    ]
  },
  {
    "name": "TOP_CATEGORIES",
    "full_name": "TOP_CATEGORIES",
    "source": "workload.py:94",
    "type": "table",
    "format": "",
    "detection": "inferred",
    "role": "output",
    "path": "",
    "key_columns": ["category"],
    "columns": [
      {"name": "category", "type": "STRING", "confidence": "evidence"},
      {"name": "total_sales_count", "type": "DECIMAL", "confidence": "evidence"},
      {"name": "category_rank", "type": "DECIMAL", "confidence": "evidence"}
    ]
  }
]

Field Definitions

Confidence Values

Type Values

Detection Values

Output Format

Every time you begin a step, sub-step, or significant action, prefix the message with a timestamp in the format [YYYY-MM-DD HH:MM:SS]. Obtain the current time by running date '+%Y-%m-%d %H:%M:%S' in bash.

Example:

[2026-03-24 14:05:32] Starting Step 1: Analyze source files for I/O...
[2026-03-24 14:05:45] Identified 4 inputs and 2 outputs
[2026-03-24 14:05:46] Step 1 complete.

Workflow

Step 0: Initialize Git

Ensure the workload directory has a git repository on the sma/migration-process branch. This is idempotent — if the orchestrator already initialized git, this is a no-op.

result = sma_api.git_ensure_ready("<workload_path>")

Step 1: Validate ASG and Load Entrypoints

Before any processing, verify the ASG file exists:

• Check dvp/04-results/*_asg.json exists. If not, STOP with error: "ERROR: No ASG found. The ASG is MANDATORY for IO identification. Run dvp-asg-generation first."
• Do NOT fall back to scanning source code without the ASG.

Then read dvp/04-results/entrypoints.json to know which files and functions to analyze. Focus on entries with status: "validated" or status: "pending".

Step 2: Scan for IO Operations

For each source file referenced by an entrypoint, scan for data read/write operations using AST analysis or pattern matching.

Input Detection Patterns

Output Detection Patterns

Step 3: Extract Names and Paths

For each detected IO operation:

1. Resolve the name/path argument:

   • Literal string → detection: "static"
   • f-string or variable → detection: "inferred", resolve if possible

2. Determine the name:

   • Files: use the filename (e.g., raw_transactions.csv) or directory name (e.g., DAILY_SALES_SUMMARY)
   • Tables: use the table name as-is (e.g., CUSTOMER_MASTER)

3. Set full_name equal to name (no database/schema prefix for basic cases)

4. Determine format from the read/write method:

   • .csv() → "csv"
   • .parquet() → "parquet"
   • .json() → "json"
   • .table() / .saveAsTable() → "" (no format for tables)

5. Record the source location as filename:lineno

Step 4: Infer Column Schemas

For each IO entry, search the source code for schema information using the following sources (ordered by confidence):

For outputs, also identify key columns from groupBy(), distinct(), or join keys used in the transformation that produces the output.

Type Mapping (Spark → Snowflake)

Step 5: Deduplicate

The same data source may be read multiple times in the code. Deduplicate by (name, type, role):

• Keep the first occurrence (lowest line number) as the source reference
• If a name appears as both input and output, keep both entries (different roles)
• Merge column information from multiple references to the same entry

Step 6: Validate Against Source

Each entry in data_io_schema.json must correspond to an actual IO operation in the source code. Do not invent entries that don't exist in the source.

Step 7: Present Findings

IO Identification & Schema Extraction
======================================

Source: dvp/01-source/workload.py

Inputs (5):
  FILES:
    1. raw_transactions.csv    workload.py:29    csv     static    5 columns
    2. returns_data.csv        workload.py:32    csv     static    4 columns
    3. exchange_rates.csv      workload.py:98    csv     static    3 columns
  TABLES:
    4. CUSTOMER_MASTER         workload.py:59            static    4 columns
    5. PRODUCT_CATALOG         workload.py:76            static    4 columns

Outputs (5):
  FILES:
    6. DAILY_SALES_SUMMARY     workload.py:50    csv     inferred  4 columns
    7. CUSTOMER_CLV            workload.py:73    csv     inferred  5 columns
    8. FOREIGN_TRANSACTIONS    workload.py:112   csv     inferred  3 columns
  TABLES:
    9. TOP_CATEGORIES          workload.py:94            inferred  3 columns
   10. UNMATCHED_TRANSACTIONS  workload.py:122           inferred  2 columns

Schema coverage: 10/10 entries have columns defined

Saved to dvp/04-results/data_io_schema.json

Next: Run dvp-synthetic-data-generator to generate test data.

Step 8: Commit Changes to Git

After IO schema is identified and written, commit the changes:

result = sma_api.git_commit("<workload_path>", """DVP IO Schema: Identified N inputs and M outputs

Inputs: N (X files, Y tables)
Outputs: M (X files, Y tables)
Schema coverage: N/N entries have columns defined
Output: dvp/04-results/data_io_schema.json""")

Verify branches:

result = sma_api.git_verify_branches("<workload_path>")

SMA Reports as Hints

When available, SMA reports can provide additional hints for IO identification:

These are hints only — the source of truth is always the actual source code. SMA reports may list operations that don't correspond to real IO (e.g., intermediate DataFrames) or miss dynamically constructed paths.

Execution Order

dvp-orchestrator  →  stage-conversion  →  dvp-notebook-to-script
      |
      v
dvp-asg-generation  →  XX_asg.json + XX_anomalies.json
      |
      v
dvp-entrypoint-identifier  →  entrypoints.json
      |
      v
dvp-code-adapter  (session injection + enableHiveSupport() + defer env vars)
      |
      v
dvp-io-schema-identifier  ← THIS SKILL  →  data_io_schema.json
      |
      v
dvp-synthetic-data-generator  →  04-results/synthetic_data/*.csv + reports
      |
      v
dvp-test-setup-generator  →  03-tests/

Inputs

Outputs

Stopping Points

• No entrypoints found: Stop and ask user to run dvp-entrypoint-identifier first
• No IO operations detected: Warn user — the workload may use dynamic paths or unsupported patterns
• Ambiguous path resolution: When a path is fully dynamic (no literal hints), mark as detection: "inferred" and flag for user review
• No schema inferred: Leave columns empty and flag for user review

Edge Cases

Embedded Library

The IO detection and schema inference logic is borrowed from the warp-suite DataIODetector:

dvp/
  dvp-orchestrator/
    warp_core/               <-- SHARED: diagnostics, spark/pandas function registries
  dvp-io-schema-identifier/
    warp/
      data_io/               <-- warp-suite DataIODetector
    scripts/
      io_schema_identifier.py

The DataIODetector operates on ASG JSON (not source code), using a 6-phase type inference pipeline:

1. ASG seed (high-confidence inferred_columns from the parser)
2. Code analysis (from inferred_output fields on transformations)
3. StructField explicit schemas
4. Filter condition analysis (evidence: col == 'value' implies STRING)
5. Function usage inference (evidence: F.sum(col) implies DECIMAL)
6. Pattern-based inference (fallback: _id implies INT, _dt implies DATE)

With join propagation between phases to spread inferred types across related tables.

Python Dependencies

pip install pydantic

Design Considerations

• Combines IO detection and schema extraction in a single pass from the ASG
• Uses evidence-based type inference before pattern-based fallbacks
• Should merge information from multiple code locations for the same table
• Column names use lowercase (matching PySpark conventions); Snowflake auto-uppercases at runtime
• The output format (columns array of {name, type} objects) is consumed by dvp-synthetic-data-generator and dvp-test-setup-generator

Final Summary

MANDATORY: After completing all steps (whether running standalone or invoked from the orchestrator), ALWAYS present this summary table:

IO Schema Identification Complete

┌──────────────┬──────────┬──────────────────────────────────────────────────────────┐
│ Step         │ Status   │ Details                                                  │
├──────────────┼──────────┼──────────────────────────────────────────────────────────┤
│ IO Schema    │ Done     │ N inputs, M outputs identified                           │
└──────────────┴──────────┴──────────────────────────────────────────────────────────┘

Output location: <output>/

Git branches:
• main — original code (unmodified)
• sma/migration-process — IO schema changes applied

Rules:

• Replace N and M with actual counts of inputs and outputs identified
• Status is Done, Skipped, or Failed
• If no IO was detected, show Skipped with reason
• If identification failed, show Failed with brief error
• The git branches section uses sma_api.git_verify_branches() to confirm both branches exist