“

Series: S3 Data Foundation · Article 11 | Level: Advanced | Reading Time: 20 min

Diff Queries: Branch Comparison and Change Tracking

Tags: #DiffQuery #BranchDiff #ChangeTracking #Nessie #Audit #coomia-dip

#TL;DR

Diff queries are an advanced application of the coomia-dip platform's time travel capability — beyond viewing historical states, they precisely pinpoint "what changed." This article dives deep into Diff query implementation under Nessie's branching model, covering multi-branch comparison semantics, three Diff algorithms (full scan, incremental file, Iceberg Changelog), schema-evolution-compatible Diff, structured output and visualization of Diff results, merge conflict detection and resolution strategies, and change subscription and audit compliance applications built on Diff.

#1. Diff Query Scenario Definitions

#1.1 Diff Query Types

Code

coomia-dip Diff Query Type Matrix:

+-----------------+-------------------+--------------------+
| Diff Type        | Comparison Axis    | OQL Syntax          |
+-----------------+-------------------+--------------------+
| Time Diff        | Same branch,       | DIFF Entity          |
|                  | different times    | FROM TIME t1         |
|                  |                    | TO TIME t2           |
+-----------------+-------------------+--------------------+
| Branch Diff      | Different branches,| DIFF Entity          |
|                  | same time          | FROM BRANCH b1       |
|                  |                    | TO BRANCH b2         |
+-----------------+-------------------+--------------------+
| Mixed Diff       | Different branches | DIFF Entity          |
|                  | + different times  | FROM BRANCH b1       |
|                  |                    |   AT TIME t1         |
|                  |                    | TO BRANCH b2         |
|                  |                    |   AT TIME t2         |
+-----------------+-------------------+--------------------+
| Entity Diff      | Single entity      | DIFF Entity          |
|                  | change history     | WHERE id = 'x'       |
|                  |                    | FROM TIME t1         |
|                  |                    | TO TIME t2           |
+-----------------+-------------------+--------------------+

#1.2 Nessie Branching Model Review

Code

Nessie Branches and Diff Relationships:

main --*--*--*--*--*--*--*--*--*-- HEAD
        \           \
         \           \--*--*--* feature-b
          \
           \--*--*--*--*-- feature-a

Diff operations:
  1. main vs feature-a  -> Branch Diff (changes since fork point)
  2. feature-a vs feature-b -> Cross-branch Diff
  3. main@t1 vs main@t2 -> Time Diff (same branch, different times)
  4. merge(feature-a, main) -> Pre-merge conflict detection Diff

#2. Diff Algorithm Implementation

#2.1 Algorithm 1: Full Scan Comparison

Python

class FullScanDiffAlgorithm:
    """Full scan comparison — simplest but slowest"""

    async def diff(
        self,
        entity_type: str,
        from_ref: DataRef,
        to_ref: DataRef,
        columns: list[str] | None = None
    ) -> DiffResult:
        from_data = await self._read_full(entity_type, from_ref)
        to_data = await self._read_full(entity_type, to_ref)

        from_map = {row['entity_id']: row for row in from_data}
        to_map = {row['entity_id']: row for row in to_data}

        changes = []

        for eid, to_row in to_map.items():
            if eid not in from_map:
                changes.append(Change('ADDED', eid, None, to_row))
            else:
                from_row = from_map[eid]
                diff_cols = self._diff_columns(from_row, to_row, columns)
                if diff_cols:
                    changes.append(Change('MODIFIED', eid,
                        {c: from_row[c] for c in diff_cols},
                        {c: to_row[c] for c in diff_cols}
                    ))

        for eid in from_map:
            if eid not in to_map:
                changes.append(Change('DELETED', eid, from_map[eid], None))

        return DiffResult(changes=changes)

Code

Full scan algorithm characteristics:
  Time complexity:  O(N + M), N and M are row counts
  Space complexity: O(N + M)
  Pros: Simple, correct
  Cons: Reads all data, slow for large datasets

