This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new c028577a45 issue #2647 : document Change Data Capture approaches
(#8006)
c028577a45 is described below
commit c028577a4535b9d58c557878e5ab3e3dbbdd939c
Author: Matt Casters <[email protected]>
AuthorDate: Wed Aug 19 15:51:02 2026 +0200
issue #2647 : document Change Data Capture approaches (#8006)
Add a how-to set covering log sniffing, filtered selection, and
snapshot comparison, with cross-links from Getting Started and the
transforms used by each pattern.
---
docs/hop-user-manual/modules/ROOT/nav.adoc | 4 +
.../ROOT/pages/getting-started/hop-next-steps.adoc | 1 +
.../pages/getting-started/hop-what-is-hop.adoc | 2 +-
.../how-to-guides/cdc-filtered-selection.adoc | 175 +++++++++++++++++++++
.../ROOT/pages/how-to-guides/cdc-log-sniffing.adoc | 149 ++++++++++++++++++
.../how-to-guides/cdc-snapshot-comparison.adoc | 161 +++++++++++++++++++
.../pages/how-to-guides/change-data-capture.adoc | 91 +++++++++++
.../modules/ROOT/pages/how-to-guides/index.adoc | 4 +
.../ROOT/pages/how-to-guides/joins-lookups.adoc | 2 +-
.../pages/pipeline/transforms/addchecksum.adoc | 5 +
.../ROOT/pages/pipeline/transforms/execinfo.adoc | 58 +++++++
.../pages/pipeline/transforms/kafkaconsumer.adoc | 6 +-
.../ROOT/pages/pipeline/transforms/mergerows.adoc | 5 +
.../pipeline/transforms/synchronizeaftermerge.adoc | 5 +
14 files changed, 665 insertions(+), 3 deletions(-)
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc
b/docs/hop-user-manual/modules/ROOT/nav.adoc
index 044ac102f6..a262ebd627 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -552,6 +552,10 @@ under the License.
* xref:hop-usps.adoc[Unique Selling Propositions]
* xref:how-to-guides/index.adoc[How-to guides]
** xref:how-to-guides/apache-hop-web-services-docker.adoc[Hop web services in
Docker]
+** xref:how-to-guides/change-data-capture.adoc[Change Data Capture]
+*** xref:how-to-guides/cdc-log-sniffing.adoc[CDC: log sniffing]
+*** xref:how-to-guides/cdc-filtered-selection.adoc[CDC: filtered selection]
+*** xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots]
** xref:how-to-guides/joins-lookups.adoc[Joins and lookups]
** xref:how-to-guides/logging-pipeline-log.adoc[Logging pipeline data with
pipeline log]
** xref:how-to-guides/logging-workflow-log.adoc[Logging workflow data with
workflow log]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-next-steps.adoc
b/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-next-steps.adoc
index d4ab225f79..c98c1a5e3d 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-next-steps.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-next-steps.adoc
@@ -30,6 +30,7 @@ Here are a couple of topics you may want to look into:
* xref:pipeline/pipelines.adoc[Pipelines] takes closer look at the various
aspects of creating and running pipelines, and contains the entire list of
transforms that are at your disposal
* xref:workflow/workflows.adoc[Workflows] takes a closer look at the various
aspects of create and running workflows, and contains the entire list of
actions that are at your disposal
* xref:fields-parameters-variables.adoc[Fields, parameters, and variables]
compares the three ways Hop moves data and configuration
+* xref:how-to-guides/change-data-capture.adoc[Change Data Capture] covers the
three ways Hop detects inserts, updates and deletes: log sniffing, filtered
selection, and snapshot comparison.
* xref:best-practices/index.adoc[Best Practices] covers a number of things you
might want to think about while using Apache Hop.
* xref:projects/index.adoc[Projects] explains how to work with projects and
environments
* xref:hop-server/deploying.adoc[Deploying Hop Server] covers the three common
ways to get a project onto a running server
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-what-is-hop.adoc
b/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-what-is-hop.adoc
index 40dd5db3fe..a7a052ca0b 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-what-is-hop.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/getting-started/hop-what-is-hop.adoc
@@ -52,7 +52,7 @@ Depending on the engine and selected functionality, your data
can be processed i
A number of common use cases for Hop are:
* Loading large data sets into databases taking advantage of the cloud,
clustered, and massively parallel processing environments.
-* Data warehouse population with built-in support for Slowly Changing
Dimensions (SCD), Change Data Capture (CDC), and surrogate key creation.
+* Data warehouse population with built-in support for Slowly Changing
Dimensions (SCD), xref:how-to-guides/change-data-capture.adoc[Change Data
Capture (CDC)], and surrogate key creation.
* Integrate between diverse data architectures, combining relational
databases, files, NoSQL databases like Neo4j, MongoDB, Cassandra etc
* Data migration between different databases and applications.
* Data profiling and data cleansing.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-filtered-selection.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-filtered-selection.adoc
new file mode 100644
index 0000000000..667e8849ff
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-filtered-selection.adoc
@@ -0,0 +1,175 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+////
+[[CdcFilteredSelection]]
+:imagesdir: ../../assets/images
+:description: Incremental CDC in Apache Hop using a timestamp or increasing ID
and the last successful workflow or pipeline run.
+:openvar: ${
+:closevar: }
+
+= CDC: filtered selection
+
+Filtered selection is incremental extract by a `WHERE` clause. The source has
a column that only moves forward — an update timestamp, a `last_modified`
column, or an always-increasing ID — and you only read rows past the last value
you already processed.
+
+The whole pattern answers one question: *what is the last time this workflow
or pipeline ran successfully?* That value (or the last ID you applied) is the
lower bound of the next query.
+
+== Prerequisite
+
+Every insert or update you care about must change the watermark column. If a
row can be updated without touching that timestamp or without getting a new ID,
this approach misses it. In that case use
xref:how-to-guides/cdc-snapshot-comparison.adoc[snapshot comparison] or
xref:how-to-guides/cdc-log-sniffing.adoc[log sniffing].
+
+Deletes are invisible unless the source sets a deleted flag, writes to a
delete table, or never physically deletes. If you need delete detection and
have no such signal, use snapshot comparison or log sniffing.
+
+== Write the watermark yourself
+
+The straightforward design is to persist the watermark *yourself* at the end
of the workflow, on the success path only. Write it anywhere that the next run
can read: a one-row control table, a small file, a properties file, even a Hop
configuration you already use.
+
+Because you only write after a successful apply, a failed run does not advance
the bound. The next success retries the same window.
+
+[source,text]
+----
+[workflow]
+
+Start
+ |
+ v
+Pipeline: read watermark --> extract WHERE col > bound --> apply
+ |
+ +-- success --> Pipeline or SQL action: write new watermark
+ |
+ +-- failure --> do not write; next run uses the old bound
+----
+
+A control table is enough:
+
+[source,sql]
+----
+-- read at the start of the extract pipeline
+SELECT last_ts FROM cdc_watermark WHERE pipeline_name = 'load_customer';
+
+-- write at the end of the workflow, success hop only
+UPDATE cdc_watermark
+ SET last_ts = ?
+ WHERE pipeline_name = 'load_customer';
+----
+
+The value you store is either the extract’s upper bound (`CURRENT_TIMESTAMP` /
pipeline start time) or `MAX(watermark_column)` of the rows you just applied.
Using the extract start time as the next lower bound is usually safer than
`MAX(column)` if the source can still commit rows with an earlier timestamp
while you run.
+
+xref:pipeline/transforms/tableinput.adoc[Table Input] can take the bound as a
`?` parameter from the previous transform, or as a Hop variable after
xref:pipeline/transforms/setvariable.adoc[Set Variables] (in a *parent*
workflow or pipeline — variables set in the same pipeline are not visible to
later transforms in that pipeline).
+
+.Timestamp window
+
+[source,sql]
+----
+SELECT id, name, city, last_modified
+ FROM customer
+ WHERE last_modified > ?
+ AND last_modified <= ?
+----
+
+.Increasing ID
+
+[source,sql]
+----
+SELECT id, name, city
+ FROM customer_events
+ WHERE event_id > ?
+----
+
+.Same query with a Hop variable (enable *Replace variables in script*)
+
+[source,sql]
+----
+SELECT id, name, city, last_modified
+ FROM customer
+ WHERE last_modified > '{openvar}LAST_SUCCESS_TS{closevar}'
+ AND last_modified <= '{openvar}THIS_RUN_TS{closevar}'
+----
+
+=== First run
+
+If the control table has no row yet, default the lower bound far enough back
for a full load (for example `1900-01-01`, or `0` for an ID).
xref:pipeline/transforms/detectemptystream.adoc[Detect empty stream] or a
coalesce in SQL both work.
+
+=== Why the success path matters
+
+If you update the watermark in the extract pipeline *before* the apply
finishes, a crash after the write skips those rows forever. Write the watermark
in the workflow *after* the apply pipeline succeeds. That is also why you must
not write it on the failure hop.
+
+== Timestamp versus increasing ID
+
+[cols="1,2,2",options="header"]
+|===
+|Watermark
+|Works when
+|Watch out for
+
+|Timestamp (`last_modified`, `update_ts`)
+|Every relevant change updates the column
+|Time zones; clock skew between source and Hop; many rows sharing the same
timestamp. Prefer a half-open window (`>` last bound, `<=` this run’s start).
Add a small overlap if the source can deliver late rows.
+
+|Increasing ID
+|Every new or changed row gets a *new, greater* ID (insert-only tables, event
tables, change-log tables)
+|In-place updates that keep the same ID are invisible. Gaps are fine; going
backwards is not.
+|===
+
+== Look the last success up from Execution information
+
+If the workflow already writes to an
xref:metadata-types/execution-information-location.adoc[Execution Information
Location], you do not have to keep a control table. The
xref:pipeline/transforms/execinfo.adoc[Execution information] transform can
find the previous successful run for you.
+
+Use operation *Find previous successful execution*. It needs two input fields:
+
+* *Execution name* — the pipeline or workflow name as stored in the location
(usually the file name without path)
+* *Execution type* — `Pipeline` or `Workflow` (the `ExecutionType` enum name)
+
+A one-row xref:pipeline/transforms/rowgenerator.adoc[Generate Rows] (or Get
Variables) is enough to feed those fields.
+
+Useful output columns:
+
+* `executionStartDate` / `executionEndDate` — the window bound
+* `failed`, `statusDescription` — sanity checks
+* no output row at all when there is no previous success (first run)
+
+[source,text]
+----
+Generate Rows (name = load_customer, type = Workflow)
+ |
+ v
+Execution information (Find previous successful execution)
+ |
+ v
+default if no row (first run: 1900-01-01)
+ |
+ v
+Table Input (WHERE last_modified > ? AND last_modified <= ?)
+ |
+ v
+Insert / Update
+----
+
+Use *Find previous successful execution*, not *Find last execution*. A failed
run must not move the watermark, or the next success skips the failed window.
+
+The run configuration of the workflow or pipeline you look up must actually
write to that Execution Information Location. Otherwise the transform never
finds a row.
+
+== Other ways to get a date range
+
+These are alternatives to a control table or Execution information, not extra
requirements:
+
+* xref:pipeline/transforms/getsystemdata.adoc[Get System Info] *Start / End
date range (Pipeline)* and *Start / End date range (Workflow)* — the older
log-table window, when those ETL log tables are in use
+* xref:pipeline/transforms/neo4j-getloginfo.adoc[Neo4j Get Logging Info] —
previous (successful) execution dates when logging goes to
`NEO4J_LOGGING_CONNECTION`
+
+== Applying the extract
+
+Filtered selection usually produces *new and changed* rows only.
xref:pipeline/transforms/insertupdate.adoc[Insert / Update] is the typical
apply. There is no `deleted` flag unless you added one in the source query.
+
+See also: xref:how-to-guides/change-data-capture.adoc[Change Data Capture
overview].
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-log-sniffing.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-log-sniffing.adoc
new file mode 100644
index 0000000000..a1e10dde54
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-log-sniffing.adoc
@@ -0,0 +1,149 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+////
+[[CdcLogSniffing]]
+:imagesdir: ../../assets/images
+:description: Ingest change events produced by log sniffers such as Debezium
or Oracle GoldenGate with Apache Hop.
+
+= CDC: log sniffing
+
+Log sniffing is CDC in the original sense: a product reads the database redo
log, write-ahead log, or trail, and emits one event per insert, update, or
delete. Apache Hop does not mine those logs. It *consumes* the events that
Debezium, Oracle GoldenGate, and similar tools already produce.
+
+If no such product is in place, use
xref:how-to-guides/cdc-filtered-selection.adoc[filtered selection] or
xref:how-to-guides/cdc-snapshot-comparison.adoc[snapshot comparison] instead.
+
+== What Hop reads
+
+The log product chooses the landing zone. Hop already has an input transform
for the common ones:
+
+[cols="1,2,2",options="header"]
+|===
+|Landing zone
+|Typical producer
+|Hop transform
+
+|Kafka topic
+|Debezium, GoldenGate for Kafka, and similar
+|xref:pipeline/transforms/kafkaconsumer.adoc[Kafka consumer]
+
+|JMS queue or topic
+|GoldenGate, ESB, or a custom publisher
+|xref:pipeline/transforms/jms-consumer.adoc[JMS consumer]
+
+|Staging / CDC table
+|GoldenGate replicating into a database, or a connector that writes to JDBC
+|xref:pipeline/transforms/tableinput.adoc[Table Input]
+
+|JSON or Avro files
+|Dumped topics, object storage, or trail-to-file
+|xref:pipeline/transforms/jsoninput.adoc[JSON Input],
xref:pipeline/transforms/jsonnormalizeinput.adoc[JSON Normalize Input],
xref:pipeline/transforms/avro-file-input.adoc[Avro File Input],
xref:pipeline/transforms/avro-decode.adoc[Avro Decode]
+|===
+
+This page does not describe how to install or configure Debezium or
GoldenGate. Treat those products as given, and design the Hop pipeline around
the records they emit.
+
+== A typical apply pipeline
+
+[source,text]
+----
+Kafka consumer / JMS consumer / Table Input
+ |
+ v
+JSON Input / Avro Decode / Select Values (parse payload)
+ |
+ v
+Switch / Case or Filter Rows (route on operation)
+ |
+ +-- insert / create / snapshot --> Insert / Update
+ +-- update --> Insert / Update or Update
+ +-- delete / tombstone --> Delete
+----
+
+When the payload already looks like a Merge Rows flag (`new` / `changed` /
`deleted`), xref:pipeline/transforms/synchronizeaftermerge.adoc[Synchronize
after merge] can apply the whole stream in one transform.
+
+== Debezium-style envelopes
+
+Debezium (and several other Kafka CDC connectors) wrap each change in a JSON
or Avro envelope. The fields Hop usually needs are:
+
+[cols="1,3",options="header"]
+|===
+|Field
+|Meaning
+
+|`op`
+|Operation: `c` create, `u` update, `d` delete, `r` snapshot / read
+
+|`after`
+|Row image after the change. Absent or null on deletes.
+
+|`before`
+|Row image before the change. Used for updates and for the key on deletes.
+
+|`source.ts_ms` (and similar)
+|Source commit time, useful for ordering or late-arrival checks
+|===
+
+A compact example (field names vary by connector version and configuration):
+
+[source,json]
+----
+{
+ "op": "u",
+ "before": { "id": 42, "name": "Ada", "city": "London" },
+ "after": { "id": 42, "name": "Ada", "city": "Paris" },
+ "source": { "ts_ms": 1710000000000, "table": "customer" }
+}
+----
+
+Read the Kafka `message` field with
xref:pipeline/transforms/jsoninput.adoc[JSON Input] (or
xref:pipeline/transforms/jsonnormalizeinput.adoc[JSON Normalize Input] when
each record is an object you want flattened). Pull `op`, the key columns, and
the `after.*` payload. For `op = d`, take the key from `before` (or from the
Kafka record key) because `after` is empty.
+
+Route with xref:pipeline/transforms/switchcase.adoc[Switch / Case] on `op`:
+
+* `c` and `r` → insert (or Insert / Update)
+* `u` → update (or Insert / Update)
+* `d` → xref:pipeline/transforms/delete.adoc[Delete]
+
+Snapshot / read events (`r`) appear when a connector dumps the current table.
Treat them as inserts or upserts so a new consumer can catch up.
+
+== GoldenGate-style records
+
+Oracle GoldenGate (and trail-based siblings) typically emit an operation type
plus an after-image, and a before-image when the configuration includes it. The
names differ by adapter:
+
+* Kafka / JSON: fields such as `op_type` (`I` / `U` / `D`), `table`, and
nested before/after objects
+* A replicate table: extra columns for operation, commit timestamp, and trail
position next to the business columns
+
+The Hop pipeline is the same as for Debezium: Table Input or Kafka consumer →
parse → Switch / Case on the operation → apply. If GoldenGate lands only the
after-image of deletes as a key, that is enough for the Delete transform.
+
+== Kafka notes that matter for CDC
+
+The xref:pipeline/transforms/kafkaconsumer.adoc[Kafka consumer] runs a
sub-pipeline that must start with an Injector. Two operating modes are common:
+
+* *Long-lived*: leave the consumer running and process batches by *Duration*
and/or *Number of records*. Use this for near-real-time apply.
+* *Scheduled drain*: enable *Stop when idle* (and an idle timeout) so a
workflow can start the pipeline, empty the topic, and finish. That is the
“queue + schedule” pattern.
+
+Prefer *Offset management: when batch completed* so a failed sub-pipeline does
not commit past records it never applied.
+
+A consumer group assigns partitions. Do not assume a global order across
partitions. Apply by primary key so a later event for the same key wins even if
another partition is behind.
+
+Delete events may be Kafka *tombstones* (a key with a null payload). Map the
key and treat a null message as a delete.
+
+== Idempotency
+
+Replay is normal: a consumer restarts, a connector rewinds, or you re-process
a staging table. Apply by primary key (Insert / Update, or a merge on the
target) so the same event applied twice does not duplicate a row. Do not use a
plain Table Output insert unless the target rejects duplicates.
+
+== When not to use this approach
+
+Hop will not read a redo log, WAL, or GoldenGate trail by itself. If the
source has no CDC product, pick
xref:how-to-guides/cdc-filtered-selection.adoc[filtered selection] or
xref:how-to-guides/cdc-snapshot-comparison.adoc[snapshot comparison].
+
+See also: xref:how-to-guides/change-data-capture.adoc[Change Data Capture
overview].
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-snapshot-comparison.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-snapshot-comparison.adoc
new file mode 100644
index 0000000000..8243e80944
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/cdc-snapshot-comparison.adoc
@@ -0,0 +1,161 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+////
+[[CdcSnapshotComparison]]
+:imagesdir: ../../assets/images
+:description: Detect new, changed, identical and deleted rows by comparing two
sorted snapshots in Apache Hop, or by storing a hash key.
+
+= CDC: comparing snapshots
+
+When the source has no change log and no reliable timestamp or increasing ID,
CDC is a comparison of two full extracts: the set you already have (reference)
and the set you just read (compare).
+
+Apache Hop does this in two ways:
+
+* xref:pipeline/transforms/mergerows.adoc[Merge Rows (diff)] compares two
streams that are *sorted on the same key* and flags every row as `new`,
`changed`, `identical`, or `deleted`. It is built for large snapshots —
millions of rows — in a single sequential pass.
+* A stored *hash key* (MD5, SHA-1, SHA-256, …) shrinks a wide row to one
column. The next run compares only `id` + hash, either with Merge Rows or
against a table you keep.
+
+Both detect deletes, which
xref:how-to-guides/cdc-filtered-selection.adoc[filtered selection] generally
cannot.
+
+== Merge Rows (diff)
+
+Use this when you can produce two extracts with the same layout and a stable
key: today’s file versus yesterday’s, source table versus target table, or this
run versus a snapshot you saved last time.
+
+[source,text]
+----
+Table Input / Text file input Table Input / Text file input
+(previous snapshot) (new snapshot)
+ | |
+ v v
+ Sort Rows (key) Sort Rows (key)
+ | |
+ +--------------+-------------------+
+ |
+ v
+ Merge Rows (diff)
+ keys = id
+ values = the columns that define “changed”
+ |
+ v
+ Synchronize after merge or Switch / Case on the flag
+----
+
+=== Sort first
+
+Merge Rows is a *streaming* compare. It does not sort. Both hops into it must
already be ordered on the *Keys to match*, same direction, same collation. Use
xref:pipeline/transforms/sort.adoc[Sort Rows] on each stream. For large sorts,
run several copies of Sort Rows and combine them with
xref:pipeline/transforms/sortedmerge.adoc[Sorted Merge].
+
+If either stream is out of order, flags are wrong and you will insert, update,
or delete the wrong rows.
+
+=== Configure the compare
+
+[cols="1,3",options="header"]
+|===
+|Option
+|CDC meaning
+
+|Reference rows origin
+|The previous snapshot, or the current target. “What we already applied.”
+
+|Compare rows origin
+|The new snapshot, or the current source. “What we have now.”
+
+|Keys to match
+|The ID that identifies a row across snapshots. Do not put these in *Values to
compare*.
+
+|Values to compare
+|The payload columns. A difference here yields `changed`.
+
+|Flag field name
+|Output field: `identical`, `changed`, `new`, or `deleted`.
+
+|Difference field name
+|Optional JSON of `{ from, to }` per changed column. Useful for audit or for a
Type-2 dimension.
+|===
+
+Flag semantics (from the Merge Rows documentation):
+
+* `identical` — key in both streams, compared values match. Output is the
reference row.
+* `changed` — key in both streams, at least one compared value differs. Output
is the compare row.
+* `new` — key only in the compare stream.
+* `deleted` — key only in the reference stream.
+
+=== Apply the flags
+
+The shortest apply path is
xref:pipeline/transforms/synchronizeaftermerge.adoc[Synchronize after merge],
which defaults to:
+
+* insert when the flag is `new`
+* update when the flag is `changed`
+* delete when the flag is `deleted`
+* do nothing for `identical`
+
+Alternatively, xref:pipeline/transforms/switchcase.adoc[Switch / Case] on the
flag and send each branch to Insert / Update, Update, or Delete.
+
+=== Scale
+
+Comparison itself walks both sorted streams once and only keeps the current
pair of rows in memory. The expensive part is usually the sort, not Merge Rows.
That is why this path stays practical at millions of rows: sort (possibly in
parallel copies), then merge-compare, then apply only the non-identical rows.
+
+xref:pipeline/transforms/tablecompare.adoc[Table Compare] is a different tool.
It audits two database tables and reports differences; it is not the
pipeline-oriented apply path described here.
+
+== Hash keys
+
+Comparing dozens of columns on every row works, but you can also fingerprint
the payload and compare one string.
+
+Use a hash when:
+
+* the row is wide and you only need “did anything change?”
+* you want to persist a compact fingerprint in a database table and avoid
re-reading the previous full snapshot
+* several pipelines should share the same change-detection key
+
+Do *not* use CRC32 or Adler-32 as a business change key. They are short
checksums, not hashes. Prefer *MD5*, *SHA-1*, or *SHA-256* on the
xref:pipeline/transforms/addchecksum.adoc[Add a checksum] transform.
+
+=== Compute and store
+
+[source,text]
+----
+Table Input (source)
+ |
+ v
+Add a checksum (SHA-256 of the payload columns;
+ set Separator so concatenated fields stay unique)
+ |
+ v
+ (optional) Table Output / Insert / Update
+ of id + hash [+ load_ts] into a control table
+----
+
+Set *Prefix*, *Separator*, and *Suffix* on Add a checksum so two different
column combinations cannot produce the same input string. A separator is the
usual requirement; a prefix such as `CUST` distinguishes the same numeric
payload used for different entities.
+
+Any column you leave out of the field list is invisible to change detection.
+
+=== Compare on the next run
+
+Compute hashes on the new extract, then either:
+
+* *Merge Rows* on `id`, with the hash as the only value to compare. Flags stay
`new` / `changed` / `identical` / `deleted`. You no longer need the previous
full snapshot — only the stored `id` + hash table, sorted on `id`.
+* *Database lookup* or *Merge Join* of the new `id` + hash against the control
table, then Filter Rows where the hash differs or the lookup misses.
+
+After a successful apply, update the control table: insert hashes for `new`
keys, replace hashes for `changed` keys, delete hashes for `deleted` keys.
+
+=== Trade-offs
+
+* A hash does not tell you *which* column changed. Keep Merge Rows’ JSON
difference field (on the full payload, for `changed` rows only) if you need
that.
+* SHA-256 collisions are not a practical concern for this use.
+* You still need a stable key. A hash without an ID cannot tell `new` from
`deleted`.
+
+== When not to use this approach
+
+If a log product already emits change events,
xref:how-to-guides/cdc-log-sniffing.adoc[log sniffing] is cheaper than reading
two full sets. If a timestamp or increasing ID covers every change you care
about *and* you do not need deletes,
xref:how-to-guides/cdc-filtered-selection.adoc[filtered selection] is a single
range query.
+
+See also: xref:how-to-guides/change-data-capture.adoc[Change Data Capture
overview].
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/change-data-capture.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/change-data-capture.adoc
new file mode 100644
index 0000000000..3818d7901f
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/change-data-capture.adoc
@@ -0,0 +1,91 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+////
+[[ChangeDataCapture]]
+:imagesdir: ../../assets/images
+:description: Choose among the three Change Data Capture approaches Apache Hop
supports: log sniffing, filtered selection, and snapshot comparison.
+
+= Change Data Capture
+
+Apache Hop can load a data warehouse or another target from *changes* in a
source, not only from full extracts. That is Change Data Capture (CDC): detect
inserts, updates and deletes, then apply only those rows downstream.
+
+Hop does not ship a single “CDC transform”. There is also no built-in Debezium
or GoldenGate connector. CDC in Hop is a *choice of approach*, depending on
what the source can already tell you:
+
+* a change log produced by another product
+* a column that only moves forward (a timestamp or an increasing ID)
+* nothing but two full snapshots you can compare
+
+The chat-level picture of “a queue plus a schedule that reads deltas” is only
one of those three. The pages below treat each pattern separately.
+
+== The three approaches
+
+[cols="1,2,2,1,1,2",options="header"]
+|===
+|Approach
+|Source of truth
+|Typical Hop entry point
+|Detects deletes?
+|Latency
+|Cost
+
+|xref:how-to-guides/cdc-log-sniffing.adoc[Log sniffing]
+|Redo / WAL / trail consumed by Debezium, Oracle GoldenGate, and similar tools
+|xref:pipeline/transforms/kafkaconsumer.adoc[Kafka consumer],
xref:pipeline/transforms/jms-consumer.adoc[JMS consumer], or
xref:pipeline/transforms/tableinput.adoc[Table Input]
+|Yes, when the log emits delete (or tombstone) events
+|Near real-time, or a scheduled drain of the topic
+|Whatever the log product costs to run; Hop only consumes the events
+
+|xref:how-to-guides/cdc-filtered-selection.adoc[Filtered selection]
+|A monotonic timestamp or an always-increasing ID
+|A watermark you wrote on the last *successful* run, then Table Input with a
`WHERE` clause
+|Only if the source marks deletes (a flag or a delete table)
+|The batch interval of the workflow
+|One cheap range query
+
+|xref:how-to-guides/cdc-snapshot-comparison.adoc[Comparing snapshots]
+|Two full extracts, sorted on a key
+|xref:pipeline/transforms/mergerows.adoc[Merge Rows (diff)], or a stored hash
key (MD5 / SHA)
+|Yes
+|Batch, after both extracts are available
+|Read (and usually sort) both full sets
+|===
+
+== How to choose
+
+Use *log sniffing* when a CDC product already writes change events to Kafka,
JMS, files, or a staging table. Hop’s job is to parse those records and apply
them. Hop is not a log miner.
+
+Use *filtered selection* when every insert or update you care about touches a
timestamp or an always-increasing ID, *and* you can answer “what is the last
time this workflow or pipeline ran successfully?”. The simplest answer is to
write that watermark yourself at the end of the success path.
xref:pipeline/transforms/execinfo.adoc[Execution information] can look the same
fact up from Hop’s own execution store.
+
+Use *snapshot comparison* when there is no change log and no reliable
watermark: daily files, legacy systems, or “today’s extract versus yesterday”.
xref:pipeline/transforms/mergerows.adoc[Merge Rows (diff)] compares two sorted
streams of millions of rows in one pass and flags each row as `new`, `changed`,
`identical`, or `deleted`.
+
+== Applying the changes
+
+All three approaches end the same way: you have rows that should be inserted,
updated, or deleted on a target. Common apply transforms are:
+
+* xref:pipeline/transforms/insertupdate.adoc[Insert / Update] for upserts when
deletes are out of scope
+* xref:pipeline/transforms/delete.adoc[Delete] when you have keys that
disappeared
+* xref:pipeline/transforms/synchronizeaftermerge.adoc[Synchronize after merge]
when Merge Rows already produced an `identical` / `changed` / `new` / `deleted`
flag
+* xref:pipeline/transforms/dimensionlookup.adoc[Dimension lookup/update] when
the *target* is a slowly changing dimension
+
+NOTE: Slowly changing dimensions are a *target* pattern. They do not detect
source changes. Pair them with one of the three CDC approaches above.
+
+== Further reading
+
+* xref:how-to-guides/cdc-log-sniffing.adoc[CDC: log sniffing]
+* xref:how-to-guides/cdc-filtered-selection.adoc[CDC: filtered selection]
+* xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots]
+* xref:how-to-guides/joins-lookups.adoc[Joins and lookups] — Merge Rows (diff)
in the broader join landscape
+* xref:metadata-types/execution-information-location.adoc[Execution
Information Location]
diff --git a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/index.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/index.adoc
index ecaeeaa86e..59cc499a06 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/index.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/index.adoc
@@ -22,6 +22,10 @@ under the License.
This page contains a collection of how-to guides to perform a variety of
tasks, configurations etc with Apache Hop.
+* xref:how-to-guides/change-data-capture.adoc[Change Data Capture]
+** xref:how-to-guides/cdc-log-sniffing.adoc[CDC: log sniffing]
+** xref:how-to-guides/cdc-filtered-selection.adoc[CDC: filtered selection]
+** xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots]
* xref:how-to-guides/joins-lookups.adoc[Joins and lookups in Apache Hop]
* xref:how-to-guides/loops-in-apache-hop.adoc[Loops in Apache Hop]
* xref:how-to-guides/run-hop-in-apache-airflow.adoc[Run Pipelines and
Workflows from Apache Airflow]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/joins-lookups.adoc
b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/joins-lookups.adoc
index 245811730d..af978013b0 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/joins-lookups.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/how-to-guides/joins-lookups.adoc
@@ -55,7 +55,7 @@ classic merge join between data sets with data coming from
two different input t
compares and merges data within two rows of data, adding a field indicator for
each row
|You have two sets of data with the same layout. You want to compare both data
sets and identify which lines in one set a new, identical, changed or deleted
compared to the other data set.
-**example***: you receive a daily file that needs to be processed. Compare
today's file with yesterday's file, so you can ignore the identical data and
only process the changes.
+**example***: you receive a daily file that needs to be processed. Compare
today's file with yesterday's file, so you can ignore the identical data and
only process the changes. That pattern is snapshot
xref:how-to-guides/cdc-snapshot-comparison.adoc[Change Data Capture].
|xref:pipeline/transforms/multimerge.adoc[**Multiway merge join**]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addchecksum.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addchecksum.adoc
index c65ed5e6b9..1abb7cb157 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addchecksum.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addchecksum.adoc
@@ -66,3 +66,8 @@ Like the prefix, this can be used to distinguish hash keys
from different topics
|Fields used in the checksum|The names of the fields to include in the
checksum calculation.
Note: You can use the "Get Fields" button to insert all input fields from
previous transforms.
|===
+
+== See also
+
+* xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots] —
storing an MD5 or SHA hash of a row and comparing it on the next run
+
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
index 2b9cd3f0a3..b7c0c60e71 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
@@ -81,3 +81,61 @@ First use Generate rows to create your fields and values to
use (children: Boole
|The input field that will contain the limit with which to search
|===
+
+== Operation types
+
+The transform always needs input rows. Which input fields are required, and
which columns are added, depends on the operation.
+
+[options="header"]
+|===
+|Operation
+|Required input fields
+|Added output
+
+|Get execution IDs
+|Include children, Limit
+|`id`
+
+|Get execution and state
+|Execution ID
+|Execution and state columns (see below)
+
+|Find executions
+|Execution parent ID
+|One output row per child execution, with execution and state columns
+
+|Find last execution
+|Execution name, Execution type
+|Execution and state columns for the most recent matching run, or no row
+
+|Find previous successful execution
+|Execution name, Execution type
+|Execution and state columns for the most recent matching run that did *not*
fail, or no row
+
+|Find child IDs
+|Execution parent ID, Execution type
+|`id`
+
+|Find parent ID
+|Execution ID
+|`id`
+
+|Get execution data
+|Execution ID, Execution parent ID
+|Collected data-profile rows
+
+|Delete execution
+|Execution ID
+|`deleted` (Boolean)
+|===
+
+Execution *type* is the `ExecutionType` name: `Pipeline`, `Workflow`,
`Transform`, or `Action`.
+
+Find last execution / Find previous successful execution / Get execution and
state add:
+
+`executionId`, `parentId`, `name`, `executionType`, `filename`, `executorXml`,
`metadataJson`, `registrationDate`, `executionStartDate`,
`runConfigurationName`, `logLevel`, `updateTime`, `loggingText`, `failed`,
`statusDescription`, `executionEndDate`.
+
+If no execution matches, the transform writes no output row for that input.
+
+*Find previous successful execution* is the operation to use for incremental
extracts: it answers “when did this workflow or pipeline last succeed?” so a
xref:pipeline/transforms/tableinput.adoc[Table Input] can filter on a timestamp
or increasing ID. How to wire that (or write the watermark yourself on the
success path) is in xref:how-to-guides/cdc-filtered-selection.adoc[CDC:
filtered selection].
+
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
index 94808fa8fb..6fd1f26d49 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
@@ -230,4 +230,8 @@ Here are some options you need to set to make this work on
a Confluent Cloud Kaf
|acks
|ALL
-|===
\ No newline at end of file
+|===
+
+== See also
+
+* xref:how-to-guides/cdc-log-sniffing.adoc[CDC: log sniffing] — consuming
Debezium, GoldenGate, or similar change events from a Kafka topic
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
index ff90515e2f..b09b7d704d 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
@@ -112,3 +112,8 @@ If you specify a field name for the differences in JSON,
you will changes appear
]
}
----
+
+== See also
+
+* xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots] —
using this transform (or a stored hash key) as a Change Data Capture approach
+* xref:pipeline/transforms/synchronizeaftermerge.adoc[Synchronize after merge]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/synchronizeaftermerge.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/synchronizeaftermerge.adoc
index efe42e6f52..45adc1cefa 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/synchronizeaftermerge.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/synchronizeaftermerge.adoc
@@ -80,3 +80,8 @@ This field is used by the transform to obtain an operation
flag for the current
If the lookup field is not found, then an exception is thrown.
This option can be used as an extra check if you wish to check updates/deletes
prior to their execution.|Not applicable
|===
+
+== See also
+
+* xref:pipeline/transforms/mergerows.adoc[Merge rows (diff)]
+* xref:how-to-guides/cdc-snapshot-comparison.adoc[CDC: comparing snapshots]
