Task: Add Unit Test for Flink-Spark Equality Delete Write

[adelly13]

Created new flink-spark-bundle module with Unit Test testEqualityDeleteWritesOnSpark() in TestFlinkSpark.java.

Procedures for testEqualityDeleteWritesOnSpark():

1. Create table in Flink
2. Write initial data
3. Create a deletestream
4. Apply equality deletes
5. Initialize spark session
6. Read table using spark
7. Compare

Current issues: running into errors with the imports as well as Invalid write distribution mode: range. Need to define sort order or partition spec.

[geruh]

We can get rid of the all of the tests but your spark + flink test

[geruh]

This is a good test for writing the equality delete using the stream execution environment! I'd also suggest creating a test for an UPSERT case by leveraging the FlinkTableEnviroment. This will help you see how, the UPSERT leverages an equality delete to replace the value.

You can add a test to TestFlinkCatalogTable.java like this:

    sql("CREATE TABLE test_table (id INT, data STRING, PRIMARY KEY(id) NOT ENFORCED) WITH ('format-version'='2', 'write.upsert.enabled'='true')");

    sql("INSERT INTO test_table VALUES (1, 'a'), (2, 'b'), (3, 'c')");

    // Perform upsert operation
    sql("INSERT INTO test_table VALUES (2, 'updated_b'), (4, 'd')");

[geruh]

To expand upon this, I'd suggest adding some DeleteFile assertions. For example, what do you expect the DeleteFile to look like we want to delete based on:

• one column: id = 1
• all columns: id = 1 and data = 'a'
• range: id > 3 (also how does flink get the values greater than 3?)

[geruh]

We should set the same warehouse as the Flink catalog to ensure we are using the same tables.

spark = SparkSession.builder()
        .appName("iceberg-spark")
        .master("local[*]")
        .config("spark.sql.catalog.hadoop_catalog", "org.apache.iceberg.spark.SparkCatalog")
        .config("spark.sql.catalog.hadoop_catalog.type", "hadoop")
        .config("spark.sql.catalog.hadoop_catalog.warehouse", CATALOG_EXTENSION.warehouse())
        .getOrCreate();