Best for:
  - Data volume < 100K rows
  - First-time Diff (no historical incremental info)

#2.2 Algorithm 2: Incremental File Comparison

Python

class IncrementalFileDiffAlgorithm:
    """Incremental file comparison — leverages Iceberg snapshot diffs"""

    async def diff(
        self,
        entity_type: str,
        from_snapshot: int,
        to_snapshot: int,
        columns: list[str] | None = None
    ) -> DiffResult:
        table = self._catalog.load_table(entity_type)

        added_files = []
        deleted_files = []

        for entry in table.inspect.entries(
            from_snapshot_id=from_snapshot,
            to_snapshot_id=to_snapshot
        ):
            if entry.status == 'ADDED':
                added_files.append(entry.data_file)
            elif entry.status == 'DELETED':
                deleted_files.append(entry.data_file)

        added_data = self._read_files(added_files, columns)
        deleted_data = self._read_files(deleted_files, columns)

        added_index = {r['entity_id']: r for r in added_data}
        deleted_index = {r['entity_id']: r for r in deleted_data}

        changes = []

        # In both sets -> MODIFIED
        for eid in set(added_index) & set(deleted_index):
            old_row = deleted_index[eid]
            new_row = added_index[eid]
            diff_cols = self._diff_columns(old_row, new_row, columns)
            if diff_cols:
                changes.append(Change('MODIFIED', eid,
                    {c: old_row[c] for c in diff_cols},
                    {c: new_row[c] for c in diff_cols}
                ))

        # Only in added -> ADDED
        for eid in set(added_index) - set(deleted_index):
            changes.append(Change('ADDED', eid, None, added_index[eid]))

        # Only in deleted -> DELETED
        for eid in set(deleted_index) - set(added_index):
            changes.append(Change('DELETED', eid, deleted_index[eid], None))

        return DiffResult(changes=changes)

Code

Incremental file algorithm characteristics:
  Time complexity:  O(C), C is the number of changed rows
  Space complexity: O(C)
  Pros: Reads only changed portions, efficient for large datasets
  Cons: Depends on Iceberg file-level tracking

Best for:
  - Data volume > 100K rows
  - Changes < 10% of total
  - Comparing two snapshots of the same Iceberg table

#2.3 Algorithm 3: Changelog Comparison

Python

class ChangelogDiffAlgorithm:
    """Diff algorithm based on Iceberg Changelog"""

    async def diff(
        self,
        entity_type: str,
        from_snapshot: int,
        to_snapshot: int
    ) -> DiffResult:
        table = self._catalog.load_table(entity_type)

        changelog = table.inspect.changelog(
            start_snapshot_id=from_snapshot,
            end_snapshot_id=to_snapshot
        )

        changes = []
        for entry in changelog:
            if entry.operation == 'INSERT':
                changes.append(Change('ADDED',
                    entry.record['entity_id'], None, entry.record))
            elif entry.operation == 'DELETE':
                changes.append(Change('DELETED',
                    entry.record['entity_id'], entry.record, None))
            elif entry.operation == 'UPDATE':
                changes.append(Change('MODIFIED',
                    entry.record['entity_id'], entry.before, entry.after))

        return DiffResult(changes=changes)

#3. Schema-Evolution-Compatible Diff

#3.1 Schema Change Types

Code

Schema Evolution Impact on Diff:

+-----------------+------------------------------------+
| Schema Change    | Diff Handling Strategy              |
+-----------------+------------------------------------+
| Add column       | New version has value, old = NULL   |
| Drop column      | Old version has value, new = DROP   |
| Rename column    | Map via Iceberg field-id            |
| Type change      | Auto promotion (int->long, etc.)    |
| Add Entity Type  | All marked as ADDED                 |
| Drop Entity Type | All marked as DELETED               |
+-----------------+------------------------------------+

#3.2 Compatible Diff Implementation

Python

