Feat: Batch ingest iceberg/file source

[chenzl25]

Is your feature request related to a problem? Please describe.

According to RFC: Combine Historical and Incremental Data
we need to support ingesting data from an external source (e.g. iceberg or file source) as historical data. This is a typical scenario of bulk loading which is expected to be faster than streaming loading data from a Kafka.

To support this feature, we need to

• support create iceberg source. feat(frontend): support create iceberg source #14971
• support iceberg catalog reader (via JNI to utilize the mature ecosystem from the Java SDK) and file enumerator.
• support iceberg data file (parquet, ORC) batch read. Recommended to use icelake. feat(batch): support iceberg scan executor #14915
• support iceberg batch plan and schedule task. feat(batch): support batch read iceberg source #15214
• feat(source): support file source batch read #14866
• data type conversion between external souce and risingwave in memory chunk columns. feat(batch): support iceberg scan executor #14915
• distributed insert select. feat(batch): distributed dml #14630
• disable overwrite on conflict check for table when it has no downstream. feat(stream): disable conflict check for stand alone table #14655 (comment)
• auto enable distributed dml for Iceberg ingestion. feat(batch): auto decide distributed dml for iceberg ingestion #15481
• automatically derive columns for iceberg source. feat(frontend): derive columns from iceberg source automatically #15415
• Support parquet format for file source #15509

Performance Improvement:

• Icelake provide file level read interface, so we can use it in the iceberg scan executor. cc @ZENOTME
• Column pruning for iceberg scan. feat(batch,optimizer): support column pruning for iceberg source #16429
• Utilize iceberg zonemap statistics to perform data skipping .aka. predicate push down.
• Accelerate scan file planning and data accessing. Use alluxio as a cache.

Others:

• Give a type compatibility table between Iceberg and RisingWave data type. feat(batch): support decimal type for iceberg type #15298
• integration test test(batch): add iceberg source integration test #15491 test(batch): support hive catalog for iceberg source #15550 feat(batch): support jdbc catalog for iceberg source #15551 feat(batch): support rest catalog for iceberg source #15535
• Support more catalog type for iceberg via jni.
• Support query iceberg table metadata feat(batch,meta): support iceberg snapshots sys table #16175 feat(meta,batch): support iceberg files system table #16180
• Read a specific snapshot from iceberg source. feat(batch): support time travel for iceberg source #15866
• Support iceberg merge-on-read. https://www.notion.so/risingwave-labs/Iceberg-Support-read-from-MoR-table-4ca08d580ff74ab6a0c3296f059bf158?pvs=4

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[lmatz]

Just try to provide the complete context, the details of the POC user request can be found: https://www.notion.so/risingwave-labs/optimize-parquet-source-for-batch-load-dc498a043d504621bf56461690b14bd7?d=84ebdf5d7469412680278059c5898be8

In short, if implementing the batch iceberg source takes much time due to its complexity, a parquet file source with decent performance is good enough to help move the POC forward. The user will consider switching RW only if RW's iceberg batch source is fast enough.

[chenzl25]

In short, if implementing the batch iceberg source takes much time due to its complexity, a parquet file source with decent performance is good enough to help move the POC forward. The user will consider switching RW only if RW's iceberg batch source is fast enough.

Must the file format be Parquet? Is it possible to use a CSV file that has been supported in our file source? If it is ok to test a CSV file first, we can support file source batch read first, to test the performance. BTW, I tested insert select from a RisingWave table to another RisingWave table last week. Can we just compare the streaming load from Kafka to a table with insert select from a table to another table?

[lmatz]

Is it possible to use a CSV file that has been supported in our file source? If it is ok to test a CSV file first,

Good point, I think it is ok as CSV is a less efficient format than Parquet in terms of read and write performance.
If we achieve decent enough performance for CSV files, we can be even faster when using Parquet. I guess it is a very convincing argument to the POC user.

BTW, I tested insert select from a RisingWave table to another RisingWave table last week. Can we just compare the streaming load from Kafka to a table with insert select from a table to another table?

I can try to communicate this first. The closer to the user's real use case, the better, but definitely nothing wrong if we use what we have at the moment, could you post the link to the last week's results? 🙏
I am also thinking of adding this to the daily performance tests

[chenzl25]

I can try to communicate this first. The closer to the user's real use case, the better, but definitely nothing wrong if we use what we have at the moment, could you post the link to the last week's results? 🙏 I am also thinking of adding this to the daily performance tests

#14630 (comment)

[chenzl25]

#15784

[chenzl25]

FInished