log time to create table

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

Author: varun-edachali-dbx

src/databricks/sql/backend/sea/queue.py

@@ -2,6 +2,7 @@
 
 from abc import ABC
 import threading
+import time
 from typing import Dict, List, Optional, Tuple, Union
 
 from databricks.sql.cloudfetch.download_manager import ResultFileDownloadManager
@@ -152,16 +153,17 @@ def __init__(
 
         self._shutdown_event = threading.Event()
 
-        self._condition = threading.Condition()
+        self._link_present = threading.Condition()
         self._error = None
         self.chunk_index_to_link: Dict[int, "ExternalLink"] = {}
+
         for link in initial_links:
             self.chunk_index_to_link[link.chunk_index] = link
             self.download_manager.add_link(self._convert_to_thrift_link(link))
         self.total_chunk_count = total_chunk_count
 
     def _get_next_chunk_index(self) -> Optional[int]:
-        with self._condition:
+        with self._link_present:
             max_chunk_index = max(self.chunk_index_to_link.keys(), default=None)
             if max_chunk_index is None:
                 return 0
@@ -175,20 +177,20 @@ def _trigger_next_batch_download(self) -> bool:
 
         try:
             links = self.backend.get_chunk_links(self._statement_id, next_chunk_index)
-            with self._condition:
+            with self._link_present:
                 self.chunk_index_to_link.update(
                     {link.chunk_index: link for link in links}
                 )
-                self._condition.notify_all()
+                self._link_present.notify_all()
             for link in links:
                 self.download_manager.add_link(self._convert_to_thrift_link(link))
         except Exception as e:
             logger.error(
                 f"LinkFetcher: Error fetching links for chunk {next_chunk_index}: {e}"
             )
-            with self._condition:
+            with self._link_present:
                 self._error = e
-                self._condition.notify_all()
+                self._link_present.notify_all()
             return False
 
         return True
@@ -197,14 +199,11 @@ def get_chunk_link(self, chunk_index: int) -> Optional["ExternalLink"]:
         if chunk_index >= self.total_chunk_count:
             return None
 
-        with self._condition:
-            if self._error:
-                raise self._error
-
+        with self._link_present:
             while chunk_index not in self.chunk_index_to_link:
                 if self._error:
                     raise self._error
-                self._condition.wait()
+                self._link_present.wait()
 
             return self.chunk_index_to_link.get(chunk_index, None)
 
@@ -308,6 +307,7 @@ def __init__(
 
     def _create_next_table(self) -> Union["pyarrow.Table", None]:
         """Create next table by retrieving the logical next downloaded file."""
+        start_time = time.time()
         if not self.download_manager:
             logger.debug("SeaCloudFetchQueue: No download manager, returning")
             return None
@@ -319,6 +319,11 @@ def _create_next_table(self) -> Union["pyarrow.Table", None]:
         row_offset = chunk_link.row_offset
         arrow_table = self._create_table_at_offset(row_offset)
 
+        end_time = time.time()
+        logger.info(
+            f"SeaCloudFetchQueue: Created table at offset {row_offset} in {end_time - start_time} seconds"
+        )
+
         self.current_chunk_index += 1
 
         return arrow_table