class SchemaAwareDiffEngine:
    """Schema-aware Diff engine"""

    def diff_with_schema_evolution(
        self,
        from_data: pa.Table,
        to_data: pa.Table,
        from_schema: IcebergSchema,
        to_schema: IcebergSchema
    ) -> DiffResult:
        unified_schema = self._unify_schemas(from_schema, to_schema)

        from_projected = self._project_to_schema(
            from_data, unified_schema
        )
        to_projected = self._project_to_schema(
            to_data, unified_schema
        )

        column_mapping = self._build_field_id_mapping(
            from_schema, to_schema
        )

        changes = []
        for eid, from_row, to_row in self._aligned_rows(
            from_projected, to_projected
        ):
            schema_changes = []
            value_changes = []

            for col in unified_schema.columns:
                from_val = from_row.get(col.field_id)
                to_val = to_row.get(col.field_id)

                if (col.field_id in from_schema
                        and col.field_id not in to_schema):
                    schema_changes.append(
                        SchemaChange('COLUMN_DROPPED', col.name)
                    )
                elif (col.field_id not in from_schema
                        and col.field_id in to_schema):
                    schema_changes.append(
                        SchemaChange('COLUMN_ADDED', col.name)
                    )
                elif from_val != to_val:
                    value_changes.append(
                        ValueChange(col.name, from_val, to_val)
                    )

            if schema_changes or value_changes:
                changes.append(EnrichedChange(
                    entity_id=eid,
                    value_changes=value_changes,
                    schema_changes=schema_changes
                ))

        return DiffResult(changes=changes)

#4. Merge Conflict Detection

#4.1 Three-Way Diff

Code

Three-Way Diff for Merge Conflict Detection:

       base (common ancestor)
      /              \
     /                \
branch-a              branch-b
     \                /
      \              /
       merge (merge point)

Three-Way Diff Logic:
  For each entity_id:
    base_val = base branch value
    a_val    = branch-a value
    b_val    = branch-b value

    Case 1: base = a = b         -> No change
    Case 2: base != a, base = b  -> a modified, take a
    Case 3: base = a, base != b  -> b modified, take b
    Case 4: base != a, base != b, a = b -> Both same, take a
    Case 5: base != a, base != b, a != b -> CONFLICT!

#4.2 Conflict Detection Implementation

Python

class ThreeWayDiff:
    """Three-way Diff conflict detection"""

    def detect_conflicts(
        self,
        base_data: dict[str, dict],
        branch_a_data: dict[str, dict],
        branch_b_data: dict[str, dict]
    ) -> MergeResult:
        all_ids = (set(base_data) | set(branch_a_data)
                   | set(branch_b_data))

        auto_resolved = []
        conflicts = []

        for eid in all_ids:
            base = base_data.get(eid)
            a = branch_a_data.get(eid)
            b = branch_b_data.get(eid)

            if base == a == b:
                continue

            if base == a and base != b:
                auto_resolved.append(Resolution(eid, 'TAKE_B', b))
            elif base != a and base == b:
                auto_resolved.append(Resolution(eid, 'TAKE_A', a))
            elif a == b:
                auto_resolved.append(Resolution(eid, 'TAKE_BOTH', a))
            else:
                prop_conflicts = self._detect_property_conflicts(
                    base, a, b
                )
                if prop_conflicts:
                    conflicts.append(Conflict(
                        entity_id=eid,
                        base_value=base,
                        branch_a_value=a,
                        branch_b_value=b,
                        conflicting_properties=prop_conflicts
                    ))
                else:
                    merged = self._auto_merge_properties(base, a, b)
                    auto_resolved.append(
                        Resolution(eid, 'AUTO_MERGE', merged)
                    )

        return MergeResult(
            auto_resolved=auto_resolved,
            conflicts=conflicts,
            has_conflicts=len(conflicts) > 0
        )

#4.3 Conflict Resolution Strategies

Code

Conflict Resolution Strategies:

