“

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

Time Travel: Deep Application of Iceberg Snapshots

Tags: #TimeTravel #Iceberg #Snapshot #VersionedQuery #TemporalData #coomia-dip

#TL;DR

Time Travel is one of the Ontology platform's core capabilities — users can query data at any historical point without maintaining their own history tables. coomia-dip leverages Apache Iceberg's snapshot mechanism, combined with Nessie's version branching model, to implement OQL's AT TIME and AT BRANCH clauses. This article fully dissects the storage principles of time travel, snapshot management strategies, compilation from OQL time travel syntax to Iceberg queries, cross-snapshot Diff queries, performance optimizations (incremental reads, snapshot pruning), and audit trail applications in compliance scenarios.

#1. Why Time Travel Matters

#1.1 Business Scenarios

Code

Core Time Travel Business Scenarios:

Scenario 1: Audit Compliance
  "What was the status of these devices at end of last month?"
  -> FETCH Device AT TIME '2024-02-29T23:59:59Z'

Scenario 2: Root Cause Analysis
  "What were the risk scores 1 hour before the system alert?"
  -> FETCH RiskScore AT TIME '2024-03-01T13:00:00Z'
     WHERE entity_id = 'asset-001'

Scenario 3: Change Comparison
  "Which customers had credit rating changes this month?"
  -> DIFF Customer
     FROM TIME '2024-03-01' TO TIME '2024-03-31'
     SELECT credit_rating

Scenario 4: What-If Analysis
  "What happens if we modify parameters on a branch?"
  -> FETCH Simulation AT BRANCH 'what-if-scenario-1'

Scenario 5: Data Recovery
  "Accidental deletion — need to restore to yesterday"
  -> Roll back to yesterday's snapshot

#1.2 Problems with Traditional Approaches

Code

Traditional Historical Data Approaches:

+-----------------+----------------+----------------+
| Approach         | Pros            | Cons            |
+-----------------+----------------+----------------+
| Manual history  | Simple, direct  | 2x storage,     |
| table           |                 | complex queries  |
+-----------------+----------------+----------------+
| SCD Type 2      | Standardized    | ETL maintenance  |
+-----------------+----------------+----------------+
| CDC logs        | Precise         | Expensive replay |
+-----------------+----------------+----------------+
| DB backups      | Complete        | Coarse, slow     |
+-----------------+----------------+----------------+
| Iceberg         | Zero extra dev  | Needs Iceberg    |
| Snapshots       | Auto-versioning | ecosystem        |
|                 | Any-time query  |                  |
|                 | Dedup (COW/MOR) |                  |
+-----------------+----------------+----------------+

#2. Iceberg Snapshot Mechanism

#2.1 Iceberg Table Structure

Code

Iceberg Table File Structure:

s3://onto-data/warehouse/
+-- entity_common/
    +-- metadata/
    |   +-- v1.metadata.json     <- table-level metadata
    |   +-- v2.metadata.json
    |   +-- snap-1001.avro       <- snapshot 1 manifest list
    |   +-- snap-1002.avro       <- snapshot 2 manifest list
    |   +-- snap-1003.avro
    +-- manifests/
    |   +-- manifest-a.avro      <- manifest file (file list)
    |   +-- manifest-b.avro
    |   +-- manifest-c.avro
    +-- data/
        +-- part-00001.parquet   <- data files (Parquet)
        +-- part-00002.parquet
        +-- part-00003.parquet

Hierarchy:
  Metadata -> Snapshot -> Manifest List -> Manifest File -> Data File

#2.2 Snapshot Chain

Code

Iceberg Snapshot Chain:

Timeline:
  t1 (initial)   t2 (insert)   t3 (update)   t4 (delete)
     |               |               |               |
     v               v               v               v
  Snap-1001      Snap-1002      Snap-1003      Snap-1004
  +------+       +------+       +------+       +------+
  | F1   |       | F1   |       | F1   |       | F1   |
  | F2   |       | F2   |       | F2   |       | F2   |
  |      |       | F3+  |       | F3   |       | F3   |
  |      |       |      |       | F4+  |       | F4   |
  |      |       |      |       |      |       | -F2  |
  +------+       +------+       +------+       +------+

  F1,F2: initial data files
  F3+:   newly added data file
  F4+:   updated data file (replaces some rows in F2)
  -F2:   marks F2 as deleted

  Key: Each snapshot is NOT a full copy, but incremental references
  -> Storage efficiency: only changed portions are stored

#2.3 COW vs MOR

Code

Copy-on-Write vs Merge-on-Read:

Copy-on-Write (COW):
  On write: create new data file replacing old file
  On read:  direct read, no extra overhead
  Best for: read-heavy workloads

Merge-on-Read (MOR):
  On write: append Delete File + Insert File
  On read:  merge Delete + Insert at read time
  Best for: write-heavy workloads

coomia-dip choices:
  entity_common -> MOR (frequent property updates)
  entity_edge   -> COW (edge relationships change less)
  entity_event  -> Append-Only (events are only appended)

#3. OQL Time Travel Compilation

#3.1 AT TIME Compilation

Python

class TimeTravelCompiler:
    """Compiles time travel OQL to Iceberg queries"""

    def compile_at_time(
        self, entity_type: str, timestamp: datetime
    ) -> str:
        """Compile AT TIME clause"""
        snapshot_id = self._find_snapshot_at(entity_type, timestamp)

        return f"""
        SELECT * FROM iceberg_catalog.onto_db.{entity_type}
        FOR VERSION AS OF {snapshot_id}
        """

    def compile_at_time_doris(
        self, entity_type: str, timestamp: datetime
    ) -> str:
        """Use Doris Iceberg time-travel syntax"""
        ts_str = timestamp.strftime('%Y-%m-%d %H:%M:%S')

        return f"""
        SELECT * FROM iceberg_catalog.onto_db.entity_common
        FOR TIME AS OF '{ts_str}'
        WHERE entity_type = '{entity_type}'
        """

    def _find_snapshot_at(
        self, entity_type: str, timestamp: datetime
    ) -> int:
        """Find the nearest snapshot before the specified time"""
        table = self._iceberg_catalog.load_table(
            "onto_db.entity_common"
        )
        snapshots = table.snapshots()

        target_snapshot = None
        for snap in sorted(snapshots, key=lambda s: s.timestamp_ms):
            if snap.timestamp_ms <= timestamp.timestamp() * 1000:
                target_snapshot = snap
            else:
                break

        if target_snapshot is None:
            raise TimeTravelError(
                f"No snapshot found before {timestamp}"
            )

        return target_snapshot.snapshot_id

#3.2 AT BRANCH Compilation

Python

class BranchTimeTravelCompiler:
    """Branch time-travel compiler"""

    def compile_at_branch(
        self, entity_type: str, branch_name: str
    ) -> str:
        """Compile AT BRANCH clause"""
        nessie_ref = self._nessie_client.get_reference(branch_name)

        return f"""
        SELECT * FROM iceberg_catalog.onto_db.entity_common
        FOR VERSION AS OF {nessie_ref.hash}
        WHERE entity_type = '{entity_type}'
        """

    def compile_branch_diff(
        self, entity_type: str,
        from_branch: str, to_branch: str
    ) -> str:
        """Compile branch DIFF query"""
        from_ref = self._nessie_client.get_reference(from_branch)
        to_ref = self._nessie_client.get_reference(to_branch)

        return f"""
        SELECT
            COALESCE(a.entity_id, b.entity_id) AS entity_id,
            CASE
                WHEN a.entity_id IS NULL THEN 'ADDED'
                WHEN b.entity_id IS NULL THEN 'DELETED'
                WHEN a.properties != b.properties THEN 'MODIFIED'
                ELSE 'UNCHANGED'
            END AS change_type,
            a.properties AS old_value,
            b.properties AS new_value
        FROM (
            SELECT * FROM iceberg_catalog.onto_db.entity_common
            FOR VERSION AS OF {from_ref.hash}
            WHERE entity_type = '{entity_type}'
        ) a
        FULL OUTER JOIN (
            SELECT * FROM iceberg_catalog.onto_db.entity_common
            FOR VERSION AS OF {to_ref.hash}
            WHERE entity_type = '{entity_type}'
        ) b ON a.entity_id = b.entity_id
        WHERE a.entity_id IS NULL
           OR b.entity_id IS NULL
           OR a.properties != b.properties
        """

