diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
index 0a0cb8f6290..ccb595dbf0e 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
@@ -24,6 +24,7 @@ import scala.util.Try
 
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.spark.sql.kyuubi.SparkSQLKeywordHelper
 import org.apache.spark.ui.SparkUIUtils.formatDuration
 
 import org.apache.kyuubi.KyuubiSQLException
@@ -117,7 +118,8 @@ class SparkSessionImpl(
       case TGetInfoType.CLI_SERVER_NAME | TGetInfoType.CLI_DBMS_NAME =>
         TGetInfoValue.stringValue("Spark SQL")
       case TGetInfoType.CLI_DBMS_VER => TGetInfoValue.stringValue(org.apache.spark.SPARK_VERSION)
-      case TGetInfoType.CLI_ODBC_KEYWORDS => TGetInfoValue.stringValue("Unimplemented")
+      case TGetInfoType.CLI_ODBC_KEYWORDS =>
+        TGetInfoValue.stringValue(SparkSQLKeywordHelper.keywords.mkString(","))
       case TGetInfoType.CLI_MAX_COLUMN_NAME_LEN |
           TGetInfoType.CLI_MAX_SCHEMA_NAME_LEN |
           TGetInfoType.CLI_MAX_TABLE_NAME_LEN => TGetInfoValue.lenValue(128)
diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/kyuubi/SparkSQLKeywordHelper.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/kyuubi/SparkSQLKeywordHelper.scala
new file mode 100644
index 00000000000..dcbba0aff92
--- /dev/null
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/sql/kyuubi/SparkSQLKeywordHelper.scala
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.kyuubi
+
+import org.apache.spark.sql.catalyst.util.SQLKeywordUtils
+
+// SQLKeywordUtils is private[sql], so access it through a helper in the Spark SQL package.
+object SparkSQLKeywordHelper {
+
+  def keywords: Seq[String] = SQLKeywordUtils.keywords
+}
diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/session/SessionSuite.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/session/SessionSuite.scala
index b89c560b30c..1dc91547a79 100644
--- a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/session/SessionSuite.scala
+++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/session/SessionSuite.scala
@@ -24,6 +24,7 @@ import org.apache.kyuubi.config.KyuubiConf._
 import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
 import org.apache.kyuubi.operation.HiveJDBCTestHelper
 import org.apache.kyuubi.service.ServiceState._
+import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TGetInfoReq, TGetInfoType, TProtocolVersion, TStatusCode}
 
 class SessionSuite extends WithSparkSQLEngine with HiveJDBCTestHelper {
   override def withKyuubiConf: Map[String, String] = {
@@ -43,7 +44,35 @@ class SessionSuite extends WithSparkSQLEngine with HiveJDBCTestHelper {
   }
 
   override protected def jdbcUrl: String =
-    s"jdbc:hive2://${engine.frontendServices.head.connectionUrl}/;#spark.ui.enabled=false"
+    s"jdbc:hive2://${engine.frontendServices.head.connectionUrl}/;" +
+      s"clientProtocolVersion=${TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V11.getValue};" +
+      "#spark.ui.enabled=false"
+
+  private def assertSQLKeywords(value: String): Unit = {
+    val keywords = value.split(",", -1).toSeq
+    assert(
+      Set("SELECT", "FROM", "WHERE").subsetOf(keywords.toSet),
+      s"Expected Spark SQL keywords, but received: $value")
+    assert(keywords.forall(_.matches("[A-Z_]+")))
+    assert(keywords == keywords.sorted)
+  }
+
+  test("KYUUBI #5860: get Spark SQL keywords through Thrift GetInfo") {
+    withSessionHandle { (client, handle) =>
+      val request = new TGetInfoReq()
+      request.setSessionHandle(handle)
+      request.setInfoType(TGetInfoType.CLI_ODBC_KEYWORDS)
+      val response = client.GetInfo(request)
+      assert(response.getStatus.getStatusCode == TStatusCode.SUCCESS_STATUS)
+      assertSQLKeywords(response.getInfoValue.getStringValue)
+    }
+  }
+
+  test("KYUUBI #5860: get Spark SQL keywords through JDBC metadata") {
+    withJdbcStatement() { statement =>
+      assertSQLKeywords(statement.getConnection.getMetaData.getSQLKeywords)
+    }
+  }
 
   test("release session if shared level is CONNECTION") {
     assert(engine.getServiceState == STARTED)