[wip] define method to merge append deltas into metadata db file

Author: jonavellecuerdo

metadata.duckdb

@@ -262,3 +262,24 @@ def test_tdm_current_records_most_recent_version(timdex_metadata_with_deltas):
             == most_recent.iloc[0]["run_timestamp"]
         )
         assert current_version.iloc[0]["run_id"] == most_recent.iloc[0]["run_id"]
+
+
+def test_tdm_compact_append_deltas(timdex_metadata_with_deltas):
+    # get record count from static db
+    metadata_db_current_count = timdex_metadata_with_deltas.conn.query(
+        """select count(*) as count from static_db.records;"""
+    ).fetchone()[0]
+    total_count = timdex_metadata_with_deltas.records_count
+
+    # compact append deltas into static db file
+    timdex_metadata_with_deltas.compact_append_deltas()
+    timdex_metadata_with_deltas.refresh()
+
+    # get updated record count from static db
+    metadata_db_updated_count = timdex_metadata_with_deltas.conn.query(
+        """select count(*) as count from static_db.records;"""
+    ).fetchone()[0]
+
+    # verify the addition of new records
+    assert metadata_db_current_count < metadata_db_updated_count
+    assert metadata_db_updated_count == total_count

tests/test_metadata.py

@@ -42,6 +42,22 @@ def test_split_s3_uri_invalid():
         client._split_s3_uri("timdex/path/to/file.txt")
 
 
+def test_list_objects(s3_bucket_mocked, tmp_path):
+    client = S3Client()
+
+    # Create a test file
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("test content")
+
+    # Upload the file
+    s3_uri = "s3://timdex/metadata/append_deltas/test.txt"
+    client.upload_file(test_file, s3_uri)
+
+    # Verify list of objects
+    s3_prefix = "s3://timdex/metadata/append_deltas"
+    assert client.list_objects(s3_prefix) == ["metadata/append_deltas/test.txt"]
+
+
 def test_upload_download_file(s3_bucket_mocked, tmp_path):
     """Test upload_file and download_file methods."""
     client = S3Client()

tests/test_s3client.py

@@ -442,6 +442,77 @@ def _create_current_records_view(self, conn: DuckDBPyConnection) -> None:
         """
         conn.execute(query)
 
+    def merge_append_deltas(self) -> None:
+        """Merge append deltas into the static metadata database file."""
+        start_time = time.perf_counter()
+
+        s3_client = S3Client()
+
+        # get filenames of append deltas
+        append_delta_filenames = (
+            self.conn.query(
+                """
+                select distinct(filename)
+                from metadata.append_deltas
+                """
+            )
+            .to_df()["filename"]
+            .to_list()
+        )
+
+        if len(append_delta_filenames) == 0:
+            logger.info("No append deltas found.")
+            return
+
+        logger.info(
+            f"Found {len(append_delta_filenames)} append delta file(s) in {self.append_deltas_path}: {append_delta_filenames}"
+        )
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # create local copy of the static metadata database (static db) file
+            local_db_path = str(Path(temp_dir) / self.metadata_database_filename)
+            if self.location_scheme == "s3":
+                s3_client.download_file(
+                    s3_uri=self.metadata_database_path, local_path=local_db_path
+                )
+            else:
+                shutil.copy(self.metadata_database_path, local_db_path)
+
+            # attach to local static db
+            self.conn.execute(f"""attach '{local_db_path}' AS local_static_db;""")
+
+            # insert records from append deltas to local static db
+            self.conn.execute(
+                """
+                insert into local_static_db.records
+                select * 
+                from metadata.append_deltas
+                """
+            )
+
+            # detach from local static db
+            self.conn.execute("""detach local_static_db;""")
+
+            # overwrite static db file with local version
+            if self.location_scheme == "s3":
+                s3_client.upload_file(
+                    local_path=local_db_path,
+                    s3_uri=self.metadata_database_path,
+                )
+            else:
+                shutil.copy(src=local_db_path, dst=self.metadata_database_path)
+
+        # delete append deltas
+        for append_delta in append_delta_filenames:
+            if self.location_scheme == "s3":
+                s3_client.delete_file(s3_uri=append_delta)
+            else:
+                shutil.copy(self.metadata_database_path, local_db_path)
+
+        logger.info(
+            f"Merged append deltas into the metadata database file: {self.metadata_database_path}, {time.perf_counter()-start_time}s"
+        )
+
     def write_append_delta_duckdb(self, filepath: str) -> None:
         """Write an append delta for an ETL parquet file.
 

timdex_dataset_api/metadata.py

@@ -58,6 +58,12 @@ def object_exists(self, s3_uri: str) -> bool:
                 return False
             raise
 
+    def list_objects(self, s3_prefix: str) -> list[str]:
+        bucket, _ = self._split_s3_uri(s3_prefix)
+        objects = [obj.key for obj in self.resource.Bucket(bucket).objects.all()]
+        logger.debug(f"Found {len(objects)} objects in {s3_prefix}: {objects}")
+        return objects
+
     def download_file(self, s3_uri: str, local_path: str | pathlib.Path) -> None:
         bucket, key = self._split_s3_uri(s3_uri)
         local_path = pathlib.Path(local_path)