+------------------+----------------------------------+
| Strategy          | Description                       |
+------------------+----------------------------------+
| TAKE_SOURCE       | Keep source branch (merge from)   |
| TAKE_TARGET       | Keep target branch (merge into)   |
| TAKE_LATEST       | Take most recently modified       |
| MANUAL            | Human decides                     |
| PROPERTY_LEVEL    | Auto-merge at property level      |
|                   | (each property takes its modifier)|
| CUSTOM_FUNCTION   | Custom merge function             |
+------------------+----------------------------------+

#5. Diff Result Structured Output

#5.1 Output Formats

Python

@dataclass
class DiffResult:
    """Diff query structured result"""
    from_ref: str
    to_ref: str
    entity_type: str
    changes: list[DiffEntry]
    summary: DiffSummary
    metadata: DiffMetadata

    def to_table(self) -> pa.Table:
        """Convert to Arrow Table"""
        return pa.table({
            'entity_id': [c.entity_id for c in self.changes],
            'change_type': [c.change_type for c in self.changes],
            'old_value': [
                json.dumps(c.old_value) for c in self.changes
            ],
            'new_value': [
                json.dumps(c.new_value) for c in self.changes
            ],
        })

    def to_json(self) -> dict:
        return {
            'from': self.from_ref,
            'to': self.to_ref,
            'entity_type': self.entity_type,
            'summary': {
                'added': self.summary.added,
                'modified': self.summary.modified,
                'deleted': self.summary.deleted,
            },
            'changes': [
                {
                    'entity_id': c.entity_id,
                    'change_type': c.change_type,
                    'old_value': c.old_value,
                    'new_value': c.new_value,
                }
                for c in self.changes
            ]
        }

#5.2 Diff Visualization

Code

Diff Visualization Output Example:

DIFF Customer FROM '2024-03-01' TO '2024-03-31'

Change Summary:
  ==================== 85% Unchanged (850/1000)
  ====                  8% Modified  (80)
  ==                    5% Added     (50)
  =                     2% Deleted   (20)

Property Change Heatmap:
+-------------+----------+-----------+
| Property     | Changes   | Intensity |
+-------------+----------+-----------+
| credit_rating| 45       | ======== |
| risk_level   | 30       | =====    |
| address      | 25       | ====     |
| phone        | 15       | ==       |
| name         | 5        | =        |
+-------------+----------+-----------+

Time Distribution:
  3/01-3/07: =======   35 changes
  3/08-3/14: =====     25 changes
  3/15-3/21: ========  40 changes
  3/22-3/28: ========= 45 changes
  3/29-3/31: ==        5 changes

#6. Change Subscription

#6.1 Diff-Based Change Notifications

Python

class DiffBasedSubscription:
    """Diff-based change subscription system"""

    def __init__(self, diff_engine: DiffEngine,
                 notification_service: NotificationService):
        self._diff_engine = diff_engine
        self._notifier = notification_service
        self._subscriptions: list[Subscription] = []

    async def check_and_notify(self):
        """Periodically check changes and notify subscribers"""
        for sub in self._subscriptions:
            last_check = sub.last_check_snapshot
            current = await self._get_current_snapshot(
                sub.entity_type
            )

            if current == last_check:
                continue

            diff = await self._diff_engine.diff(
                entity_type=sub.entity_type,
                from_snapshot=last_check,
                to_snapshot=current,
                columns=sub.watched_properties
            )

            if diff.summary.total > 0:
                matched = self._filter_changes(
                    diff.changes, sub.filter
                )
                if matched:
                    await self._notifier.send(
                        subscriber=sub.subscriber,
                        notification=ChangeNotification(
                            entity_type=sub.entity_type,
                            changes=matched,
                            summary=DiffSummary(
                                added=sum(
                                    1 for c in matched
                                    if c.change_type == 'ADDED'
                                ),
                                modified=sum(
                                    1 for c in matched
                                    if c.change_type == 'MODIFIED'
                                ),
                                deleted=sum(
                                    1 for c in matched
                                    if c.change_type == 'DELETED'
                                ),
                            )
                        )
                    )

            sub.last_check_snapshot = current

#7. Performance Optimization

#7.1 Diff Caching

