Operational Metadata¶

Column Definitions¶

Operational metadata are automatically generated columns that provide lineage, data provenance, and processing context. These columns are added to your tables without requiring manual SQL modifications.

Note

Operational metadata columns are defined in the project level configuration file. under the operational_metadata key.

Project-level configuration:

lhp.yaml - Project operational metadata configuration¶

# LakehousePlumber Project Configuration
name: my_lakehouse_project
version: "1.0"

operational_metadata:
  columns:
    _processing_timestamp:
      expression: "F.current_timestamp()"
      description: "When the record was processed by the pipeline"
      applies_to: ["streaming_table", "materialized_view", "view"]

    _source_file_path:
      expression: "F.col('_metadata.file_path')"
      description: "Source file path for lineage tracking"
      applies_to: ["view"]

    _record_hash:
      expression: "F.xxhash64(*[F.col(c) for c in df.columns])"
      description: "Hash of all record fields for change detection"
      applies_to: ["streaming_table", "materialized_view", "view"]
      additional_imports:
        - "from pyspark.sql.functions import xxhash64"

    _pipeline_name:
      expression: "F.lit('${pipeline_name}')"
      description: "Name of the processing pipeline"
      applies_to: ["streaming_table", "materialized_view", "view"]

Version Requirements¶

LakehousePlumber supports version enforcement to ensure consistent code generation across development and CI environments. This prevents “works on my machine” issues and ensures reproducible builds.

Basic configuration:

lhp.yaml - Version enforcement examples¶

# LakehousePlumber Project Configuration
name: my_lakehouse_project
version: "1.0"

# Enforce version requirements (optional)
required_lhp_version: ">=0.4.1,<0.5.0"  # Allow patch updates within 0.4.x

Version specification formats:

Version requirement examples¶

# Exact version pin (strict)
required_lhp_version: "==0.4.1"

# Allow patch updates only
required_lhp_version: "~=0.4.1"          # Equivalent to >=0.4.1,<0.5.0

# Range with exclusions
required_lhp_version: ">=0.4.1,<0.5.0,!=0.4.3"  # Exclude known bad version

# Allow minor updates
required_lhp_version: ">=0.4.0,<1.0.0"

Behavior:

When required_lhp_version is set, lhp validate and lhp generate will fail if the installed version doesn’t satisfy the requirement
Informational commands like lhp show skip version checking to allow inspection even with mismatches
Version checking uses PEP 440 version specifiers

Emergency bypass:

Bypass version checking in emergencies¶

# Temporarily bypass version checking
export LHP_IGNORE_VERSION=1
lhp generate -e dev

# Or inline
LHP_IGNORE_VERSION=1 lhp validate -e prod

CI/CD integration:

CI pipeline with version enforcement¶

# Install exact version matching project requirements
pip install "lakehouse-plumber$(yq -r .required_lhp_version lhp.yaml | sed 's/^//')"

# Or use range-compatible version
pip install "lakehouse-plumber>=0.4.1,<0.5.0"

# Validate and generate (will fail if version mismatch)
lhp validate -e prod
lhp generate -e prod

Note

Version enforcement is optional. Projects without required_lhp_version work normally with any installed LakehousePlumber version.

Warning

Use the bypass environment variable (LHP_IGNORE_VERSION=1) only in emergencies. It’s not recommended for production environments as it defeats the purpose of version consistency.

Event Log Configuration¶

LakehousePlumber supports project-level event log configuration in lhp.yaml. When configured, event log blocks are automatically injected into all pipeline resource files during generation — no -pc flag or pipeline_config.yaml required.

lhp.yaml - Event log configuration¶

name: my_lakehouse_project
version: "1.0"

event_log:
  catalog: "${catalog}"
  schema: _meta
  name_suffix: "_event_log"

When event_log is defined, each generated pipeline resource will include an event_log block with the table name derived from the pipeline name (e.g., bronze_load_event_log).

Individual pipelines can override or opt out of project-level event logging through pipeline_config.yaml.

Target Type Compatibility¶

The applies_to field controls which DLT table types can use each operational metadata column. LHP automatically filters columns based on the target type to prevent runtime errors.

Purpose of target type restrictions:

When defining operational metadata columns at the project level, the applies_to field serves as a safeguard mechanism to protect end users from accidentally using incompatible columns in their pipeline configurations. This is a defensive design pattern that prevents common mistakes.

Best practice for project administrators:

Set restrictive applies_to values for source-specific columns (e.g., CloudFiles metadata)
Use broader applies_to values for universal columns (e.g., timestamps, pipeline names)
This protects pipeline developers from runtime failures and provides clear usage guidance

