Use lightweight /status endpoint for SEA heartbeat instead of full GetStatement

Changed checkStatementAlive from GET /sql/statements/{id} (~21KB response
with full manifest/result data) to GET /sql/statements/{id}/status (~100
bytes with just state/error/sqlState). Reduces heartbeat bandwidth by
~99% per tick.

Co-authored-by: Isaac
Signed-off-by: Gopal Lal <gopal.lal@databricks.com>

Author: gopalldb

src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java

@@ -37,6 +37,7 @@
 import com.databricks.jdbc.model.core.ExternalLink;
 import com.databricks.jdbc.model.core.ResultData;
 import com.databricks.jdbc.model.core.ResultManifest;
+import com.databricks.jdbc.model.core.StatementStatus;
 import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import com.databricks.sdk.WorkspaceClient;
 import com.databricks.sdk.core.ApiClient;
@@ -417,13 +418,13 @@ public DatabricksResultSet executeStatementAsync(
   @Override
   public boolean checkStatementAlive(StatementId typedStatementId) throws SQLException {
     String statementId = typedStatementId.toSQLExecStatementId();
-    String getStatusPath = String.format(STATEMENT_PATH_WITH_ID, statementId);
+    // Use lightweight /status endpoint (~100 bytes) instead of full GetStatement (~21KB)
+    String statusPath = String.format(STATEMENT_STATUS_PATH_WITH_ID, statementId);
     try {
-      GetStatementRequest request = new GetStatementRequest().setStatementId(statementId);
-      Request req = new Request(Request.GET, getStatusPath, apiClient.serialize(request));
-      req.withHeaders(getHeaders("getStatement"));
-      GetStatementResponse response = apiClient.execute(req, GetStatementResponse.class);
-      StatementState state = response.getStatus().getState();
+      Request req = new Request(Request.GET, statusPath, (String) null);
+      req.withHeaders(getHeaders("getStatementStatus"));
+      StatementStatus status = apiClient.execute(req, StatementStatus.class);
+      StatementState state = status.getState();
       // Terminal states mean the operation is no longer alive
       return state != StatementState.CANCELED
           && state != StatementState.CLOSED

src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/PathConstants.java

@@ -8,6 +8,7 @@ public class PathConstants {
   public static final String STATEMENT_PATH = BASE_PATH + "statements/";
   public static final String STATEMENT_PATH_WITH_ID = STATEMENT_PATH + "%s";
   public static final String CANCEL_STATEMENT_PATH_WITH_ID = STATEMENT_PATH + "%s/cancel";
+  public static final String STATEMENT_STATUS_PATH_WITH_ID = STATEMENT_PATH + "%s/status";
   public static final String RESULT_CHUNK_PATH = STATEMENT_PATH_WITH_ID + "/result/chunks/%s";
   public static final String TELEMETRY_PATH = "/telemetry-ext";
   public static final String TELEMETRY_PATH_UNAUTHENTICATED = "/telemetry-unauth";

src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java

@@ -1324,12 +1324,8 @@ public void testCheckStatementAlive_succeededState_returnsTrue() throws Exceptio
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
     StatementStatus status = new StatementStatus().setState(StatementState.SUCCEEDED);
-    GetStatementResponse response = new GetStatementResponse();
-    response.setStatus(status);
-    response.setStatementId(STATEMENT_ID.toSQLExecStatementId());
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
-        .thenReturn(response);
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class))).thenReturn(status);
 
     assertTrue(databricksSdkClient.checkStatementAlive(STATEMENT_ID));
   }
@@ -1342,11 +1338,8 @@ public void testCheckStatementAlive_runningState_returnsTrue() throws Exception
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
     StatementStatus status = new StatementStatus().setState(StatementState.RUNNING);
-    GetStatementResponse response = new GetStatementResponse();
-    response.setStatus(status);
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
-        .thenReturn(response);
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class))).thenReturn(status);
 
     assertTrue(databricksSdkClient.checkStatementAlive(STATEMENT_ID));
   }
@@ -1359,11 +1352,8 @@ public void testCheckStatementAlive_canceledState_returnsFalse() throws Exceptio
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
     StatementStatus status = new StatementStatus().setState(StatementState.CANCELED);
-    GetStatementResponse response = new GetStatementResponse();
-    response.setStatus(status);
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
-        .thenReturn(response);
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class))).thenReturn(status);
 
     assertFalse(databricksSdkClient.checkStatementAlive(STATEMENT_ID));
   }
@@ -1376,11 +1366,8 @@ public void testCheckStatementAlive_closedState_returnsFalse() throws Exception
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
     StatementStatus status = new StatementStatus().setState(StatementState.CLOSED);
-    GetStatementResponse response = new GetStatementResponse();
-    response.setStatus(status);
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
-        .thenReturn(response);
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class))).thenReturn(status);
 
     assertFalse(databricksSdkClient.checkStatementAlive(STATEMENT_ID));
   }
@@ -1393,11 +1380,8 @@ public void testCheckStatementAlive_failedState_returnsFalse() throws Exception
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
     StatementStatus status = new StatementStatus().setState(StatementState.FAILED);
-    GetStatementResponse response = new GetStatementResponse();
-    response.setStatus(status);
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
-        .thenReturn(response);
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class))).thenReturn(status);
 
     assertFalse(databricksSdkClient.checkStatementAlive(STATEMENT_ID));
   }
@@ -1409,7 +1393,7 @@ public void testCheckStatementAlive_exceptionWrapped() throws Exception {
     DatabricksSdkClient databricksSdkClient =
         new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
 
-    when(apiClient.execute(any(Request.class), eq(GetStatementResponse.class)))
+    when(apiClient.execute(any(Request.class), eq(StatementStatus.class)))
         .thenThrow(new RuntimeException("Network error"));
 
     DatabricksSQLException exception =