Add a test

kssenii · Oct 15, 2023 · f3056b1 · f3056b1
1 parent a7f389a
commit f3056b1
Show file tree

Hide file tree

Showing 3 changed files with 61 additions and 2 deletions.
diff --git a/src/Storages/DataLakes/IcebergMetadataParser.cpp b/src/Storages/DataLakes/IcebergMetadataParser.cpp
@@ -32,6 +32,8 @@ namespace ErrorCodes
 template <typename Configuration, typename MetadataReadHelper>
 struct IcebergMetadataParser<Configuration, MetadataReadHelper>::Impl
 {
+    Poco::Logger * log = &Poco::Logger::get("IcebergMetadataParser");
+
     /**
      * Useful links:
      * - https://iceberg.apache.org/spec/
@@ -299,7 +301,10 @@ struct IcebergMetadataParser<Configuration, MetadataReadHelper>::Impl
                     throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected to find {} in data path: {}", configuration.url.key, data_path);
 
                 if (status == 2)
-                    keys.erase(file_path);
+                {
+                    LOG_TEST(log, "Got delete file for {}", file_path);
+                    chassert(!keys.contains(file_path));
+                }
                 else
                     keys.insert(file_path);
             }

diff --git a/tests/integration/helpers/cluster.py b/tests/integration/helpers/cluster.py
@@ -32,7 +32,6 @@
     import nats
     import ssl
     import meilisearch
-    import pyspark
     from confluent_kafka.avro.cached_schema_registry_client import (
         CachedSchemaRegistryClient,
     )
@@ -631,6 +630,7 @@ def __init__(
             logging.debug(f"Removed :{self.instances_dir}")
 
         if with_spark:
+            import pyspark
             # if you change packages, don't forget to update them in docker/test/integration/runner/dockerd-entrypoint.sh
             (
                 pyspark.sql.SparkSession.builder.appName("spark_test")

diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py
@@ -313,3 +313,57 @@ def test_types(started_cluster, format_version):
             ["e", "Nullable(Bool)"],
         ]
     )
+
+
+@pytest.mark.parametrize("format_version", ["1", "2"])
+def test_delete_files(started_cluster, format_version):
+    instance = started_cluster.instances["node1"]
+    spark = started_cluster.spark_session
+    minio_client = started_cluster.minio_client
+    bucket = started_cluster.minio_bucket
+    TABLE_NAME = "test_delete_files_" + format_version
+
+    write_iceberg_from_df(
+        spark,
+        generate_data(spark, 0, 100),
+        TABLE_NAME,
+        mode="overwrite",
+        format_version=format_version,
+    )
+
+    files = upload_directory(
+        minio_client, bucket, f"/iceberg_data/default/{TABLE_NAME}/", ""
+    )
+
+    create_iceberg_table(instance, TABLE_NAME)
+
+    assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
+
+    spark.sql(f"DELETE FROM {TABLE_NAME} WHERE a >= 0")
+    files = upload_directory(
+        minio_client, bucket, f"/iceberg_data/default/{TABLE_NAME}/", ""
+    )
+
+    assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 0
+    assert instance.contains_in_log("Got delete file for")
+
+    write_iceberg_from_df(
+        spark,
+        generate_data(spark, 100, 200),
+        TABLE_NAME,
+        mode="upsert",
+        format_version=format_version,
+    )
+
+    files = upload_directory(
+        minio_client, bucket, f"/iceberg_data/default/{TABLE_NAME}/", ""
+    )
+
+    assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
+
+    spark.sql(f"DELETE FROM {TABLE_NAME} WHERE a >= 150")
+    files = upload_directory(
+        minio_client, bucket, f"/iceberg_data/default/{TABLE_NAME}/", ""
+    )
+
+    assert int(instance.query(f"SELECT count() FROM {TABLE_NAME}")) == 50