Add low-level create table and table exists API

core/src/main/java/org/opensearch/sql/storage/Table.java

@@ -16,6 +16,18 @@
  */
 public interface Table {
 
+  /**
+   * Check if current table exists.
+   * @return true if exists, otherwise false
+   */
+  boolean exists();
+
+  /**
+   * Create table given table schema.
+   * @param schema table schema
+   */
+  void create(Map<String, ExprType> schema);
+
   /**
    * Get the {@link ExprType} for each field in the table.
    */

core/src/test/java/org/opensearch/sql/analysis/AnalyzerTestBase.java

@@ -51,6 +51,16 @@ public Table getTable(String name) {
   @Bean
   protected Table table() {
     return new Table() {
+      @Override
+      public boolean exists() {
+        return true;
+      }
+
+      @Override
+      public void create(Map<String, ExprType> schema) {
+        throw new UnsupportedOperationException("Create table is not supported");
+      }
+
       @Override
       public Map<String, ExprType> getFieldTypes() {
         return typeMapping();

core/src/test/java/org/opensearch/sql/config/TestConfig.java

@@ -63,6 +63,16 @@ protected StorageEngine storageEngine() {
       @Override
       public Table getTable(String name) {
         return new Table() {
+          @Override
+          public boolean exists() {
+            return true;
+          }
+
+          @Override
+          public void create(Map<String, ExprType> schema) {
+            throw new UnsupportedOperationException("Create table is not supported");
+          }
+
           @Override
           public Map<String, ExprType> getFieldTypes() {
             return typeMapping;

core/src/test/java/org/opensearch/sql/planner/PlannerTest.java

@@ -119,6 +119,16 @@ protected PhysicalPlan analyze(LogicalPlan logicalPlan) {
 
   protected class MockTable extends LogicalPlanNodeVisitor<PhysicalPlan, Object> implements Table {
 
+    @Override
+    public boolean exists() {
+      return true;
+    }
+
+    @Override
+    public void create(Map<String, ExprType> schema) {
+      throw new UnsupportedOperationException("Create table is not supported");
+    }
+
     @Override
     public Map<String, ExprType> getFieldTypes() {
       throw new UnsupportedOperationException();

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java

@@ -22,6 +22,20 @@ public interface OpenSearchClient {
 
   String META_CLUSTER_NAME = "CLUSTER_NAME";
 
+  /**
+   * Check if the given index exists.
+   * @param indexName index name
+   * @return true if exists, otherwise false
+   */
+  boolean exists(String indexName);
+
+  /**
+   * Create OpenSearch index based on the given mappings.
+   * @param indexName index name
+   * @param mappings  index mappings
+   */
+  void createIndex(String indexName, Map<String, Object> mappings);
+
   /**
    * Fetch index mapping(s) according to index expression given.
    *

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java

@@ -18,6 +18,9 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.opensearch.action.admin.indices.create.CreateIndexRequest;
+import org.opensearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.opensearch.action.admin.indices.exists.indices.IndicesExistsResponse;
 import org.opensearch.action.admin.indices.get.GetIndexResponse;
 import org.opensearch.action.admin.indices.mapping.get.GetMappingsResponse;
 import org.opensearch.action.admin.indices.settings.get.GetSettingsResponse;
@@ -50,6 +53,28 @@ public OpenSearchNodeClient(NodeClient client) {
     this.resolver = new IndexNameExpressionResolver(client.threadPool().getThreadContext());
   }
 
+  @Override
+  public boolean exists(String indexName) {
+    try {
+      IndicesExistsResponse checkExistResponse = client.admin().indices()
+          .exists(new IndicesExistsRequest(indexName)).actionGet();
+      return checkExistResponse.isExists();
+    } catch (Exception e) {
+      throw new IllegalStateException("Failed to check if index [" + indexName + "] exists", e);
+    }
+  }
+
+  @Override
+  public void createIndex(String indexName, Map<String, Object> mappings) {
+    try {
+      // TODO: 1.pass index settings (the number of primary shards, etc); 2.check response?
+      CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName).mapping(mappings);
+      client.admin().indices().create(createIndexRequest).actionGet();
+    } catch (Exception e) {
+      throw new IllegalStateException("Failed to create index [" + indexName + "]", e);
+    }
+  }
+
   /**
    * Get field mappings of index by an index expression. Majority is copied from legacy
    * LocalClusterState.

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java

@@ -23,6 +23,7 @@
 import org.opensearch.action.search.ClearScrollRequest;
 import org.opensearch.client.RequestOptions;
 import org.opensearch.client.RestHighLevelClient;
+import org.opensearch.client.indices.CreateIndexRequest;
 import org.opensearch.client.indices.GetIndexRequest;
 import org.opensearch.client.indices.GetIndexResponse;
 import org.opensearch.client.indices.GetMappingsRequest;
@@ -46,6 +47,26 @@ public class OpenSearchRestClient implements OpenSearchClient {
   /** OpenSearch high level REST client. */
   private final RestHighLevelClient client;
 
+  @Override
+  public boolean exists(String indexName) {
+    try {
+      return client.indices().exists(
+          new GetIndexRequest(indexName), RequestOptions.DEFAULT);
+    } catch (IOException e) {
+      throw new IllegalStateException("Failed to check if index [" + indexName + "] exist", e);
+    }
+  }
+
+  @Override
+  public void createIndex(String indexName, Map<String, Object> mappings) {
+    try {
+      client.indices().create(
+          new CreateIndexRequest(indexName).mapping(mappings), RequestOptions.DEFAULT);
+    } catch (IOException e) {
+      throw new IllegalStateException("Failed to create index [" + indexName + "]", e);
+    }
+  }
+
   @Override
   public Map<String, IndexMapping> getIndexMappings(String... indexExpression) {
     GetMappingsRequest request = new GetMappingsRequest().indices(indexExpression);

opensearch/src/main/java/org/opensearch/sql/opensearch/data/type/OpenSearchDataType.java

@@ -9,12 +9,15 @@
 import static org.opensearch.sql.data.type.ExprCoreType.STRING;
 import static org.opensearch.sql.data.type.ExprCoreType.UNKNOWN;
 
+import com.google.common.collect.BiMap;
+import com.google.common.collect.ImmutableBiMap;
 import com.google.common.collect.ImmutableMap;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
+import org.opensearch.sql.data.type.ExprCoreType;
 import org.opensearch.sql.data.type.ExprType;
 
 /**
@@ -52,6 +55,39 @@ public boolean shouldCast(ExprType other) {
 
   OPENSEARCH_BINARY(Arrays.asList(UNKNOWN), "binary");
 
+  /**
+   * Bidirectional mapping between OpenSearch type name and ExprType.
+   */
+  private static final BiMap<String, ExprType> OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING =
+      ImmutableBiMap.<String, ExprType>builder()
+          .put("text", OPENSEARCH_TEXT)
+          .put("text_keyword", OPENSEARCH_TEXT_KEYWORD)
+          .put("keyword", ExprCoreType.STRING)
+          .put("byte", ExprCoreType.BYTE)
+          .put("short", ExprCoreType.SHORT)
+          .put("integer", ExprCoreType.INTEGER)
+          .put("long", ExprCoreType.LONG)
+          .put("float", ExprCoreType.FLOAT)
+          .put("double", ExprCoreType.DOUBLE)
+          .put("boolean", ExprCoreType.BOOLEAN)
+          .put("nested", ExprCoreType.ARRAY)
+          .put("object", ExprCoreType.STRUCT)
+          .put("date", ExprCoreType.TIMESTAMP)
+          .put("ip", OPENSEARCH_IP)
+          .put("geo_point", OPENSEARCH_GEO_POINT)
+          .put("binary", OPENSEARCH_BINARY)
+          .build();
+
+  /**
+   * Mapping from extra OpenSearch type name which may map to same ExprType as above.
+   */
+  private static final Map<String, ExprType> EXTRA_OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING =
+      ImmutableMap.<String, ExprType>builder()
+          .put("half_float", ExprCoreType.FLOAT)
+          .put("scaled_float", ExprCoreType.DOUBLE)
+          .put("date_nanos", ExprCoreType.TIMESTAMP)
+          .build();
+
   /**
    * The mapping between Type and legacy JDBC type name.
    */
@@ -70,6 +106,27 @@ public boolean shouldCast(ExprType other) {
    */
   private final String jdbcType;
 
+  /**
+   * Convert OpenSearch type string to ExprType.
+   * @param openSearchType OpenSearch type string
+   * @return expr type
+   */
+  public static ExprType getExprType(String openSearchType) {
+    if (OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING.containsKey(openSearchType)) {
+      return OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING.get(openSearchType);
+    }
+    return EXTRA_OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING.getOrDefault(openSearchType, UNKNOWN);
+  }
+
+  /**
+   * Convert ExprType to OpenSearch type string.
+   * @param type expr type
+   * @return OpenSearch type string
+   */
+  public static String getOpenSearchType(ExprType type) {
+    return OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING.inverse().get(type);
+  }
+
   @Override
   public List<ExprType> getParent() {
     return parents;

opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java

@@ -10,7 +10,6 @@
 import static org.opensearch.sql.data.model.ExprValueUtils.stringValue;
 import static org.opensearch.sql.opensearch.client.OpenSearchClient.META_CLUSTER_NAME;
 
-import com.google.common.collect.ImmutableMap;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -39,33 +38,6 @@ public class OpenSearchDescribeIndexRequest implements OpenSearchSystemRequest {
 
   private static final String DEFAULT_IS_AUTOINCREMENT = "NO";
 
-  /**
-   * Type mapping from OpenSearch data type to expression type in our type system in query
-   * engine. TODO: geo, ip etc.
-   */
-  private static final Map<String, ExprType> OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING =
-      ImmutableMap.<String, ExprType>builder()
-          .put("text", OpenSearchDataType.OPENSEARCH_TEXT)
-          .put("text_keyword", OpenSearchDataType.OPENSEARCH_TEXT_KEYWORD)
-          .put("keyword", ExprCoreType.STRING)
-          .put("byte", ExprCoreType.BYTE)
-          .put("short", ExprCoreType.SHORT)
-          .put("integer", ExprCoreType.INTEGER)
-          .put("long", ExprCoreType.LONG)
-          .put("float", ExprCoreType.FLOAT)
-          .put("half_float", ExprCoreType.FLOAT)
-          .put("scaled_float", ExprCoreType.DOUBLE)
-          .put("double", ExprCoreType.DOUBLE)
-          .put("boolean", ExprCoreType.BOOLEAN)
-          .put("nested", ExprCoreType.ARRAY)
-          .put("object", ExprCoreType.STRUCT)
-          .put("date", ExprCoreType.TIMESTAMP)
-          .put("date_nanos", ExprCoreType.TIMESTAMP)
-          .put("ip", OpenSearchDataType.OPENSEARCH_IP)
-          .put("geo_point", OpenSearchDataType.OPENSEARCH_GEO_POINT)
-          .put("binary", OpenSearchDataType.OPENSEARCH_BINARY)
-          .build();
-
   /**
    * OpenSearch client connection.
    */
@@ -132,7 +104,7 @@ public Integer getMaxResultWindow() {
   }
 
   private ExprType transformESTypeToExprType(String openSearchType) {
-    return OPENSEARCH_TYPE_TO_EXPR_TYPE_MAPPING.getOrDefault(openSearchType, ExprCoreType.UNKNOWN);
+    return OpenSearchDataType.getExprType(openSearchType);
   }
 
   private ExprTupleValue row(String fieldName, String fieldType, int position, String clusterName) {

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java

@@ -7,6 +7,7 @@
 package org.opensearch.sql.opensearch.storage;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -18,6 +19,7 @@
 import org.opensearch.sql.common.utils.StringUtils;
 import org.opensearch.sql.data.type.ExprType;
 import org.opensearch.sql.opensearch.client.OpenSearchClient;
+import org.opensearch.sql.opensearch.data.type.OpenSearchDataType;
 import org.opensearch.sql.opensearch.data.value.OpenSearchExprValueFactory;
 import org.opensearch.sql.opensearch.planner.logical.OpenSearchLogicalIndexAgg;
 import org.opensearch.sql.opensearch.planner.logical.OpenSearchLogicalIndexScan;
@@ -72,6 +74,23 @@ public OpenSearchIndex(OpenSearchClient client, Settings settings, String indexN
     this.indexName = new OpenSearchRequest.IndexName(indexName);
   }
 
+  @Override
+  public boolean exists() {
+    return client.exists(indexName.toString());
+  }
+
+  @Override
+  public void create(Map<String, ExprType> schema) {
+    Map<String, Object> mappings = new HashMap<>();
+    Map<String, Object> properties = new HashMap<>();
+    mappings.put("properties", properties);
+
+    for (Map.Entry<String, ExprType> colType : schema.entrySet()) {
+      properties.put(colType.getKey(), OpenSearchDataType.getOpenSearchType(colType.getValue()));
+    }
+    client.createIndex(indexName.toString(), mappings);
+  }
+
   /*
    * TODO: Assume indexName doesn't have wildcard.
    *  Need to either handle field name conflicts

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/system/OpenSearchSystemIndex.java

@@ -38,6 +38,17 @@ public OpenSearchSystemIndex(
     this.systemIndexBundle = buildIndexBundle(client, indexName);
   }
 
+  @Override
+  public boolean exists() {
+    return true; // TODO: implement for system index later
+  }
+
+  @Override
+  public void create(Map<String, ExprType> schema) {
+    throw new UnsupportedOperationException(
+        "OpenSearch system index is predefined and cannot be created");
+  }
+
   @Override
   public Map<String, ExprType> getFieldTypes() {
     return systemIndexBundle.getLeft().getMapping();