SQL: Introduce ODBC mode, similar to JDBC (#34825)

Close #34720

(cherry picked from commit b97546a)
(cherry picked from commit df125af)

Author: costin

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -239,7 +239,8 @@ private static String[] sqlAcknowledgementMessages(OperationMode currentMode, Op
                 switch (currentMode) {
                     case TRIAL:
                     case PLATINUM:
-                        return new String[] { "JDBC support will be disabled, but you can continue to use SQL CLI and REST endpoint" };
+                        return new String[] {
+                                "JDBC and ODBC support will be disabled, but you can continue to use SQL CLI and REST endpoint" };
                 }
                 break;
         }
@@ -627,6 +628,20 @@ public synchronized boolean isJdbcAllowed() {
         return licensed && localStatus.active;
     }
 
+    /**
+     * Determine if ODBC support should be enabled.
+     * <p>
+     * ODBC is available only in for {@link OperationMode#PLATINUM} and {@link OperationMode#TRIAL} licences
+     */
+    public synchronized boolean isOdbcAllowed() {
+        Status localStatus = status;
+        OperationMode operationMode = localStatus.mode;
+
+        boolean licensed = operationMode == OperationMode.TRIAL || operationMode == OperationMode.PLATINUM;
+
+        return licensed && localStatus.active;
+    }
+
     public synchronized boolean isTrialLicense() {
         return status.mode == OperationMode.TRIAL;
     }

x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/SqlQueryResponse.java

@@ -167,7 +167,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
      * Serializes the provided value in SQL-compatible way based on the client mode
      */
     public static XContentBuilder value(XContentBuilder builder, Mode mode, Object value) throws IOException {
-        if (mode == Mode.JDBC && value instanceof ReadableDateTime) {
+        if (Mode.isDriver(mode) && value instanceof ReadableDateTime) {
             // JDBC cannot parse dates in string format
             builder.value(((ReadableDateTime) value).getMillis());
         } else {

x-pack/plugin/sql/sql-proto/src/main/java/org/elasticsearch/xpack/sql/proto/Mode.java

@@ -13,7 +13,8 @@
  */
 public enum Mode {
     PLAIN,
-    JDBC;
+    JDBC, 
+    ODBC;
 
     public static Mode fromString(String mode) {
         if (mode == null) {
@@ -27,4 +28,8 @@ public static Mode fromString(String mode) {
     public String toString() {
         return this.name().toLowerCase(Locale.ROOT);
     }
+
+    public static boolean isDriver(Mode mode) {
+        return mode == JDBC || mode == ODBC;
+    }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlPlugin.java

@@ -64,6 +64,11 @@ public SqlPlugin(Settings settings) {
                                 throw LicenseUtils.newComplianceException("jdbc");
                             }
                             break;
+                        case ODBC:
+                            if (licenseState.isOdbcAllowed() == false) {
+                                throw LicenseUtils.newComplianceException("odbc");
+                            }
+                            break;
                         case PLAIN:
                             if (licenseState.isSqlAllowed() == false) {
                                 throw LicenseUtils.newComplianceException(XPackField.SQL);

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/TransportSqlQueryAction.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.xpack.sql.action.SqlQueryResponse;
 import org.elasticsearch.xpack.sql.execution.PlanExecutor;
 import org.elasticsearch.xpack.sql.proto.ColumnInfo;
+import org.elasticsearch.xpack.sql.proto.Mode;
 import org.elasticsearch.xpack.sql.session.Configuration;
 import org.elasticsearch.xpack.sql.session.Cursors;
 import org.elasticsearch.xpack.sql.session.RowSet;
@@ -30,7 +31,6 @@
 import java.util.List;
 
 import static java.util.Collections.unmodifiableList;
-import static org.elasticsearch.xpack.sql.proto.Mode.JDBC;
 
 public class TransportSqlQueryAction extends HandledTransportAction<SqlQueryRequest, SqlQueryResponse> {
     private final PlanExecutor planExecutor;
@@ -76,7 +76,7 @@ public static void operation(PlanExecutor planExecutor, SqlQueryRequest request,
     static SqlQueryResponse createResponse(SqlQueryRequest request, SchemaRowSet rowSet) {
         List<ColumnInfo> columns = new ArrayList<>(rowSet.columnCount());
         for (Schema.Entry entry : rowSet.schema()) {
-            if (request.mode() == JDBC) {
+            if (Mode.isDriver(request.mode())) {
                 columns.add(new ColumnInfo("", entry.name(), entry.type().esType, entry.type().jdbcType,
                         entry.type().displaySize));
             } else {