rptest/datalake: test metadata interoperability with 3rd party system

tests/rptest/services/spark_service.py

@@ -149,6 +149,25 @@ def escape_identifier(self, table: str) -> str:
     def engine_name():
         return QueryEngineType.SPARK
 
+    def run_sample_maintenance_task(self, namespace, table) -> None:
+        # Metadata query
+        # https://iceberg.apache.org/docs/1.6.1/spark-queries/#files
+        initial_parquet_files = self.run_query_fetch_one(
+            f"SELECT count(*) FROM {namespace}.{table}.files")[0]
+
+        # Want at least 2 files to be able to assert that optimization did something.
+        assert initial_parquet_files >= 2, f"Expecting at least 2 files, got {initial_parquet_files}"
+
+        # Spark Procedures provided by Iceberg SQL Extensions
+        # https://iceberg.apache.org/docs/1.6.1/spark-procedures/#rewrite_data_files
+        self.run_query_fetch_one(
+            f"CALL `redpanda-iceberg-catalog`.system.rewrite_data_files(\"{namespace}.{table}\")"
+        )
+
+        optimized_parquet_files = self.run_query_fetch_one(
+            f"SELECT count(*) FROM {namespace}.{table}.files")[0]
+        assert optimized_parquet_files < initial_parquet_files, f"Expecting fewer files after optimize, got {optimized_parquet_files}"
+
     def make_client(self):
         assert self.spark_host
         return hive.connect(host=self.spark_host, port=self.spark_port)

tests/rptest/services/trino_service.py

@@ -139,6 +139,22 @@ def make_client(self):
     def escape_identifier(self, table: str) -> str:
         return f'"{table}"'
 