#4. Snapshot Management Strategy

#4.1 Automatic Snapshot Policy

Python

class SnapshotPolicy:
    """Snapshot management policy"""

    def should_create_snapshot(
        self, table: IcebergTable, last_snapshot_time: datetime
    ) -> bool:
        now = datetime.utcnow()
        time_since_last = now - last_snapshot_time

        # Policy 1: Time-interval trigger
        if time_since_last >= self._config.snapshot_interval:
            return True

        # Policy 2: Change-volume trigger
        pending_changes = table.pending_changes_count()
        if pending_changes >= self._config.change_threshold:
            return True

        # Policy 3: Checkpoint times (hourly, daily)
        if self._is_checkpoint_time(now):
            return True

        return False

    def cleanup_snapshots(self, table: IcebergTable):
        """Clean up expired snapshots"""
        now = datetime.utcnow()
        snapshots = table.snapshots()
        keep_snapshots = set()

        for snap in snapshots:
            snap_time = datetime.fromtimestamp(snap.timestamp_ms / 1000)
            age = now - snap_time

            # Rule 1: Keep all snapshots from last 24 hours
            if age < timedelta(hours=24):
                keep_snapshots.add(snap.snapshot_id)
                continue

            # Rule 2: Last 7 days — keep one per hour
            if age < timedelta(days=7):
                if snap_time.minute == 0:
                    keep_snapshots.add(snap.snapshot_id)
                continue

            # Rule 3: Last 90 days — keep one per day
            if age < timedelta(days=90):
                if snap_time.hour == 0 and snap_time.minute == 0:
                    keep_snapshots.add(snap.snapshot_id)
                continue

            # Rule 4: Beyond 90 days — keep one per month
            if snap_time.day == 1 and snap_time.hour == 0:
                keep_snapshots.add(snap.snapshot_id)

        # Always keep the latest snapshot
        keep_snapshots.add(snapshots[-1].snapshot_id)

        for snap in snapshots:
            if snap.snapshot_id not in keep_snapshots:
                table.expire_snapshot(snap.snapshot_id)

#4.2 Retention Configuration

Code

Snapshot Retention Policy:

+------------------+----------+----------------------+
| Time Range        | Granularity| Estimated Count     |
+------------------+----------+----------------------+
| Last 24 hours     | Every commit| ~100 (every 15 min)|
| Last 7 days       | Hourly   | ~144                 |
| Last 90 days      | Daily    | ~83                  |
| Beyond 90 days    | Monthly  | ~12/year             |
+------------------+----------+----------------------+
| Total             |          | ~340/year            |
+------------------+----------+----------------------+

Storage overhead:
  Each snapshot metadata ~10KB
  340 snapshots ~ 3.4 MB metadata
  Data files are shared (dedup); extra storage depends on changes

#5. Performance Optimization

#5.1 Incremental Reads

Python

class IncrementalReader:
    """Read only the differences between two snapshots"""

    def read_changes(
        self, table: IcebergTable,
        from_snapshot: int, to_snapshot: int
    ) -> ChangeSet:
        scan = table.scan(
            snapshot_id=to_snapshot
        ).use_ref(from_snapshot)

        added_files = []
        deleted_files = []

        for manifest in scan.plan_files():
            if manifest.status == ManifestEntryStatus.ADDED:
                added_files.append(manifest.file)
            elif manifest.status == ManifestEntryStatus.DELETED:
                deleted_files.append(manifest.file)

        added_data = self._read_data_files(added_files)
        deleted_data = self._read_data_files(deleted_files)

        return ChangeSet(added=added_data, deleted=deleted_data)

