┌─────────────────────────────────────────────────┐
│                  Query Engine Layer               │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌────────────────┐ │
│  │Doris │ │Spark │ │Flink │ │ Trino/Presto   │ │
│  └──┬───┘ └──┬───┘ └──┬───┘ └──────┬─────────┘ │
│     └────────┴────────┴─────────────┘           │
│                    │                             │
│  ┌─────────────────▼───────────────────────────┐│
│  │         Nessie Catalog                       ││
│  │    (Version Control + Metadata Management)   ││
│  └─────────────────┬───────────────────────────┘│
│                    │                             │
│  ┌─────────────────▼───────────────────────────┐│
│  │         Apache Iceberg                       ││
│  │    (Table Format + Snapshots + Schema)       ││
│  │                                              ││
│  │  metadata.json → manifest-list → manifest    ││
│  │       → data files (Parquet)                 ││
│  └─────────────────┬───────────────────────────┘│
│                    │                             │
│  ┌─────────────────▼───────────────────────────┐│
│  │      Object Storage (MinIO/S3)               ││
│  └──────────────────────────────────────────────┘│
└─────────────────────────────────────────────────┘

{
  "format-version": 2,
  "table-uuid": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "location": "s3://coomia-dip-lakehouse/warehouse/ontology_db/objects",
  "last-sequence-number": 42,
  "last-updated-ms": 1711234567890,
  "schemas": [
    {
      "schema-id": 0,
      "type": "struct",
      "fields": [
        {"id": 1, "name": "tenant_id", "type": "string", "required": true},
        {"id": 2, "name": "object_rid", "type": "string", "required": true},
        {"id": 3, "name": "object_type", "type": "string", "required": true},
        {"id": 4, "name": "properties", "type": "string", "required": false},
        {"id": 5, "name": "created_at", "type": "timestamptz", "required": true}
      ]
    }
  ],
  "current-schema-id": 0,
  "partition-specs": [
    {
      "spec-id": 0,
      "fields": [
        {"source-id": 1, "field-id": 1000, "name": "tenant_id", "transform": "identity"},
        {"source-id": 5, "field-id": 1001, "name": "created_at_day", "transform": "day"}
      ]
    }
  ],
  "current-snapshot-id": 123456789,
  "snapshots": ["..."]
}

Snapshot 1 (t0)        Snapshot 2 (t1)        Snapshot 3 (t2)
┌──────────┐           ┌──────────┐           ┌──────────┐
│ ML-001   │           │ ML-002   │           │ ML-003   │
│ ┌──────┐ │           │ ┌──────┐ │           │ ┌──────┐ │
│ │ M-A  │─┼──► F1,F2  │ │ M-A  │─┼──► F1,F2  │ │ M-A  │─┼──► F1,F2
│ └──────┘ │           │ ├──────┤ │           │ ├──────┤ │
│          │           │ │ M-B  │─┼──► F3     │ │ M-B  │─┼──► F3
│          │           │ └──────┘ │           │ ├──────┤ │
│          │           │          │           │ │ M-C  │─┼──► F4 (-F2)
└──────────┘           └──────────┘           │ └──────┘ │
                                              └──────────┘

ML = Manifest List, M = Manifest, F = Data File

Manifest Entry:
  file: "s3://warehouse/data/00001.parquet"
  partition: {tenant_id="acme", created_at_day=2026-03-20}
  record_count: 50000
  column_sizes: {1: 1200000, 2: 800000, 3: 600000}
  null_value_counts: {1: 0, 2: 0, 3: 200}
  lower_bounds: {1: "acme", 2: "obj-00001", 5: "2026-03-20T00:00:00Z"}
  upper_bounds: {1: "acme", 2: "obj-50000", 5: "2026-03-20T23:59:59Z"}

# data-Layer/iceberg/ontology_writer.py
from pyiceberg.catalog import load_catalog
from pyiceberg.expressions import EqualTo

class OntologyObjectWriter:
    """Transactional writer for Ontology objects"""

    def __init__(self):
        self.catalog = load_catalog("nessie", **{
            "uri": "http://nessie-server:19120/api/v2",
            "ref": "main",
            "warehouse": "s3://coomia-dip-lakehouse/warehouse",
        })

    def upsert_objects(self, tenant_id: str, objects: list[dict]):
        """Atomically upsert a batch of Ontology objects"""
        table = self.catalog.load_table("ontology_db.ontology_objects")

        with table.transaction() as txn:
            # Delete old versions
            txn.delete(EqualTo("tenant_id", tenant_id))

            # Write new versions
            import pyarrow as pa
            df = pa.Table.from_pylist(objects)
            txn.append(df)
            # Transaction commits atomically