+    def run_sample_maintenance_task(self, namespace, table) -> None:
+        # See Trino metadata tables documentation
+        # https://trino.io/docs/current/connector/iceberg.html#files-table
+        initial_parquet_files = self.count_table(namespace, f"{table}$files")
+
+        # Want at least 2 files to be able to assert that optimization did something.
+        assert initial_parquet_files >= 2, f"Expecting at least 2 files, got {initial_parquet_files}"
+
+        # Optimize the table to rewrite the data.
+        # https://trino.io/docs/current/connector/iceberg.html#alter-table-execute
+        self.run_query_fetch_one(
+            f"ALTER TABLE {namespace}.{table} EXECUTE optimize")
+
+        optimized_parquet_files = self.count_table(namespace, f"{table}$files")
+        assert optimized_parquet_files < initial_parquet_files, f"Expecting fewer files after optimize, got {optimized_parquet_files}"
+
     @staticmethod
     def dict_to_conf(d: dict[str, Optional[str | bool]]):
         """

tests/rptest/tests/datalake/3rdparty_maintenance_test.py

@@ -0,0 +1,86 @@
+# Copyright 2024 Vectorized, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+from ducktape.mark import matrix
+
+from rptest.services.cluster import cluster
+from rptest.services.redpanda import SISettings
+from rptest.tests.datalake.datalake_services import DatalakeServices
+from rptest.tests.datalake.datalake_verifier import DatalakeVerifier
+from rptest.tests.datalake.query_engine_base import QueryEngineType
+from rptest.tests.datalake.utils import supported_storage_types
+from rptest.tests.redpanda_test import RedpandaTest
+
+
+class Datalake3rdPartyMaintenanceTest(RedpandaTest):
+    def __init__(self, test_ctx, *args, **kwargs):
+        super().__init__(test_ctx,
+                         num_brokers=1,
+                         si_settings=SISettings(test_ctx),
+                         extra_rp_conf={
+                             "iceberg_enabled": "true",
+                             "iceberg_catalog_commit_interval_ms": 5000
+                         },
+                         *args,
+                         **kwargs)
+
+        self.test_ctx = test_ctx
+        self.topic_name = "test"
+        self.num_partitions = 10
+
+        self.produced_messages = 0
+
+    def setUp(self):
+        # redpanda will be started by DatalakeServices
+        pass
+
+    @cluster(num_nodes=4)
+    @matrix(cloud_storage_type=supported_storage_types(),
+            query_engine=[QueryEngineType.SPARK, QueryEngineType.TRINO],
+            filesystem_catalog_mode=[True, False])
+    def test_e2e_basic(self, cloud_storage_type, query_engine,
+                       filesystem_catalog_mode):
+        """
+        This test verifies that Redpanda can continue to work with Iceberg
+        metadata written by third-party query engines. We use an optimize operation
+        with a third-party query engine to trigger a rewrite of the data files
+        and metadata.
+        """
+        with DatalakeServices(self.test_ctx,
+                              redpanda=self.redpanda,
+                              filesystem_catalog_mode=filesystem_catalog_mode,
+                              include_query_engines=[query_engine]) as dl:
+            dl.create_iceberg_enabled_topic(self.topic_name,
+                                            partitions=self.num_partitions)
+            # Write some data to the topic.
+            self._translate_sample_data(dl)
+
+            # Run maintenance to rewrite the data.
+            dl.query_engine(query_engine).run_sample_maintenance_task(
+                "redpanda", self.topic_name)
+
+            # Verify consistency post rewrite.
+            DatalakeVerifier.oneshot(self.redpanda, self.topic_name,
+                                     dl.query_engine(query_engine))
+
+            # Produce additional messages to the topic to make sure we correctly
+            # interoperate with the metadata written by Trino.
+            self._translate_sample_data(dl)
+
+            # Verify consistency with the additional messages.
+            DatalakeVerifier.oneshot(self.redpanda, self.topic_name,
+                                     dl.query_engine(query_engine))
+
+    def _translate_sample_data(self, dl):
+        NUM_MSG_PER_SAMPLE = 100
+        self.produced_messages += NUM_MSG_PER_SAMPLE
+
+        dl.produce_to_topic(self.topic_name, 1024, NUM_MSG_PER_SAMPLE)
+        # Wait for all messages (including the ones we just wrote) to be translated.
+        dl.wait_for_translation(self.topic_name,
+                                msg_count=self.produced_messages)

tests/rptest/tests/datalake/datalake_services.py

@@ -16,7 +16,7 @@
 from rptest.services.redpanda import RedpandaService
 from rptest.services.spark_service import SparkService
 from rptest.services.trino_service import TrinoService
-from rptest.tests.datalake.query_engine_base import QueryEngineType
+from rptest.tests.datalake.query_engine_base import QueryEngineBase, QueryEngineType
 from rptest.services.redpanda_connect import RedpandaConnectService
 from rptest.tests.datalake.query_engine_factory import get_query_engine_by_type
 
@@ -85,6 +85,13 @@ def __enter__(self):
     def __exit__(self, *args, **kwargs):
         self.tearDown()
 
+    def query_engine(self, type: QueryEngineType) -> QueryEngineBase:
+        for e in self.query_engines:
+            assert isinstance(e, QueryEngineBase)
+            if e.engine_name() == type:
+                return e
+        raise Exception(f"Query engine {type} not found")
+
     def trino(self) -> TrinoService:
         trino = self.service(QueryEngineType.TRINO)
         assert trino, "Missing Trino service"

tests/rptest/tests/datalake/datalake_verifier.py

@@ -287,3 +287,12 @@ def stop(self):
         self.logger.debug(f"queried offsets: {self._max_queried_offsets}")
 
         assert self._max_queried_offsets == self._max_consumed_offsets, "Mismatch between maximum offsets in topic vs iceberg table"
+
+    @staticmethod
+    def oneshot(redpanda: RedpandaService,
+                topic: str,
+                query_engine: QueryEngineBase,
+                progress_timeout_sec=30):
+        verifier = DatalakeVerifier(redpanda, topic, query_engine)
+        verifier.start()
+        verifier.wait(progress_timeout_sec=progress_timeout_sec)

tests/rptest/tests/datalake/query_engine_base.py

@@ -33,12 +33,9 @@ def run_query(self, query):
         client = self.make_client()
         assert client
         try:
-            try:
-                cursor = client.cursor()
-                cursor.execute(query)
-                yield cursor
-            finally:
-                cursor.close()
+            cursor = client.cursor()
+            cursor.execute(query)
+            yield cursor
         finally:
             client.close()
 
@@ -50,6 +47,10 @@ def run_query_fetch_all(self, query):
         with self.run_query(query) as cursor:
             return cursor.fetchall()
 
+    def run_query_fetch_one(self, query):
+        with self.run_query(query) as cursor:
+            return cursor.fetchone()
+
     def count_table(self, namespace, table) -> int:
         query = f"select count(*) from {namespace}.{self.escape_identifier(table)}"
         with self.run_query(query) as cursor:
@@ -59,3 +60,16 @@ def max_translated_offset(self, namespace, table, partition) -> int:
         query = f"select max(redpanda.offset) from {namespace}.{self.escape_identifier(table)} where redpanda.partition={partition}"
         with self.run_query(query) as cursor:
             return cursor.fetchone()[0]
+
+    def run_sample_maintenance_task(self, namespace, table) -> None:
+        """
+        Subclasses should implement this method to run a maintenance task on the
+        given table.
+
+        Useful to test that redpanda can still work with the table after the
+        maintenance task is run.
+
+        The method must guarantee that a maintenance task was run on the table
+        after it returns for the tests to be valid.
+        """
+        raise NotImplementedError