#5.2 Snapshot Pruning

Code

Snapshot Pruning Optimization:

Problem: AT TIME query needs to scan all snapshots to find target
  Linear search over 1000 snapshots -> slow

Optimization: Snapshot Index
  Build B-tree index on snapshot timestamps
  O(log N) lookup for target snapshot

+------------------------------------------+
|         Snapshot Index (B-tree)            |
|                                           |
|  2024-01-01 -> snap-1001                  |
|  2024-01-02 -> snap-1002                  |
|  2024-01-03 -> snap-1003                  |
|  ...                                      |
|  2024-03-15 -> snap-1075                  |
|                                           |
|  Lookup AT TIME '2024-02-15':             |
|  -> Binary search -> snap-1045 (3 comps)  |
+------------------------------------------+

#5.3 Partition-Level Time Travel

Code

Partition-Level Time Travel Optimization:

Standard time travel: read entire table state at time T
  -> Even if querying just one entity_type, all partitions' snapshots

Optimization: Partition-level snapshots
  entity_common partitioned by entity_type
  Each partition tracks snapshot state independently

  FETCH Person AT TIME '2024-01-01'
  -> Only needs entity_type='Person' partition snapshot
  -> Skips Device, Document, Company partitions

Effect: Query time from 2s -> 200ms (90% reduction)

#6. Diff Query Implementation

#6.1 Change Type Detection

Python

class DiffQueryExecutor:
    """Diff query executor"""

    async def execute_diff(
        self,
        entity_type: str,
        from_ref: TimeOrBranchRef,
        to_ref: TimeOrBranchRef,
        properties: list[str] | None = None
    ) -> DiffResult:
        from_data = await self._fetch_at(entity_type, from_ref)
        to_data = await self._fetch_at(entity_type, to_ref)

        from_index = {row['entity_id']: row for row in from_data}
        to_index = {row['entity_id']: row for row in to_data}

        changes = []

        for eid, to_row in to_index.items():
            if eid not in from_index:
                changes.append(DiffEntry(
                    entity_id=eid,
                    change_type='ADDED',
                    old_value=None,
                    new_value=to_row
                ))
            else:
                from_row = from_index[eid]
                modified_props = self._find_modified_properties(
                    from_row, to_row, properties
                )
                if modified_props:
                    changes.append(DiffEntry(
                        entity_id=eid,
                        change_type='MODIFIED',
                        old_value={k: from_row[k] for k in modified_props},
                        new_value={k: to_row[k] for k in modified_props}
                    ))

        for eid in from_index:
            if eid not in to_index:
                changes.append(DiffEntry(
                    entity_id=eid,
                    change_type='DELETED',
                    old_value=from_index[eid],
                    new_value=None
                ))

        return DiffResult(
            from_ref=from_ref,
            to_ref=to_ref,
            entity_type=entity_type,
            changes=changes,
            summary=DiffSummary(
                added=sum(1 for c in changes if c.change_type == 'ADDED'),
                modified=sum(
                    1 for c in changes if c.change_type == 'MODIFIED'
                ),
                deleted=sum(
                    1 for c in changes if c.change_type == 'DELETED'
                ),
            )
        )

#6.2 Diff Query Example

Code

Complete Diff Query Example:

OQL:
  DIFF Customer
  FROM TIME '2024-03-01' TO TIME '2024-03-31'
  SELECT credit_rating, risk_level

Result:
+------------+------------+--------------+--------------+
| entity_id   | change_type| old_value     | new_value     |
+------------+------------+--------------+--------------+
| cust-001    | MODIFIED   | {credit: A}  | {credit: B}  |
| cust-015    | MODIFIED   | {risk: LOW}  | {risk: HIGH} |
| cust-042    | ADDED      | null         | {credit: A,  |
|             |            |              |  risk: LOW}  |
| cust-088    | DELETED    | {credit: C,  | null         |
|             |            |  risk: HIGH} |              |
+------------+------------+--------------+--------------+

