clean up code

Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>

Author: vikrantpuppala

src/main/java/com/databricks/jdbc/api/impl/ExecutionResultFactory.java

@@ -3,6 +3,7 @@
 import com.databricks.jdbc.api.impl.arrow.ArrowStreamResult;
 import com.databricks.jdbc.api.impl.arrow.LazyThriftInlineArrowResult;
 import com.databricks.jdbc.api.impl.arrow.StreamingInlineArrowResult;
+import com.databricks.jdbc.api.impl.thrift.StreamingColumnarResult;
 import com.databricks.jdbc.api.impl.volume.VolumeOperationResult;
 import com.databricks.jdbc.api.internal.IDatabricksConnectionContext;
 import com.databricks.jdbc.api.internal.IDatabricksSession;

src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingInlineArrowResult.java

@@ -134,7 +134,7 @@ public StreamingInlineArrowResult(
       }
     }
 
-    LOGGER.info(
+    LOGGER.debug(
         "[STREAMING] StreamingInlineArrowResult initialized - firstBatchRows={}, maxRows={}, maxBatchesInMemory={}",
         currentBatch != null ? currentBatch.getRowCount() : 0,
         maxRows,
@@ -236,7 +236,7 @@ public boolean next() throws DatabricksSQLException {
         currentChunkIterator = chunk.getChunkIterator();
         currentChunkIterator.nextRow();
 
-        LOGGER.info(
+        LOGGER.debug(
             "[CONSUMER] Moved to batch {} - globalRow={}, batchesInMemory={}",
             currentBatch.getBatchIndex(),
             globalRowIndex,
@@ -295,7 +295,7 @@ public void close() {
       provider.close();
     }
 
-    LOGGER.info(
+    LOGGER.debug(
         "[STREAMING] Closed - totalRowsFetched={}, rowsConsumed={}", totalRows, globalRowIndex + 1);
   }
 

src/main/java/com/databricks/jdbc/api/impl/streaming/ThriftStreamingProvider.java

@@ -179,7 +179,7 @@ private ThriftStreamingProvider(
     this.maxBatchesInMemory = Math.max(2, maxBatchesInMemory);
     this.batchReadyTimeoutSeconds = timeoutSeconds;
 
-    LOGGER.info(
+    LOGGER.debug(
         "Creating ThriftStreamingProvider: maxBatches={}, timeout={}s, processor={}",
         this.maxBatchesInMemory,
         timeoutSeconds,
@@ -193,14 +193,14 @@ private ThriftStreamingProvider(
     totalRowsFetched.addAndGet(initialBatch.getRowCount());
     nextRowOffset.set(initialBatch.getRowCount());
 
-    LOGGER.info(
+    LOGGER.debug(
         "Initial batch processed: rows={}, hasMoreRows={}",
         initialBatch.getRowCount(),
         initialBatch.hasMoreRows());
 
     if (!initialBatch.hasMoreRows()) {
       endOfStreamReached = true;
-      LOGGER.info("Single batch result - all data in initial response");
+      LOGGER.debug("Single batch result - all data in initial response");
     }
 
     // Start prefetch thread
@@ -337,14 +337,21 @@ public boolean isEndOfStreamReached() {
   public void close() {
     if (closed) return;
 
-    LOGGER.info("Closing ThriftStreamingProvider, total rows: {}", totalRowsFetched.get());
+    LOGGER.debug("Closing ThriftStreamingProvider, total rows: {}", totalRowsFetched.get());
     closed = true;
 
     notifyConsumerAdvanced();
     notifyBatchAvailable();
 
+    // Interrupt and wait for prefetch thread to terminate before releasing resources
     if (prefetchThread != null) {
       prefetchThread.interrupt();
+      try {
+        prefetchThread.join(5000); // Wait up to 5s for clean shutdown
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        LOGGER.debug("Interrupted while waiting for prefetch thread to terminate");
+      }
     }
 
     // Release all batches using type-safe release action
@@ -431,7 +438,7 @@ private void fetchNextBatchInternal() throws DatabricksSQLException {
 
     if (!batch.hasMoreRows()) {
       endOfStreamReached = true;
-      LOGGER.info("End of stream at batch {}", batchIndex);
+      LOGGER.debug("End of stream at batch {}", batchIndex);
     }
 
     notifyBatchAvailable();

src/main/java/com/databricks/jdbc/api/impl/thrift/StreamingColumnarResult.java

@@ -114,7 +114,7 @@ public StreamingColumnarResult(
       currentBatch = provider.getCurrentBatch();
     }
 
-    LOGGER.info(
+    LOGGER.debug(
         "[STREAMING] StreamingColumnarResult initialized - firstBatchRows={}, maxRows={}, maxBatchesInMemory={}",
         currentBatch != null ? currentBatch.getRowCount() : 0,
         maxRows,
@@ -204,7 +204,7 @@ public boolean next() throws DatabricksSQLException {
         currentBatchRowIndex = 0;
 
         // Log batch transition
-        LOGGER.info(
+        LOGGER.debug(
             "[CONSUMER] Moved to batch {} - globalRow={}, batchesInMemory={}",
             currentBatch.getBatchIndex(),
             globalRowIndex,
@@ -220,7 +220,7 @@ public boolean next() throws DatabricksSQLException {
 
     // Log progress periodically (every 500K rows)
     if (globalRowIndex > 0 && globalRowIndex % 500000 == 0) {
-      LOGGER.info(
+      LOGGER.debug(
           "[CONSUMER] Progress - rows={}, batch={}, batchesInMemory={}",
           globalRowIndex,
           currentBatch.getBatchIndex(),
@@ -274,7 +274,7 @@ public void close() {
       provider.close();
     }
 
-    LOGGER.info(
+    LOGGER.debug(
         "[STREAMING] Closed - totalRowsFetched={}, rowsConsumed={}", totalRows, globalRowIndex + 1);
   }
 

src/main/java/com/databricks/jdbc/api/impl/thrift/ThriftBatch.java

@@ -47,7 +47,9 @@ public TFetchResultsResp fetchNextBatch() throws DatabricksSQLException {
           "ThriftBatchFetcher is closed", DatabricksDriverErrorCode.STATEMENT_CLOSED);
     }
 
-    LOGGER.debug("Fetching next batch for statement {}", statement.getStatementId());
+    LOGGER.debug(
+        "Fetching next batch for statement {}",
+        statement != null ? statement.getStatementId() : "null");
     long startTime = System.currentTimeMillis();
 
     TFetchResultsResp response = session.getDatabricksClient().getMoreResults(statement);