feat: Implement streaming prefetch for Thrift inline results

Implements proactive prefetching with sliding window for both Thrift columnar
and inline Arrow results, eliminating blocking at batch boundaries.

Key components:
- ThriftStreamingProvider<T>: Generic streaming provider with type-safe batches
- StreamingBatch<T>: Type-safe batch container with lifecycle management
- ThriftResponseProcessor<T>: Pluggable processors for Columnar and Arrow
- StreamingColumnarResult: Streaming variant for Thrift columnar results
- StreamingInlineArrowResult: Streaming variant for inline Arrow results

Configuration:
- EnableInlineStreaming: Toggle streaming (default: enabled)
- ThriftMaxBatchesInMemory: Sliding window size (default: 3)

Author: vikrantpuppala

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

@@ -1198,6 +1198,26 @@ public boolean isStreamingChunkProviderEnabled() {
     return getParameter(DatabricksJdbcUrlParams.ENABLE_STREAMING_CHUNK_PROVIDER).equals("1");
   }
 
+  @Override
+  public boolean isInlineStreamingEnabled() {
+    return getParameter(DatabricksJdbcUrlParams.ENABLE_INLINE_STREAMING).equals("1");
+  }
+
+  @Override
+  public boolean isCloudFetchEnabled() {
+    return getParameter(DatabricksJdbcUrlParams.ENABLE_CLOUD_FETCH).equals("1");
+  }
+
+  @Override
+  public int getThriftMaxBatchesInMemory() {
+    try {
+      return Integer.parseInt(getParameter(DatabricksJdbcUrlParams.THRIFT_MAX_BATCHES_IN_MEMORY));
+    } catch (NumberFormatException e) {
+      LOGGER.warn("Invalid value for ThriftMaxBatchesInMemory, using default value 3");
+      return 3;
+    }
+  }
+
   @Override
   public int getLinkPrefetchWindow() {
     return Integer.parseInt(getParameter(DatabricksJdbcUrlParams.LINK_PREFETCH_WINDOW));

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

@@ -2,7 +2,9 @@
 
 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.volume.VolumeOperationResult;
+import com.databricks.jdbc.api.internal.IDatabricksConnectionContext;
 import com.databricks.jdbc.api.internal.IDatabricksSession;
 import com.databricks.jdbc.api.internal.IDatabricksStatementInternal;
 import com.databricks.jdbc.common.util.DatabricksThriftUtil;
@@ -95,9 +97,9 @@ private static IExecutionResult getResultHandler(
     LOGGER.info("Processing result of format {} from Thrift server", resultFormat);
     switch (resultFormat) {
       case COLUMN_BASED_SET:
-        return new LazyThriftResult(resultsResp, parentStatement, session);
+        return createThriftColumnarResult(resultsResp, parentStatement, session);
       case ARROW_BASED_SET:
-        return new LazyThriftInlineArrowResult(resultsResp, parentStatement, session);
+        return createInlineArrowResult(resultsResp, parentStatement, session);
       case URL_BASED_SET:
         return new ArrowStreamResult(resultsResp, parentStatement, session);
       case ROW_BASED_SET:
@@ -109,6 +111,54 @@ private static IExecutionResult getResultHandler(
     }
   }
 
+  /**
+   * Creates the appropriate result handler for Thrift columnar results. Uses
+   * StreamingColumnarResult by default for improved throughput, otherwise falls back to
+   * LazyThriftResult if disabled.
+   */
+  private static IExecutionResult createThriftColumnarResult(
+      TFetchResultsResp resultsResp,
+      IDatabricksStatementInternal parentStatement,
+      IDatabricksSession session)
+      throws DatabricksSQLException {
+    IDatabricksConnectionContext connectionContext = session.getConnectionContext();
+
+    // Streaming is enabled by default (ENABLE_INLINE_STREAMING defaults to "1")
+    if (connectionContext.isInlineStreamingEnabled()) {
+      LOGGER.info("Using StreamingColumnarResult for improved throughput (default)");
+      int maxBatchesInMemory = connectionContext.getThriftMaxBatchesInMemory();
+      return new StreamingColumnarResult(
+          resultsResp, parentStatement, session, maxBatchesInMemory, 300); // 5 minute timeout
+    } else {
+      LOGGER.info("Using LazyThriftResult (streaming explicitly disabled)");
+      return new LazyThriftResult(resultsResp, parentStatement, session);
+    }
+  }
+
+  /**
+   * Creates the appropriate result handler for inline Arrow results. Uses
+   * StreamingInlineArrowResult by default for improved throughput, otherwise falls back to
+   * LazyThriftInlineArrowResult.
+   */
+  private static IExecutionResult createInlineArrowResult(
+      TFetchResultsResp resultsResp,
+      IDatabricksStatementInternal parentStatement,
+      IDatabricksSession session)
+      throws DatabricksSQLException {
+    IDatabricksConnectionContext connectionContext = session.getConnectionContext();
+
+    // Streaming is enabled by default (ENABLE_INLINE_STREAMING defaults to "1")
+    if (connectionContext.isInlineStreamingEnabled()) {
+      LOGGER.info("Using StreamingInlineArrowResult for improved throughput (default)");
+      int maxBatchesInMemory = connectionContext.getThriftMaxBatchesInMemory();
+      return new StreamingInlineArrowResult(
+          resultsResp, parentStatement, session, maxBatchesInMemory, 300); // 5 minute timeout
+    } else {
+      LOGGER.info("Using LazyThriftInlineArrowResult (streaming explicitly disabled)");
+      return new LazyThriftInlineArrowResult(resultsResp, parentStatement, session);
+    }
+  }
+
   static IExecutionResult getResultSet(Object[][] rows) {
     return new InlineJsonResult(rows);
   }

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

@@ -388,7 +388,6 @@ protected static Object getObjectWithComplexTypeHandling(
 
     if (!isComplexDatatypeSupportEnabled && isComplexType(requiredType)) {
       LOGGER.debug("Complex datatype support is disabled, converting complex type to STRING");
-
       Object result =
           chunkIterator.getColumnObjectAtCurrentRow(
               columnIndex, ColumnInfoTypeName.STRING, "STRING", columnInfo);