# data-Layer/iceberg/schema_evolution.py
class IcebergSchemaEvolution:
    """Schema evolution operations"""

    def add_derived_property(self, table_name: str, prop_name: str, data_type: str):
        """Add a derived property column to an Ontology type"""
        table = self.catalog.load_table(table_name)
        with table.update_schema() as update:
            update.add_column(prop_name, data_type, doc=f"Derived: {prop_name}")

    def rename_property(self, table_name: str, old_name: str, new_name: str):
        """Rename a property column (existing data files unchanged)"""
        table = self.catalog.load_table(table_name)
        with table.update_schema() as update:
            update.rename_column(old_name, new_name)

    def widen_type(self, table_name: str, column: str, new_type: str):
        """Widen a column type (e.g., int → long)"""
        table = self.catalog.load_table(table_name)
        with table.update_schema() as update:
            update.update_column(column, new_type)

# data-Layer/iceberg/partition_evolution.py
class PartitionEvolution:
    """Online partition strategy evolution"""

    def evolve_to_hourly(self, table_name: str):
        """Evolve from daily to hourly partitioning"""
        table = self.catalog.load_table(table_name)

        # Old data keeps daily partitions; new data uses hourly
        with table.update_spec() as update:
            update.remove_field("created_at_day")
            update.add_field(
                source_column_name="created_at",
                transform="hour",
                name="created_at_hour",
            )
        # No historical data rewrite required!

    def add_bucket_partition(self, table_name: str, column: str, num_buckets: int):
        """Add bucket partitioning for high-cardinality columns"""
        table = self.catalog.load_table(table_name)
        with table.update_spec() as update:
            update.add_field(
                source_column_name=column,
                transform=f"bucket[{num_buckets}]",
                name=f"{column}_bucket",
            )

# data-Layer/iceberg/time_travel.py
class TimeTravelQuery:
    """Time-travel query operations"""

    def query_at_snapshot(self, table_name: str, snapshot_id: int):
        """Query data at a specific snapshot version"""
        table = self.catalog.load_table(table_name)
        scan = table.scan(snapshot_id=snapshot_id)
        return scan.to_pandas()

    def query_at_timestamp(self, table_name: str, timestamp_ms: int):
        """Query data as it existed at a specific timestamp"""
        table = self.catalog.load_table(table_name)

        target_snapshot = None
        for snapshot in table.metadata.snapshots:
            if snapshot.timestamp_ms <= timestamp_ms:
                target_snapshot = snapshot

        if target_snapshot:
            scan = table.scan(snapshot_id=target_snapshot.snapshot_id)
            return scan.to_pandas()

    def incremental_read(self, table_name: str, from_snapshot: int, to_snapshot: int):
        """Read only the changes between two snapshots"""
        table = self.catalog.load_table(table_name)
        scan = table.scan(
            snapshot_id=to_snapshot,
            options={"start-snapshot-id": str(from_snapshot)},
        )
        return scan.to_pandas()

// data-Layer/flink/IcebergCDCSink.java
public class IcebergCDCSink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000);

        MySqlSource<String> source = MySqlSource.<String>builder()
            .hostname("mysql-source")
            .port(3306)
            .databaseList("business_db")
            .tableList("business_db.orders")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        DataStream<RowData> cdcStream = env
            .fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC")
            .map(new CdcToRowDataMapper());

        Map<String, String> catalogProps = Map.of(
            "type", "iceberg",
            "catalog-impl", "org.apache.iceberg.nessie.NessieCatalog",
            "uri", "http://nessie-server:19120/api/v2",
            "ref", "main",
            "warehouse", "s3://coomia-dip-lakehouse/warehouse"
        );

        TableLoader tableLoader = TableLoader.fromCatalog(
            CatalogUtil.buildIcebergCatalog("nessie", catalogProps, new Configuration()),
            TableIdentifier.of("ontology_db", "orders")
        );

        FlinkSink.forRowData(cdcStream)
            .tableLoader(tableLoader)
            .equalityFieldColumns(List.of("order_id"))
            .upsert(true)
            .build();

        env.execute("CDC to Iceberg");
    }
}

-- Create an Iceberg catalog in Doris
CREATE CATALOG iceberg_catalog PROPERTIES (
    "type" = "iceberg",
    "iceberg.catalog.type" = "rest",
    "iceberg.catalog.uri" = "http://nessie-server:19120/api/v2",
    "iceberg.catalog.warehouse" = "s3://coomia-dip-lakehouse/warehouse",
    "s3.endpoint" = "http://minio:9000",
    "s3.access_key" = "${MINIO_ACCESS_KEY}",
    "s3.secret_key" = "${MINIO_SECRET_KEY}"
);

