re-introduce schema_bytes for better abstraction (likely temporary)

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

Author: varun-edachali-dbx

src/databricks/sql/cloud_fetch_queue.py

@@ -135,21 +135,23 @@ def __init__(
         self,
         max_download_threads: int,
         ssl_options: SSLOptions,
+        schema_bytes: bytes,
         lz4_compressed: bool = True,
         description: Optional[List[Tuple[Any, ...]]] = None,
     ):
         """
         Initialize the base CloudFetchQueue.
 
         Args:
-            schema_bytes: Arrow schema bytes
             max_download_threads: Maximum number of download threads
             ssl_options: SSL options for downloads
+            schema_bytes: Arrow schema bytes
             lz4_compressed: Whether the data is LZ4 compressed
             description: Column descriptions
         """
         self.lz4_compressed = lz4_compressed
         self.description = description
+        self.schema_bytes = schema_bytes
         self._ssl_options = ssl_options
         self.max_download_threads = max_download_threads
 
@@ -191,7 +193,6 @@ def next_n_rows(self, num_rows: int) -> "pyarrow.Table":
         """Get up to the next n rows of the cloud fetch Arrow dataframes."""
         if not self.table:
             # Return empty pyarrow table to cause retry of fetch
-            logger.info("SeaCloudFetchQueue: No table available, returning empty table")
             return self._create_empty_table()
 
         logger.info("SeaCloudFetchQueue: Retrieving up to {} rows".format(num_rows))
@@ -309,6 +310,7 @@ def __init__(
         super().__init__(
             max_download_threads=max_download_threads,
             ssl_options=ssl_options,
+            schema_bytes=b"",
             lz4_compressed=lz4_compressed,
             description=description,
         )
@@ -435,11 +437,11 @@ def __init__(
         super().__init__(
             max_download_threads=max_download_threads,
             ssl_options=ssl_options,
+            schema_bytes=schema_bytes,
             lz4_compressed=lz4_compressed,
             description=description,
         )
 
-        self.schema_bytes = schema_bytes
         self.start_row_index = start_row_offset
         self.result_links = result_links or []
 

src/databricks/sql/result_set.py

@@ -272,6 +272,18 @@ def _fill_results_buffer(self):
         self.results = results
         self.has_more_rows = has_more_rows
 
+    def _convert_columnar_table(self, table):
+        column_names = [c[0] for c in self.description]
+        ResultRow = Row(*column_names)
+        result = []
+        for row_index in range(table.num_rows):
+            curr_row = []
+            for col_index in range(table.num_columns):
+                curr_row.append(table.get_item(col_index, row_index))
+            result.append(ResultRow(*curr_row))
+
+        return result
+
     def merge_columnar(self, result1, result2) -> "ColumnTable":
         """
         Function to merge / combining the columnar results into a single result