[Feature][EnginePlugin] Support Apache Iceberg engine plugin for modern data lake

### Linkis Component

linkis-engineconn-plugins

### What happened

**English:**

Linkis currently lacks native support for Apache Iceberg, the leading open table format for huge analytic datasets. Iceberg has become the de facto standard for modern data lakes, adopted by Netflix, Apple, Adobe, Tencent, ByteDance, and major cloud providers (AWS, Alibaba Cloud, Tencent Cloud).

**Market Demand:**
- **Industry Standard**: Apache top-level project, widely recognized as the data lake table format standard
- **Enterprise Adoption**: Netflix, Apple, Adobe, Tencent, ByteDance use Iceberg for PB-scale data lakes
- **Cloud Native**: AWS, Alibaba Cloud, Tencent Cloud provide managed Iceberg services
- **Active Ecosystem**: Integrated with Spark, Flink, Trino, Hive, and most big data engines
- **Future-Proof**: Replacing traditional Hive tables in modern data architectures

**Technical Advantages:**
- **ACID Transactions**: Full ACID guarantees on data lakes
- **Time Travel**: Query historical snapshots for audit and rollback
- **Schema Evolution**: Add, drop, rename columns without rewriting data
- **Hidden Partitioning**: Automatic partition management, user-friendly
- **Snapshot Isolation**: Concurrent reads and writes without conflicts
- **Incremental Processing**: Efficient change data capture (CDC)

**Strategic Value:**
Iceberg is not just a table format but a foundation for modern data architecture:
- **Data Lake Standard**: Industry converging on Iceberg as the standard
- **Cloud Migration**: Essential for cloud-native data lake deployments
- **Query Performance**: Metadata pruning significantly improves query speed
- **Storage Efficiency**: Partition evolution and compaction reduce storage costs

---

**中文:**

Linkis目前缺乏对Apache Iceberg的原生支持，Iceberg是大规模分析数据集的领先开放表格式。Iceberg已成为现代数据湖的事实标准，被Netflix、Apple、Adobe、腾讯、字节跳动和主要云厂商（AWS、阿里云、腾讯云）采用。

**市场需求:**
- **行业标准**: Apache顶级项目，被公认为数据湖表格式标准
- **企业采用**: Netflix、Apple、Adobe、腾讯、字节跳动使用Iceberg处理PB级数据湖
- **云原生**: AWS、阿里云、腾讯云提供托管Iceberg服务
- **活跃生态**: 与Spark、Flink、Trino、Hive等大数据引擎集成
- **面向未来**: 在现代数据架构中替代传统Hive表

**技术优势:**
- **ACID事务**: 数据湖上的完整ACID保证
- **时间旅行**: 查询历史快照用于审计和回滚
- **Schema演进**: 添加、删除、重命名列无需重写数据
- **隐藏分区**: 自动分区管理，用户友好
- **快照隔离**: 并发读写无冲突
- **增量处理**: 高效的变更数据捕获(CDC)

**战略价值:**
Iceberg不仅是表格式，更是现代数据架构的基础：
- **数据湖标准**: 行业正在向Iceberg标准收敛
- **云迁移**: 云原生数据湖部署的必备组件
- **查询性能**: 元数据修剪显著提高查询速度
- **存储效率**: 分区演进和压缩降低存储成本

### What you expected to happen

**English:**

Linkis should provide an Apache Iceberg engine plugin with the following capabilities:

1. **Iceberg Table Operations:**
- CREATE/DROP Iceberg tables
- ALTER TABLE for schema evolution
- INSERT INTO for batch writes
- MERGE INTO for upsert operations
- DELETE and UPDATE statements

2. **Catalog Support:**
- Hive Metastore catalog
- Hadoop catalog
- REST catalog
- AWS Glue catalog
- Custom catalog implementations

3. **Time Travel Queries:**
- Query by snapshot ID: `SELECT ... FOR SYSTEM_TIME AS OF snapshot_id`
- Query by timestamp: `SELECT ... FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00'`
- List table snapshots and history
- Rollback to previous snapshots