Summary:
  Added: 1, Modified: 2, Deleted: 1, Total changes: 4

#7. Compliance and Audit

#7.1 Audit Trail Generation

Python

class AuditTrailGenerator:
    """Auto-generate audit trails via time travel"""

    async def generate_audit_trail(
        self,
        entity_type: str,
        entity_id: str,
        time_range: tuple[datetime, datetime]
    ) -> list[AuditEntry]:
        start, end = time_range
        snapshots = self._get_snapshots_in_range(
            entity_type, start, end
        )

        trail = []
        prev_state = None

        for snap in snapshots:
            current_state = await self._fetch_entity_at_snapshot(
                entity_type, entity_id, snap.snapshot_id
            )

            if prev_state is None:
                if current_state is not None:
                    trail.append(AuditEntry(
                        timestamp=snap.timestamp,
                        action='CREATE',
                        entity_id=entity_id,
                        changes=current_state
                    ))
            elif current_state is None:
                trail.append(AuditEntry(
                    timestamp=snap.timestamp,
                    action='DELETE',
                    entity_id=entity_id,
                    changes=prev_state
                ))
            else:
                diff = self._compute_diff(prev_state, current_state)
                if diff:
                    trail.append(AuditEntry(
                        timestamp=snap.timestamp,
                        action='UPDATE',
                        entity_id=entity_id,
                        changes=diff
                    ))

            prev_state = current_state

        return trail

#8. Testing Strategy

Python

class TestTimeTravel:

    async def test_at_time_returns_historical_state(self):
        await insert_entity('Person', 'p1', {'name': 'Alice', 'age': 30})
        t1 = datetime.utcnow()

        await update_entity('p1', {'age': 31})
        t2 = datetime.utcnow()

        result_t1 = await execute(
            f"FETCH Person WHERE entity_id = 'p1' AT TIME '{t1}'"
        )
        assert result_t1[0]['age'] == 30

        result_t2 = await execute(
            f"FETCH Person WHERE entity_id = 'p1' AT TIME '{t2}'"
        )
        assert result_t2[0]['age'] == 31

    async def test_diff_detects_all_change_types(self):
        t1 = datetime.utcnow()
        await insert_entity('Item', 'i1', {'value': 100})
        await update_entity('i2', {'value': 200})
        await delete_entity('i3')
        t2 = datetime.utcnow()

        diff = await execute(
            f"DIFF Item FROM TIME '{t1}' TO TIME '{t2}'"
        )
        assert any(d.change_type == 'ADDED' for d in diff.changes)
        assert any(d.change_type == 'MODIFIED' for d in diff.changes)
        assert any(d.change_type == 'DELETED' for d in diff.changes)

#Key Takeaways

Iceberg snapshots provide zero-cost time travel for the Ontology platform: no manual history table maintenance — every data change auto-creates a snapshot, and users query any historical moment via the AT TIME clause.
COW and MOR strategies must be chosen based on read/write ratios: entity_common (frequent updates) uses MOR, entity_edge (less updates) uses COW, entity_event (append-only) uses Append-Only.
Snapshot management balances storage cost and temporal precision: recent snapshots at fine granularity (every commit), older ones at coarse granularity (daily/monthly), with annual metadata overhead of only ~3.4 MB.
Diff queries are an advanced time travel application: comparing data between two time points or branches supports change tracking, audit compliance, and What-If analysis.
Incremental reads and partition-level time travel are the performance keys: avoiding full-table snapshot scans through incremental file lists and partition pruning reduces query time by 90%.

#Next Article

Next up: S3-11 "Diff Queries: Branch Comparison and Change Tracking" will dive deeper into Diff queries under Nessie's branching model, including multi-branch merge conflict detection, schema-evolution-compatible Diff, and Diff result visualization.

Tags: #TimeTravel #Iceberg #Snapshot #VersionedQuery #TemporalData #COW #MOR #AuditTrail #DiffQuery #coomia-dip #DataFoundation