-- Federated query: hot data in Doris, cold data in Iceberg
SELECT
    d.object_type,
    d.display_name,
    i.historical_value,
    i.snapshot_time
FROM doris_catalog.ontology_db.hot_objects d
JOIN iceberg_catalog.ontology_db.cold_objects i
    ON d.object_rid = i.object_rid
WHERE d.tenant_id = 'acme'
  AND i.snapshot_time >= '2026-01-01';

# data-Layer/iceberg/read_optimization.py
class IcebergReadOptimizer:
    """Read optimization configuration"""

    def configure_scan(self, table, filters: dict):
        scan = table.scan()

        # 1. Column pruning — read only needed columns
        scan = scan.select("object_rid", "display_name", "properties")

        # 2. Predicate pushdown — skip files using statistics
        if "tenant_id" in filters:
            scan = scan.filter(EqualTo("tenant_id", filters["tenant_id"]))

        # 3. Partition pruning — skip irrelevant partitions
        if "created_at" in filters:
            scan = scan.filter(
                GreaterThanOrEqual("created_at", filters["created_at"])
            )

        return scan

# data-Layer/iceberg/table_maintenance.py
class IcebergTableMaintenance:
    """Periodic table maintenance"""

    def compact_small_files(self, table_name: str):
        """Merge small files (fixes fragmentation from streaming writes)"""
        table = self.catalog.load_table(table_name)
        table.rewrite_data_files(
            target_size_in_bytes=128 * 1024 * 1024,
            min_file_size_bytes=64 * 1024 * 1024,
            max_file_size_bytes=256 * 1024 * 1024,
        )

    def expire_snapshots(self, table_name: str, retention_days: int = 30):
        """Expire old snapshots to reclaim storage"""
        table = self.catalog.load_table(table_name)
        cutoff = datetime.now() - timedelta(days=retention_days)
        table.expire_snapshots().expire_older_than(
            int(cutoff.timestamp() * 1000)
        ).commit()

    def remove_orphan_files(self, table_name: str):
        """Clean up files not referenced by any snapshot"""
        table = self.catalog.load_table(table_name)
        table.remove_orphan_files(
            older_than=int((datetime.now() - timedelta(days=3)).timestamp() * 1000)
        )

    def rewrite_manifests(self, table_name: str):
        """Rewrite manifest files to optimize query planning"""
        table = self.catalog.load_table(table_name)
        table.rewrite_manifests()

-- Spark SQL: migrate a Hive table to Iceberg in-place
CALL nessie.system.migrate('ontology_db.legacy_objects');

-- Verify migration
DESCRIBE EXTENDED nessie.ontology_db.legacy_objects;

# data-Layer/iceberg/migration.py
class HiveToIcebergMigration:
    """Incremental Hive-to-Iceberg migration"""

    def shadow_table_migration(self, hive_table: str, iceberg_table: str):
        """Shadow table pattern: dual-write until validation passes"""
        # 1. Create Iceberg shadow table (same schema)
        self._create_shadow_table(hive_table, iceberg_table)

        # 2. Full sync historical data
        self._full_sync(hive_table, iceberg_table)

        # 3. Start incremental sync (CDC dual-write)
        self._start_incremental_sync(hive_table, iceberg_table)

        # 4. Validate data consistency
        self._validate_consistency(hive_table, iceberg_table)

        # 5. Switch read routing
        self._switch_reads(iceberg_table)

        # 6. Stop Hive writes
        self._stop_hive_writes(hive_table)

# data-Layer/iceberg/auto_maintenance.py
class IcebergAutoMaintenance:
    """Automated Iceberg table maintenance"""

    def __init__(self):
        self.scheduler = BackgroundScheduler()
        self.maintenance = IcebergTableMaintenance()

    def start(self):
        # Daily at 2 AM: expire old snapshots
        self.scheduler.add_job(
            self._expire_all_tables, 'cron', hour=2, minute=0
        )

        # Daily at 3 AM: compact small files
        self.scheduler.add_job(
            self._compact_all_tables, 'cron', hour=3, minute=0
        )

        # Weekly Sunday at 4 AM: clean orphan files
        self.scheduler.add_job(
            self._cleanup_orphans, 'cron', day_of_week='sun', hour=4
        )

        self.scheduler.start()

write.metadata.delete-after-commit.enabled=true
write.metadata.previous-versions-max=20