Code

Diff Caching Strategy:

Problem: Frequent Diff queries re-read the same snapshots

Solution: Layered cache
  L1: Snapshot metadata cache (Manifest List/File)
      -> Avoid repeated S3 metadata reads
      -> TTL: 1 hour

  L2: File content cache (hot data files as Arrow)
      -> Avoid repeated Parquet deserialization
      -> LRU: 512 MB

  L3: Diff result cache (Diff results for same snapshot pair)
      -> Completely avoid recomputation
      -> Key: (from_snap, to_snap, entity_type, columns)
      -> TTL: Valid until snapshot expires

#7.2 Parallel Diff

Python

class ParallelDiffExecutor:
    """Parallel Diff executor"""

    async def diff_parallel(
        self,
        entity_type: str,
        from_snap: int,
        to_snap: int,
        parallelism: int = 8
    ) -> DiffResult:
        file_groups = self._partition_files(
            entity_type, from_snap, to_snap, parallelism
        )

        tasks = [
            self._diff_file_group(group, from_snap, to_snap)
            for group in file_groups
        ]
        partial_results = await asyncio.gather(*tasks)

        return self._merge_diff_results(partial_results)

#8. Testing Strategy

Python

class TestDiffQueries:

    async def test_time_diff_detects_modifications(self):
        t1 = await snapshot_after(
            insert('Person', 'p1', {'name': 'Alice', 'age': 30})
        )
        t2 = await snapshot_after(update('p1', {'age': 31}))
        diff = await execute_diff('Person', t1, t2)
        assert diff.summary.modified == 1
        assert diff.changes[0].old_value['age'] == 30
        assert diff.changes[0].new_value['age'] == 31

    async def test_branch_diff(self):
        await create_branch('test-branch', from_ref='main')
        await on_branch('test-branch', lambda:
            insert('Person', 'p2', {'name': 'Bob'})
        )
        diff = await execute_diff_branch(
            'Person', 'main', 'test-branch'
        )
        assert diff.summary.added == 1

    async def test_three_way_merge_conflict(self):
        base = await current_snapshot()
        await on_branch('a', lambda: update('p1', {'age': 35}))
        await on_branch('b', lambda: update('p1', {'age': 40}))
        result = three_way_diff(base, 'a', 'b')
        assert result.has_conflicts
        assert 'age' in result.conflicts[0].conflicting_properties

    async def test_schema_evolution_diff(self):
        t1 = await current_snapshot()
        await add_column('Person', 'department', 'STRING')
        await update('p1', {'department': 'Engineering'})
        t2 = await current_snapshot()
        diff = await execute_diff('Person', t1, t2)
        assert any(
            sc.type == 'COLUMN_ADDED'
            for c in diff.changes for sc in c.schema_changes
        )

#Key Takeaways

Diff queries are the core "change awareness" capability: they tell you not just "what the data is" but "what changed," forming the foundation for audit compliance, change tracking, and collaborative development.
Three Diff algorithms suit different scenarios: full scan for small datasets, incremental file comparison leverages Iceberg snapshot diffs for large data efficiency, and Changelog provides the most precise change records.
Schema-evolution Diff requires field-id matching: using Iceberg's field-id rather than column names correctly handles column renames, additions, and deletions.
Three-way Diff supports branch merge conflict detection: ancestor-based three-way comparison precisely identifies property-level conflicts, supporting both auto-merge and manual resolution.
Diff result caching is the performance key: Diff results for the same snapshot pair are immutable and can be cached indefinitely until snapshots expire.

#Next Article

Next up: S3-12 "Analytics Engine: Ontology Wrapper for OLAP Capabilities" will show how Doris's OLAP capabilities are wrapped into Ontology-semantic analytical queries, enabling metric aggregation, multidimensional analysis, and real-time dashboards.

Tags: #DiffQuery #BranchDiff #ChangeTracking #Nessie #ThreeWayMerge #SchemaEvolution #ChangeSubscription #Audit #coomia-dip #DataFoundation