Package org.apache.spark.sql.connector.catalog

Interface Changelog

@Evolving public interface Changelog
The central connector interface for Change Data Capture (CDC).

Connectors implement this minimal interface to expose change data. Spark handles post-processing (carry-over removal, update detection, net change computation) based on the properties declared by the connector.

The columns returned by columns() must include the following metadata columns:

  • _change_type (STRING) — the kind of change: insert, delete, update_preimage, or update_postimage
  • _commit_version — the commit version containing this change. Must be either LongType or StringType; all other types are rejected. The column's natural ordering (numeric for LongType, lexicographic for StringType) must match commit order, because the netChanges post-processing path sorts rows of a given row identity by this column to determine the first and last events.
  • _commit_timestamp (TIMESTAMP) -- the timestamp of the commit. All rows belonging to a single _commit_version must share the same _commit_timestamp. For streaming reads with post-processing enabled, two additional requirements apply:
    1. All rows of a single commit must appear in the same micro-batch (i.e. micro-batch boundaries align with commit boundaries).
    2. Each micro-batch's rows must have _commit_timestamp strictly greater than the maximum _commit_timestamp of any prior micro-batch.
    Streaming post-processing uses _commit_timestamp as event time with a zero-delay watermark, so once a micro-batch observes max event time T the global watermark advances to T. Both Spark's late-event filter and its state-eviction predicate then use eventTime <= T -- so any later row at _commit_timestamp <= T (whether from the same commit split across batches, a different commit emitted later, or simply an out-of-order commit) is silently dropped as late. Requirement 1 keeps a single commit's rows together; requirement 2 keeps distinct commits in strictly increasing event-time order across batches. Multiple distinct commits with equal _commit_timestamp are allowed within a single micro-batch -- only across batches does timestamp progression need to be strictly increasing. Atomic-commit CDC connectors (e.g. Delta versions, Iceberg snapshots) that derive _commit_timestamp from wall-clock time at commit time naturally satisfy both requirements. _commit_timestamp must be non-NULL on every row of a streaming read engaging post-processing; both the row-level Aggregate path and the netChanges transformWithState path raise CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP on a violation

Streaming reads support carry-over removal, update detection, and net change computation. Two streaming-specific behaviors to be aware of:

  • Output is buffered until the watermark advances past the commit. When a micro-batch ingests a commit, that commit's output rows are buffered in state and not emitted in the same batch. They are emitted by a later micro-batch -- whichever one advances the watermark past the commit's _commit_timestamp. The last commit's output is emitted when the source terminates.
  • netChanges only merges changes that are buffered together. When each row identity appears in at most one commit within any buffered window, the streaming output is the same as computeUpdates. Cross-commit merging only happens when several commits touch the same row before the earliest one's output has been released. For full-range collapse, use a batch read.

Pushdown contract. When any post-processing pass applies (carry-over removal, update detection, or netChanges), Spark only pushes predicates that reference _commit_version, _commit_timestamp, or columns named by rowId() to the connector's SupportsPushDownFilters / SupportsPushDownV2Filters. Predicates on _change_type, the rowVersion() column, or non-rowId data columns are kept above the scan: pushing them would drop one half of a delete/insert pair within a row-identity group and silently break post-processing. Catalyst's pushdown rules enforce this via the rewrite operators, so connectors do not need to code the restriction themselves -- but must not bypass it via connector-specific options. When no post-processing pass applies, Spark does not impose any CDC-specific predicate-pushdown restriction. SupportsPushDownRequiredColumns (column pruning) is unrestricted in either case: Spark's pruning already respects what the rewrite operators reference.