4. **Advanced Features:**
- Partition evolution (change partitioning scheme without rewriting data)
- Hidden partitioning (automatic partition values)
- Metadata tables (snapshots, files, manifests, partitions)
- Table maintenance (expire snapshots, rewrite data files, compact manifests)

5. **Integration with Existing Engines:**
- Enhance Spark engine for Iceberg operations
- Enhance Flink engine for Iceberg streaming writes
- Enhance Trino/Presto engine for Iceberg queries
- Unified Iceberg table management across engines

6. **Integration with Linkis:**
- Unified task submission interface
- Resource management integration
- Permission control integration
- Metadata catalog integration

---

**中文:**

Linkis应该提供Apache Iceberg引擎插件，具备以下能力：

1. **Iceberg表操作:**
- CREATE/DROP Iceberg表
- ALTER TABLE用于schema演进
- INSERT INTO用于批量写入
- MERGE INTO用于upsert操作
- DELETE和UPDATE语句

2. **Catalog支持:**
- Hive Metastore catalog
- Hadoop catalog
- REST catalog
- AWS Glue catalog
- 自定义catalog实现

3. **时间旅行查询:**
- 按快照ID查询: `SELECT ... FOR SYSTEM_TIME AS OF snapshot_id`
- 按时间戳查询: `SELECT ... FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00'`
- 列出表快照和历史
- 回滚到先前快照

4. **高级功能:**
- 分区演进（更改分区方案无需重写数据）
- 隐藏分区（自动分区值）
- 元数据表（快照、文件、清单、分区）
- 表维护（过期快照、重写数据文件、压缩清单）

5. **与现有引擎集成:**
- 增强Spark引擎支持Iceberg操作
- 增强Flink引擎支持Iceberg流式写入
- 增强Trino/Presto引擎支持Iceberg查询
- 跨引擎统一Iceberg表管理

6. **与Linkis集成:**
- 统一的任务提交接口
- 资源管理集成
- 权限控制集成
- 元数据目录集成

### How to reproduce

**English:**

Current situation:
1. Users need to manually configure Spark/Flink with Iceberg libraries
2. No dedicated Iceberg table management interface in Linkis
3. Cannot leverage Linkis's unified task submission for Iceberg operations
4. Limited support for Iceberg-specific features (time travel, schema evolution)

Use case example:
```sql
-- Create Iceberg table with partitioning
CREATE TABLE iceberg_catalog.db.events (
event_id BIGINT,
user_id BIGINT,
event_type STRING,
event_time TIMESTAMP,
event_data STRING
) USING iceberg
PARTITIONED BY (days(event_time))
TBLPROPERTIES (
'write.format.default' = 'parquet',
'write.parquet.compression-codec' = 'zstd'
);

-- Insert data
INSERT INTO iceberg_catalog.db.events VALUES
(1, 1001, 'click', TIMESTAMP '2024-12-20 10:00:00', '{"page": "home"}'),
(2, 1002, 'view', TIMESTAMP '2024-12-20 11:00:00', '{"product_id": 123}');

-- Time travel query (not supported without plugin)
SELECT * FROM iceberg_catalog.db.events
FOR SYSTEM_TIME AS OF '2024-12-20 10:30:00';

-- Schema evolution
ALTER TABLE iceberg_catalog.db.events
ADD COLUMN session_id STRING;

-- Partition evolution (change from daily to hourly partitioning)
ALTER TABLE iceberg_catalog.db.events
DROP PARTITION FIELD days(event_time);

ALTER TABLE iceberg_catalog.db.events
ADD PARTITION FIELD hours(event_time);

-- Query metadata tables
SELECT * FROM iceberg_catalog.db.events.snapshots;
SELECT * FROM iceberg_catalog.db.events.files;
```