Target types:

``view`` - Source views created by load actions (@dp.temporary_view())
``streaming_table`` - Live tables with streaming updates (@dp.materialized_view())
``materialized_view`` - Batch-computed views for analytics (@dp.temporary_view())

Source-specific metadata limitations:

Warning

Metadata columns that depend on CloudFiles features (like _metadata.file_path) are only available in views that load data from CloudFiles sources. These columns will cause runtime errors if used with JDBC, SQL, Delta, or custom_datasource sources.
Custom data sources may provide their own metadata columns depending on their implementation, but CloudFiles-specific metadata will not be available.

Usage in YAML Files¶

Operational metadata can be configured at multiple levels with additive behavior - columns from all levels are combined together:

Important

Additive Behavior: Operational metadata columns are never overridden between levels. Instead, columns from preset + flowgroup + action levels are combined together. The only exception is operational_metadata: false at action level, which disables all metadata.

Preset level

presets/bronze_layer.yaml¶

name: bronze_layer
version: "1.0"

defaults:
  operational_metadata: ["_processing_timestamp", "_source_file_path"]

FlowGroup level

pipelines/customer_ingestion/load_customers.yaml¶

pipeline: customer_ingestion
flowgroup: load_customers
presets: ["bronze_layer"]
operational_metadata: ["_record_hash"]  # Adds to preset columns

actions:
  - name: load_customer_files
    type: load
    source:
      type: cloudfiles
      path: "/mnt/landing/customers/*.json"
      format: json
    target: v_customers_raw

Action level

Action-specific metadata configuration¶

actions:
  - name: load_with_custom_metadata
    type: load
    source:
      type: cloudfiles
      path: "/mnt/data/*.parquet"
      format: parquet
    operational_metadata:  # Adds to flowgroup + preset columns
      - "_pipeline_name"
      - "_custom_business_logic"
    target: v_enriched_data

  - name: load_without_metadata
    type: load
    source:
      type: sql
      sql: "SELECT * FROM source_table"
            operational_metadata: false  # Disables all metadata
     target: v_clean_data

Additive behavior example:

Complete example showing additive behavior¶

# Preset defines base columns
# presets/bronze_layer.yaml
defaults:
  operational_metadata: ["_processing_timestamp"]

# FlowGroup adds more columns
pipeline: customer_ingestion
flowgroup: load_customers
operational_metadata: ["_source_file_path", "_record_hash"]

actions:
  - name: load_customer_files
    type: load
    source:
      type: cloudfiles
      path: "/mnt/data/*.json"
    # Action adds even more columns
    operational_metadata:
      - "_pipeline_name"
      - "_custom_business_logic"
    target: v_customers_raw

# Final result: ALL columns combined
# ✓ _processing_timestamp      (from preset)
# ✓ _source_file_path          (from flowgroup)
# ✓ _record_hash               (from flowgroup)
# ✓ _pipeline_name             (from action)
# ✓ _custom_business_logic     (from action)

Usage Patterns¶

Enable all available columns:

operational_metadata: true

Select specific columns:

operational_metadata:
  - "_processing_timestamp"
  - "_source_file_path"
  - "_record_hash"

Disable metadata:

operational_metadata: false

Generated Python code:

Generated DLT code with operational metadata¶

@dp.temporary_view()
def v_customers_raw():
    """Load customer files from landing zone"""
    df = spark.readStream \
        .format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/mnt/landing/customers/*.json")

    # Add operational metadata columns
    df = df.withColumn('_processing_timestamp', F.current_timestamp())
    df = df.withColumn('_source_file_path', F.col('_metadata.file_path'))
    df = df.withColumn('_record_hash', F.xxhash64(*[F.col(c) for c in df.columns]))

    return df

Danger

When you add operational metadata columns to an upstream action, if your downstream action is a transformation, for example SQL transform, you need to make sure they are included in the SQL query.

Internal Implementation Note¶

The codebase maintains strict semantic separation between single and multi-document YAML files:

load_yaml_file() - For single-document files (configs, templates, presets)
- Validates exactly one document exists
- Raises MultiDocumentError (LHP-IO-003) for empty files or files with multiple documents
- Used for templates, presets, configs, and other single-document files
load_yaml_documents_all() - For multi-document files (flowgroup files only)
- Returns list of all documents
- Used exclusively for flowgroup YAML files that may contain multiple flowgroups

This strict validation prevents accidental misuse and catches bugs early. If you encounter a MultiDocumentError, the error message will guide you to the correct loading method.