In case of null links check for end-of-stream (databricks#1138)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->
NO_CHANGELOG=true

In case of no/null links, check for end-of-stream. This will prevent
driver making fetch calls to the server for closed statements.

## Testing
<!-- Describe how the changes have been tested-->
- manual testing
- unit tests

## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

Author: jayantsing-db

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

@@ -116,7 +116,8 @@ private static ChunkProvider createRemoteChunkProvider(
 
       // Convert ExternalLinks to ChunkLinkFetchResult for the provider
       ChunkLinkFetchResult initialLinks =
-          convertToChunkLinkFetchResult(resultData.getExternalLinks());
+          convertToChunkLinkFetchResult(
+              resultData.getExternalLinks(), resultManifest.getTotalChunkCount());
 
       return new StreamingChunkProvider(
           linkFetcher,
@@ -377,11 +378,17 @@ private void setColumnInfo(TGetResultSetMetadataResp resultManifest) {
    * Converts a collection of ExternalLinks to a ChunkLinkFetchResult.
    *
    * @param externalLinks The external links to convert, may be null
-   * @return A ChunkLinkFetchResult, or null if input is null or empty
+   * @param totalChunkCount The total chunk count from result manifest, may be null
+   * @return A ChunkLinkFetchResult, or endOfStream if chunk count is zero, or null if unknown
    */
   private static ChunkLinkFetchResult convertToChunkLinkFetchResult(
-      Collection<ExternalLink> externalLinks) {
+      Collection<ExternalLink> externalLinks, Long totalChunkCount) {
     if (externalLinks == null || externalLinks.isEmpty()) {
+      // If total chunk count is zero, return end of stream
+      if (totalChunkCount != null && totalChunkCount == 0) {
+        LOGGER.debug("Total chunk count is zero, returning end of stream");
+        return ChunkLinkFetchResult.endOfStream();
+      }
       return null;
     }
 
@@ -396,6 +403,13 @@ private static ChunkLinkFetchResult convertToChunkLinkFetchResult(
     long nextFetchIndex = hasMore ? lastLink.getNextChunkIndex() : -1;
     long nextRowOffset = lastLink.getRowOffset() + lastLink.getRowCount();
 
+    LOGGER.debug(
+        "Converting ExternalLinks to ChunkLinkFetchResult: linkCount={}, hasMore={}, nextFetchIndex={}, nextRowOffset={}",
+        linkList.size(),
+        hasMore,
+        nextFetchIndex,
+        nextRowOffset);
+
     return ChunkLinkFetchResult.of(linkList, hasMore, nextFetchIndex, nextRowOffset);
   }
 
@@ -406,12 +420,17 @@ private static ChunkLinkFetchResult convertToChunkLinkFetchResult(
    * ChunkLinkFetchResult format used by StreamingChunkProvider.
    *
    * @param resultsResp The Thrift fetch results response containing initial links
-   * @return A ChunkLinkFetchResult, or null if no links
+   * @return A ChunkLinkFetchResult, or endOfStream if no more rows, or null if unknown
    */
   private static ChunkLinkFetchResult convertThriftLinksToChunkLinkFetchResult(
       TFetchResultsResp resultsResp) {
     List<TSparkArrowResultLink> resultLinks = resultsResp.getResults().getResultLinks();
     if (resultLinks == null || resultLinks.isEmpty()) {
+      // If hasMoreRows is false, return end of stream
+      if (!resultsResp.hasMoreRows) {
+        LOGGER.debug("No result links and hasMoreRows is false, returning end of stream");
+        return ChunkLinkFetchResult.endOfStream();
+      }
       return null;
     }
 

src/test/java/com/databricks/jdbc/api/impl/arrow/ArrowStreamResultTest.java

@@ -609,4 +609,70 @@ private Object[][] createTestData(Schema schema, int rows) {
     }
     return data;
   }
+
+  // ==================== End of Stream Conversion Tests ====================
+
+  @Test
+  public void testSeaEmptyLinksWithZeroChunkCountReturnsEndOfStream() throws Exception {
+    // Enable StreamingChunkProvider
+    Properties props = new Properties();
+    props.setProperty("EnableStreamingChunkProvider", "1");
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContextFactory.create(JDBC_URL, props);
+
+    assertTrue(connectionContext.isStreamingChunkProviderEnabled());
+
+    DatabricksSession localSession = new DatabricksSession(connectionContext, mockedSdkClient);
+
+    // Create result manifest with zero chunks and empty external links
+    ResultManifest resultManifest =
+        new ResultManifest()
+            .setTotalChunkCount(0L)
+            .setTotalRowCount(0L)
+            .setResultCompression(CompressionCodec.NONE)
+            .setSchema(new ResultSchema().setColumns(new ArrayList<>()).setColumnCount(0L));
+
+    ResultData localResultData = new ResultData().setExternalLinks(new ArrayList<>());
+
+    // Should create result successfully with end-of-stream signal
+    ArrowStreamResult result =
+        new ArrowStreamResult(
+            resultManifest, localResultData, STATEMENT_ID, localSession, mockHttpClient);
+
+    assertNotNull(result);
+    assertFalse(result.hasNext(), "Empty result should have no data");
+    assertDoesNotThrow(result::close);
+  }
+
+  @Test
+  public void testSeaNullLinksWithZeroChunkCountReturnsEndOfStream() throws Exception {
+    // Enable StreamingChunkProvider
+    Properties props = new Properties();
+    props.setProperty("EnableStreamingChunkProvider", "1");
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContextFactory.create(JDBC_URL, props);
+
+    assertTrue(connectionContext.isStreamingChunkProviderEnabled());
+
+    DatabricksSession localSession = new DatabricksSession(connectionContext, mockedSdkClient);
+
+    // Create result manifest with zero chunks and null external links
+    ResultManifest resultManifest =
+        new ResultManifest()
+            .setTotalChunkCount(0L)
+            .setTotalRowCount(0L)
+            .setResultCompression(CompressionCodec.NONE)
+            .setSchema(new ResultSchema().setColumns(new ArrayList<>()).setColumnCount(0L));
+
+    ResultData localResultData = new ResultData().setExternalLinks(null);
+
+    // Should create result successfully with end-of-stream signal
+    ArrowStreamResult result =
+        new ArrowStreamResult(
+            resultManifest, localResultData, STATEMENT_ID, localSession, mockHttpClient);
+
+    assertNotNull(result);
+    assertFalse(result.hasNext(), "Empty result should have no data");
+    assertDoesNotThrow(result::close);
+  }
 }