**Performance Requirements:**
- Support PB-scale Iceberg tables
- Time travel queries with snapshot metadata pruning
- Efficient schema and partition evolution
- Concurrent read/write without blocking

---

**中文:**

当前情况：
1. 用户需要手动配置Spark/Flink与Iceberg库
2. Linkis中没有专门的Iceberg表管理接口
3. 无法利用Linkis的统一任务提交进行Iceberg操作
4. 对Iceberg特定功能（时间旅行、schema演进）的支持有限

使用场景示例：
```sql
-- 创建带分区的Iceberg表
CREATE TABLE iceberg_catalog.db.events (
event_id BIGINT,
user_id BIGINT,
event_type STRING,
event_time TIMESTAMP,
event_data STRING
) USING iceberg
PARTITIONED BY (days(event_time))
TBLPROPERTIES (
'write.format.default' = 'parquet',
'write.parquet.compression-codec' = 'zstd'
);

-- 插入数据
INSERT INTO iceberg_catalog.db.events VALUES
(1, 1001, 'click', TIMESTAMP '2024-12-20 10:00:00', '{"page": "home"}'),
(2, 1002, 'view', TIMESTAMP '2024-12-20 11:00:00', '{"product_id": 123}');

-- 时间旅行查询（没有插件不支持）
SELECT * FROM iceberg_catalog.db.events
FOR SYSTEM_TIME AS OF '2024-12-20 10:30:00';

-- Schema演进
ALTER TABLE iceberg_catalog.db.events
ADD COLUMN session_id STRING;

-- 分区演进（从按天分区改为按小时分区）
ALTER TABLE iceberg_catalog.db.events
DROP PARTITION FIELD days(event_time);

ALTER TABLE iceberg_catalog.db.events
ADD PARTITION FIELD hours(event_time);

-- 查询元数据表
SELECT * FROM iceberg_catalog.db.events.snapshots;
SELECT * FROM iceberg_catalog.db.events.files;
```

**性能要求:**
- 支持PB级Iceberg表
- 时间旅行查询带快照元数据修剪
- 高效的schema和分区演进
- 并发读写无阻塞

### Anything else

**English:**

**📚 Implementation Guide:**

Before starting development, please refer to the official Linkis engine plugin development guide:
- **English Guide**: https://linkis.apache.org/docs/latest/development/new-engine-conn
- **Chinese Guide**: https://linkis.apache.org/zh-CN/docs/latest/development/new-engine-conn

This guide provides:
- Complete engine plugin architecture overview
- Step-by-step development instructions
- Code structure and naming conventions
- Testing and debugging best practices
- Deployment and configuration guide

**Implementation Approach:**

**Recommended Strategy: Enhance Existing Engines (Not Standalone Plugin)**

Based on the Linkis architecture and Iceberg's design philosophy, the recommended approach is **Option B**: Enhance existing Spark/Flink/Trino engines with Iceberg support, rather than creating a standalone Iceberg engine plugin.

