diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md
index 29b843a24..8350a797e 100644
--- a/NEXT_CHANGELOG.md
+++ b/NEXT_CHANGELOG.md
@@ -17,6 +17,7 @@
 - Arrow schema deserialization failures (Thrift metadata path) now surface a dedicated driver error code `ARROW_SCHEMA_PARSING_ERROR` (vendor code `22000`) and a proper SQLSTATE `22000` (Data Exception) on the thrown `SQLException`, instead of the generic `RESULT_SET_ERROR` (1004) and the enum name as SQLSTATE. The exception message is unchanged.
 
 ### Fixed
+- Fixed `MetadataOperationTimeout` not being applied when metadata operations use SHOW commands. Operations like `getTables`, `getSchemas`, and `getColumns` now respect the `MetadataOperationTimeout` connection property instead of hanging indefinitely with no timeout.
 
 - Reclassify transient/mis-categorized server errors so callers can identify
   retryable failures. The remap is applied at all Thrift error sites
diff --git a/src/main/java/com/databricks/jdbc/common/DatabricksJdbcConstants.java b/src/main/java/com/databricks/jdbc/common/DatabricksJdbcConstants.java
index 11c31c1c0..857d67c69 100644
--- a/src/main/java/com/databricks/jdbc/common/DatabricksJdbcConstants.java
+++ b/src/main/java/com/databricks/jdbc/common/DatabricksJdbcConstants.java
@@ -121,6 +121,7 @@ public final class DatabricksJdbcConstants {
    * concurrent-modification errors where the operation is potentially retryable.
    */
   public static final String SERIALIZATION_FAILURE_SQLSTATE = "40001";
+
   /** Standard SQL state for data exception (SQLSTATE 22000). */
   public static final String DATA_EXCEPTION_SQLSTATE = "22000";
 
diff --git a/src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java b/src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java
index 168b1a0a5..ada68f693 100644
--- a/src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java
+++ b/src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java
@@ -248,16 +248,23 @@ public DatabricksResultSet executeStatement(
       parentStatement.setStatementId(typedStatementId);
     }
 
-    int timeoutInSeconds =
-        parentStatement != null ? parentStatement.getStatement().getQueryTimeout() : 0;
+    int timeoutInSeconds;
+    if (parentStatement != null) {
+      timeoutInSeconds = parentStatement.getStatement().getQueryTimeout();
+    } else if (statementType == StatementType.METADATA) {
+      timeoutInSeconds = connectionContext.getMetadataOperationTimeout();
+    } else {
+      timeoutInSeconds = 0;
+    }
 
-    // Create timeout handler
+    // Create timeout handler — use OPERATION_TIMEOUT_ERROR for metadata to match
+    // the native Thrift metadata path (DatabricksThriftAccessor.fetchMetadataResults)
+    DatabricksDriverErrorCode timeoutErrorCode =
+        (statementType == StatementType.METADATA && parentStatement == null)
+            ? DatabricksDriverErrorCode.OPERATION_TIMEOUT_ERROR
+            : DatabricksDriverErrorCode.STATEMENT_EXECUTION_TIMEOUT;
     TimeoutHandler timeoutHandler =
-        TimeoutHandler.forStatement(
-            timeoutInSeconds,
-            typedStatementId,
-            this,
-            DatabricksDriverErrorCode.STATEMENT_EXECUTION_TIMEOUT);
+        TimeoutHandler.forStatement(timeoutInSeconds, typedStatementId, this, timeoutErrorCode);
 
     StatementState responseState = response.getStatus().getState();
     while (responseState == StatementState.PENDING || responseState == StatementState.RUNNING) {
diff --git a/src/main/java/com/databricks/jdbc/dbclient/impl/thrift/DatabricksThriftAccessor.java b/src/main/java/com/databricks/jdbc/dbclient/impl/thrift/DatabricksThriftAccessor.java
index 2ff8dbfca..a8fc8623e 100644
--- a/src/main/java/com/databricks/jdbc/dbclient/impl/thrift/DatabricksThriftAccessor.java
+++ b/src/main/java/com/databricks/jdbc/dbclient/impl/thrift/DatabricksThriftAccessor.java
@@ -237,7 +237,7 @@ DatabricksResultSet execute(
 
       TGetOperationStatusResp statusResp =
           pollTillOperationFinished(
-              response, parentStatement, session, statementId, sessionDebugInfo);
+              response, parentStatement, session, statementId, sessionDebugInfo, statementType);
       boolean isDirectResults = hasResultDataInDirectResults(response);
       if (isDirectResults) {
         // The first response has result data
@@ -303,10 +303,17 @@ private TGetOperationStatusResp pollTillOperationFinished(
       IDatabricksStatementInternal parentStatement,
       IDatabricksSession session,
       StatementId statementId,
-      String sessionDebugInfo)
+      String sessionDebugInfo,
+      StatementType statementType)
       throws SQLException, TException {
-    int timeoutInSeconds =
-        (parentStatement == null) ? 0 : parentStatement.getStatement().getQueryTimeout();
+    int timeoutInSeconds;
+    if (parentStatement != null) {
+      timeoutInSeconds = parentStatement.getStatement().getQueryTimeout();
+    } else if (statementType == StatementType.METADATA) {
+      timeoutInSeconds = connectionContext.getMetadataOperationTimeout();
+    } else {
+      timeoutInSeconds = 0;
+    }
 
     TGetOperationStatusResp statusResp = null;
     if (response.isSetDirectResults()) {
diff --git a/src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java b/src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java
index 66cd3afde..fcb9b8ab3 100644
--- a/src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java
+++ b/src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java
@@ -602,6 +602,113 @@ public void testExecuteStatementWithTimeoutExpired() throws Exception {
     verify(databricksSdkClient).cancelStatement(eq(STATEMENT_ID));
   }
 
+  @Test
+  public void testMetadataOperationUsesMetadataTimeout() throws Exception {
+    // MetadataOperationTimeout=1 with parentStatement=null (metadata path)
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(
+            JDBC_URL,
+            new Properties() {
+              {
+                setProperty("MetadataOperationTimeout", "1");
+                setProperty("asyncExecPollInterval", "1000");
+              }
+            });
+    DatabricksSdkClient databricksSdkClient =
+        spy(new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient));
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+
+    CreateSessionResponse sessionResponse = new CreateSessionResponse().setSessionId(SESSION_ID);
+    when(apiClient.execute(any(Request.class), eq(CreateSessionResponse.class)))
+        .thenReturn(sessionResponse);
+    connection.open();
+
+    ExecuteStatementResponse executeResponse =
+        new ExecuteStatementResponse()
+            .setStatementId(STATEMENT_ID.toSQLExecStatementId())
+            .setStatus(new StatementStatus().setState(StatementState.RUNNING));
+    GetStatementResponse runningResponse =
+        new GetStatementResponse()
+            .setStatus(new StatementStatus().setState(StatementState.RUNNING));
+
+    when(apiClient.execute(
+            argThat(req -> req != null && STATEMENT_PATH.equals(req.getUrl())),
+            eq(ExecuteStatementResponse.class)))
+        .thenReturn(executeResponse);
+    when(apiClient.execute(
+            argThat(
+                req ->
+                    req != null
+                        && req.getUrl() != null
+                        && req.getUrl().contains(STATEMENT_ID.toSQLExecStatementId())),
+            eq(GetStatementResponse.class)))
+        .thenReturn(runningResponse);
+
+    // Metadata with parentStatement=null should use MetadataOperationTimeout (1s)
+    DatabricksTimeoutException exception =
+        assertThrows(
+            DatabricksTimeoutException.class,
+            () ->
+                databricksSdkClient.executeStatement(
+                    "SHOW SCHEMAS IN ALL CATALOGS",
+                    warehouse,
+                    new java.util.HashMap<>(),
+                    StatementType.METADATA,
+                    connection.getSession(),
+                    null, // parentStatement=null (metadata path)
+                    null));
+
+    assertTrue(exception.getMessage().contains("timed-out after 1 seconds"));
+    verify(databricksSdkClient).cancelStatement(eq(STATEMENT_ID));
+  }
+
+  @Test
+  public void testNonMetadataWithNullParentHasNoTimeout() throws Exception {
+    // Non-metadata with parentStatement=null should have timeout=0 (infinite)
+    // Use a short poll interval so the test completes quickly
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(
+            JDBC_URL,
+            new Properties() {
+              {
+                setProperty("MetadataOperationTimeout", "1");
+              }
+            });
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+
+    CreateSessionResponse sessionResponse = new CreateSessionResponse().setSessionId(SESSION_ID);
+    when(apiClient.execute(any(Request.class), eq(CreateSessionResponse.class)))
+        .thenReturn(sessionResponse);
+    connection.open();
+
+    // Return SUCCEEDED immediately so the test completes
+    ExecuteStatementResponse executeResponse =
+        new ExecuteStatementResponse()
+            .setStatementId(STATEMENT_ID.toSQLExecStatementId())
+            .setStatus(new StatementStatus().setState(StatementState.SUCCEEDED));
+
+    when(apiClient.execute(
+            argThat(req -> req != null && STATEMENT_PATH.equals(req.getUrl())),
+            eq(ExecuteStatementResponse.class)))
+        .thenReturn(executeResponse);
+
+    // Non-METADATA with parentStatement=null: no timeout applied, should succeed
+    assertDoesNotThrow(
+        () ->
+            databricksSdkClient.executeStatement(
+                "SELECT 1",
+                warehouse,
+                new java.util.HashMap<>(),
+                StatementType.SQL,
+                connection.getSession(),
+                null,
+                null));
+  }
+
   @Test
   public void testServerSideTimeoutThrowsTimeoutException() throws Exception {