Since:
4.2.0

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final String
    CHANGE_TYPE_DELETE
    Constant for the _change_type value of a row deleted from the table.
    static final String
    CHANGE_TYPE_INSERT
    Constant for the _change_type value of a row inserted into the table.
    static final String
    CHANGE_TYPE_UPDATE_POSTIMAGE
    Constant for the _change_type value of an update's post-image row.
    static final String
    CHANGE_TYPE_UPDATE_PREIMAGE
    Constant for the _change_type value of an update's pre-image row.

  • Method Summary

    Modifier and Type
    Method
    Description
    Column[]
    columns()
    Returns the columns of this changelog, including data columns and the required metadata columns (_change_type, _commit_version, _commit_timestamp).
    boolean
    containsCarryoverRows()
    Whether the raw change data may contain identical insert/delete carry-over pairs produced by copy-on-write file rewrites.
    boolean
    containsIntermediateChanges()
    Whether the raw change data may contain multiple intermediate states per row identity within the requested changelog range (across all commit versions in the range).
    String
    name()
    A name to identify this changelog.
    ScanBuilder
    newScanBuilder(CaseInsensitiveStringMap options)
    Returns a new ScanBuilder for reading the change data.
    boolean
    representsUpdateAsDeleteAndInsert()
    Whether updates in the raw change data are represented as delete+insert pairs rather than fully materialized update_preimage and update_postimage entries.
    default NamedReference[]
    rowId()
    Returns the columns that uniquely identify a row, used for carry-over removal, update detection, and net change computation.
    default NamedReference
    rowVersion()
    Returns the column that holds the row version — the commit version at which the row's content was last modified.

  • Field Details

    • CHANGE_TYPE_INSERT

      static final String CHANGE_TYPE_INSERT
      Constant for the _change_type value of a row inserted into the table.
      See Also:

    • CHANGE_TYPE_DELETE

      static final String CHANGE_TYPE_DELETE
      Constant for the _change_type value of a row deleted from the table.
      See Also:

    • CHANGE_TYPE_UPDATE_PREIMAGE

      static final String CHANGE_TYPE_UPDATE_PREIMAGE
      Constant for the _change_type value of an update's pre-image row.
      See Also:

    • CHANGE_TYPE_UPDATE_POSTIMAGE

      static final String CHANGE_TYPE_UPDATE_POSTIMAGE
      Constant for the _change_type value of an update's post-image row.
      See Also:

  • Method Details

    • name

      String name()
      A name to identify this changelog.

    • columns

      Column[] columns()
      Returns the columns of this changelog, including data columns and the required metadata columns (_change_type, _commit_version, _commit_timestamp).

    • containsCarryoverRows

      boolean containsCarryoverRows()
      Whether the raw change data may contain identical insert/delete carry-over pairs produced by copy-on-write file rewrites.

      When true and the CDC query's deduplicationMode is not none, Spark will remove carry-over pairs from the raw change data. If false, the connector guarantees that no carry-over pairs are present in the raw change data and Spark will skip carry-over removal entirely.

    • containsIntermediateChanges

      boolean containsIntermediateChanges()
      Whether the raw change data may contain multiple intermediate states per row identity within the requested changelog range (across all commit versions in the range).

      When true and the CDC query's deduplicationMode is netChanges, Spark will collapse multiple changes per row identity into the net effect. If false, the connector guarantees at most one change per row identity across the entire changelog range, and Spark will skip net change computation.

      Note this flag is range-scoped (across all commits in the request), not micro-batch-scoped.

    • representsUpdateAsDeleteAndInsert

      boolean representsUpdateAsDeleteAndInsert()
      Whether updates in the raw change data are represented as delete+insert pairs rather than fully materialized update_preimage and update_postimage entries.

      When true and the CDC query's computeUpdates option is enabled, Spark will derive update_preimage/update_postimage from insert/delete pairs in the raw change data. If false, the connector guarantees that update pre/post-images are already present in the raw change data.

    • newScanBuilder

      ScanBuilder newScanBuilder(CaseInsensitiveStringMap options)
      Returns a new ScanBuilder for reading the change data.
      Parameters:
      options - read options (case-insensitive string map)

    • rowId

      default NamedReference[] rowId()
      Returns the columns that uniquely identify a row, used for carry-over removal, update detection, and net change computation.

      The default implementation throws UnsupportedOperationException. Connectors must override this method when any of containsCarryoverRows(), representsUpdateAsDeleteAndInsert(), or containsIntermediateChanges() returns true. Each referenced column must be non-nullable.

    • rowVersion

      default NamedReference rowVersion()
      Returns the column that holds the row version — the commit version at which the row's content was last modified. The row version has these properties:
      • Assigned the current commit version when the row is initially inserted.
      • Bumped to the current commit version when the row's content is updated.
      • Preserved when the row is rewritten by a copy-on-write operation without a content change — it is NOT bumped to the current commit version.
      The row version is distinct from _commit_version. _commit_version identifies the commit that emitted this change row; the row version identifies the commit that last wrote the row's content. For a delete+insert pair produced within a single commit, both halves share the same row version if the pair is a copy-on-write carry-over, and have different row versions (old on the delete, new on the insert) if the pair is a true update.

      Spark uses the row version to distinguish copy-on-write carry-over from update without scanning data columns, for both carry-over removal and update detection.

      The default implementation throws UnsupportedOperationException. Connectors must override this method when containsCarryoverRows() or representsUpdateAsDeleteAndInsert() returns true. The referenced column must be non-nullable.