**Rationale:**
1. **Iceberg is a table format, not a compute engine**: Iceberg provides table abstraction, but actual computation happens in Spark/Flink/Trino
2. **Leverage existing optimizations**: Spark/Flink already have mature Iceberg integration with optimized query planning
3. **Avoid duplication**: No need to reimplement query engine, just enhance table management
4. **Better performance**: Use native engine optimizations (e.g., Spark's Iceberg pushdowns)

**Phase 1: Enhance Spark Engine for Iceberg**

Maven dependencies (add to Spark engine plugin):
```xml

org.apache.iceberg
iceberg-spark-runtime-3.3_2.12
1.4.3

```

Configuration enhancement:
```properties
# Iceberg catalog configuration
spark.sql.catalog.iceberg_catalog=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.type=hive
spark.sql.catalog.iceberg_catalog.uri=thrift://metastore:9083
spark.sql.catalog.iceberg_catalog.warehouse=hdfs://namenode:8020/warehouse/iceberg

# Default catalog for Iceberg tables
spark.sql.defaultCatalog=iceberg_catalog

# Iceberg extensions
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
```

**Phase 2: Iceberg Management Module**

Create a dedicated Iceberg management module:
```
linkis-iceberg-manager/
├── src/main/scala/
│ ├── org/apache/linkis/iceberg/
│ │ ├── catalog/CatalogManager.scala
│ │ ├── snapshot/SnapshotManager.scala
│ │ ├── metadata/MetadataExplorer.scala
│ │ └── maintenance/TableMaintenance.scala
├── src/main/resources/
│ └── iceberg-manager.properties
└── pom.xml
```

**Key Components:**

1. **Catalog Manager:**
```scala
class IcebergCatalogManager {
def createCatalog(catalogName: String, catalogType: String, properties: Map[String, String]): Unit = {
catalogType match {
case "hive" =>
registerHiveCatalog(catalogName, properties)
case "hadoop" =>
registerHadoopCatalog(catalogName, properties)
case "rest" =>
registerRESTCatalog(catalogName, properties)
}
}

def listCatalogs(): Seq[String] = {
// List all registered Iceberg catalogs
}

def getCatalogProperties(catalogName: String): Map[String, String] = {
// Get catalog configuration
}
}
```

2. **Snapshot Manager:**
```scala
class IcebergSnapshotManager {
def listSnapshots(tableName: String): Seq[Snapshot] = {
val table = loadTable(tableName)
table.snapshots().asScala.toSeq
}

def rollbackToSnapshot(tableName: String, snapshotId: Long): Unit = {
val table = loadTable(tableName)
table.manageSnapshots()
.rollbackTo(snapshotId)
.commit()
}

def expireSnapshots(tableName: String, olderThan: Timestamp): Unit = {
val table = loadTable(tableName)
table.expireSnapshots()
.expireOlderThan(olderThan.getTime)
.commit()
}
}
```

3. **Metadata Explorer:**
```scala
class IcebergMetadataExplorer {
def exploreSnapshots(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.snapshots")
}

def exploreFiles(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.files")
}

def exploreManifests(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.manifests")
}
}
```

**Phase 3: Linkis Console Integration**

Add Iceberg table management UI to Linkis Console:
- Catalog browser
- Table schema viewer
- Snapshot timeline viewer
- Time travel query builder
- Schema evolution wizard
- Partition evolution tool
- Table maintenance dashboard

**Configuration:**
```properties
# Linkis Iceberg configuration
wds.linkis.iceberg.enabled=true
wds.linkis.iceberg.default.catalog=iceberg_catalog
wds.linkis.iceberg.default.catalog.type=hive
wds.linkis.iceberg.default.catalog.uri=thrift://metastore:9083
wds.linkis.iceberg.default.warehouse=hdfs://namenode:8020/warehouse/iceberg

# Snapshot retention
wds.linkis.iceberg.snapshot.retention.days=7
wds.linkis.iceberg.snapshot.min.retain=2

# File management
wds.linkis.iceberg.file.rewrite.threshold=512MB
wds.linkis.iceberg.manifest.compact.threshold=100
```

**Phase 4: Advanced Features**

1. **Incremental Processing:**
```scala
// Read incremental changes from Iceberg table
val incrementalDF = spark.read
.format("iceberg")
.option("start-snapshot-id", startSnapshotId)
.option("end-snapshot-id", endSnapshotId)
.table("iceberg_catalog.db.events")
```

2. **Branch and Tag Support:**
```scala
// Create branch for experimentation
ALTER TABLE iceberg_catalog.db.events
CREATE BRANCH experiment_branch;

// Query specific branch
SELECT * FROM iceberg_catalog.db.events.branch_experiment_branch;

// Create tag for release
ALTER TABLE iceberg_catalog.db.events
CREATE TAG release_v1.0;
```

**Testing Requirements:**
- Unit tests for catalog, snapshot, and metadata managers
- Integration tests with Hive Metastore and Hadoop filesystem
- Integration tests with Spark, Flink, Trino engines
- Time travel query correctness tests
- Schema evolution compatibility tests
- Partition evolution data integrity tests
- Concurrent read/write stress tests
- PB-scale table performance benchmarks

**Documentation Needs:**
- Iceberg integration architecture guide
- Catalog configuration guide for different types (Hive, Hadoop, REST, Glue)
- Time travel query user guide
- Schema and partition evolution best practices
- Table maintenance and optimization guide
- Migration guide from Hive tables to Iceberg tables
- Troubleshooting guide for common issues

**Estimated Effort:**
- Spark engine Iceberg enhancement: 0.5 person-month
- Flink engine Iceberg enhancement: 0.5 person-month
- Iceberg management module: 1 person-month
- Console UI integration: 0.5 person-month
- Testing and documentation: 0.5 person-month
- Total: 3 person-months (using enhancement approach)

**Expected ROI:**
- Cloud-native capability demonstration
- Enterprise-grade data lake support
- Seamless integration with cloud vendor Iceberg services
- ROI: 2.5-4.0x

---

**中文:**

**📚 实现指引:**

开始开发前，请参考Linkis官方引擎插件开发指南：
- **中文指南**: https://linkis.apache.org/zh-CN/docs/latest/development/new-engine-conn
- **英文指南**: https://linkis.apache.org/docs/latest/development/new-engine-conn

该指南提供：
- 完整的引擎插件架构概述
- 分步开发说明
- 代码结构和命名规范
- 测试和调试最佳实践
- 部署和配置指南

**实施方法:**

**推荐策略：增强现有引擎（非独立插件）**

基于Linkis架构和Iceberg的设计哲学，推荐采用**方案B**：增强现有Spark/Flink/Trino引擎支持Iceberg，而不是创建独立的Iceberg引擎插件。

**理由:**
1. **Iceberg是表格式，不是计算引擎**: Iceberg提供表抽象，实际计算发生在Spark/Flink/Trino中
2. **利用现有优化**: Spark/Flink已有成熟的Iceberg集成和优化的查询规划
3. **避免重复**: 无需重新实现查询引擎，只需增强表管理
4. **更好的性能**: 使用原生引擎优化（如Spark的Iceberg下推）

**阶段1：增强Spark引擎支持Iceberg**

Maven依赖（添加到Spark引擎插件）：
```xml

org.apache.iceberg
iceberg-spark-runtime-3.3_2.12
1.4.3

```

配置增强：
```properties
# Iceberg catalog配置
spark.sql.catalog.iceberg_catalog=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.type=hive
spark.sql.catalog.iceberg_catalog.uri=thrift://metastore:9083
spark.sql.catalog.iceberg_catalog.warehouse=hdfs://namenode:8020/warehouse/iceberg

# Iceberg表的默认catalog
spark.sql.defaultCatalog=iceberg_catalog

# Iceberg扩展
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
```

**阶段2：Iceberg管理模块**

创建专门的Iceberg管理模块：
```
linkis-iceberg-manager/
├── src/main/scala/
│ ├── org/apache/linkis/iceberg/
│ │ ├── catalog/CatalogManager.scala
│ │ ├── snapshot/SnapshotManager.scala
│ │ ├── metadata/MetadataExplorer.scala
│ │ └── maintenance/TableMaintenance.scala
├── src/main/resources/
│ └── iceberg-manager.properties
└── pom.xml
```

**关键组件:**

1. **Catalog管理器:**
```scala
class IcebergCatalogManager {
def createCatalog(catalogName: String, catalogType: String, properties: Map[String, String]): Unit = {
catalogType match {
case "hive" =>
registerHiveCatalog(catalogName, properties)
case "hadoop" =>
registerHadoopCatalog(catalogName, properties)
case "rest" =>
registerRESTCatalog(catalogName, properties)
}
}

def listCatalogs(): Seq[String] = {
// 列出所有注册的Iceberg catalogs
}

def getCatalogProperties(catalogName: String): Map[String, String] = {
// 获取catalog配置
}
}
```

2. **快照管理器:**
```scala
class IcebergSnapshotManager {
def listSnapshots(tableName: String): Seq[Snapshot] = {
val table = loadTable(tableName)
table.snapshots().asScala.toSeq
}

def rollbackToSnapshot(tableName: String, snapshotId: Long): Unit = {
val table = loadTable(tableName)
table.manageSnapshots()
.rollbackTo(snapshotId)
.commit()
}

def expireSnapshots(tableName: String, olderThan: Timestamp): Unit = {
val table = loadTable(tableName)
table.expireSnapshots()
.expireOlderThan(olderThan.getTime)
.commit()
}
}
```

3. **元数据浏览器:**
```scala
class IcebergMetadataExplorer {
def exploreSnapshots(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.snapshots")
}

def exploreFiles(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.files")
}

def exploreManifests(tableName: String): DataFrame = {
spark.read.format("iceberg")
.load(s"$tableName.manifests")
}
}
```

**阶段3：Linkis控制台集成**

在Linkis控制台添加Iceberg表管理UI：
- Catalog浏览器
- 表schema查看器
- 快照时间线查看器
- 时间旅行查询构建器
- Schema演进向导
- 分区演进工具
- 表维护仪表板

**配置:**
```properties
# Linkis Iceberg配置
wds.linkis.iceberg.enabled=true
wds.linkis.iceberg.default.catalog=iceberg_catalog
wds.linkis.iceberg.default.catalog.type=hive
wds.linkis.iceberg.default.catalog.uri=thrift://metastore:9083
wds.linkis.iceberg.default.warehouse=hdfs://namenode:8020/warehouse/iceberg

# 快照保留
wds.linkis.iceberg.snapshot.retention.days=7
wds.linkis.iceberg.snapshot.min.retain=2

# 文件管理
wds.linkis.iceberg.file.rewrite.threshold=512MB
wds.linkis.iceberg.manifest.compact.threshold=100
```

**阶段4：高级功能**

1. **增量处理:**
```scala
// 从Iceberg表读取增量变化
val incrementalDF = spark.read
.format("iceberg")
.option("start-snapshot-id", startSnapshotId)
.option("end-snapshot-id", endSnapshotId)
.table("iceberg_catalog.db.events")
```

2. **分支和标签支持:**
```scala
// 创建实验分支
ALTER TABLE iceberg_catalog.db.events
CREATE BRANCH experiment_branch;

// 查询特定分支
SELECT * FROM iceberg_catalog.db.events.branch_experiment_branch;

// 为发布创建标签
ALTER TABLE iceberg_catalog.db.events
CREATE TAG release_v1.0;
```

**测试要求:**
- Catalog、快照和元数据管理器的单元测试
- 与Hive Metastore和Hadoop文件系统的集成测试
- 与Spark、Flink、Trino引擎的集成测试
- 时间旅行查询正确性测试
- Schema演进兼容性测试
- 分区演进数据完整性测试
- 并发读写压力测试
- PB级表性能基准测试

**文档需求:**
- Iceberg集成架构指南
- 不同类型Catalog配置指南（Hive、Hadoop、REST、Glue）
- 时间旅行查询用户指南
- Schema和分区演进最佳实践
- 表维护和优化指南
- 从Hive表迁移到Iceberg表的迁移指南
- 常见问题故障排除指南

**预计工作量:**
- Spark引擎Iceberg增强：0.5人月
- Flink引擎Iceberg增强：0.5人月
- Iceberg管理模块：1人月
- 控制台UI集成：0.5人月
- 测试和文档：0.5人月
- 总计：3人月（使用增强方法）

**预期ROI:**
- 云原生能力展示
- 企业级数据湖支持
- 与云厂商Iceberg服务无缝对接
- ROI: 2.5-4.0倍

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!