content: /**
   * Copyright 2020 Alibaba Group Holding Limited.
   *
   * <p>Licensed 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
   *
   * <p>http://www.apache.org/licenses/LICENSE-2.0
   *
   * <p>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 com.alibaba.graphscope.groot.schema.ddl;
  
  import com.alibaba.graphscope.groot.common.exception.DdlException;
  import com.alibaba.graphscope.groot.common.schema.wrapper.GraphDef;
  import com.alibaba.graphscope.groot.common.schema.wrapper.LabelId;
  import com.alibaba.graphscope.groot.common.schema.wrapper.PropertyDef;
  import com.alibaba.graphscope.groot.common.schema.wrapper.TypeDef;
  import com.alibaba.graphscope.groot.common.schema.wrapper.TypeEnum;
  import com.alibaba.graphscope.groot.operation.Operation;
  import com.alibaba.graphscope.proto.groot.TypeDefPb;
  import com.google.protobuf.ByteString;
  import com.google.protobuf.InvalidProtocolBufferException;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  
  public abstract class AbstractCreateTypeExecutor extends AbstractDdlExecutor {
  
      private static final String NAME_REGEX = "^\\w{1,128}$";
  
      @Override
      public DdlResult execute(ByteString ddlBlob, GraphDef graphDef, int partitionCount)
              throws InvalidProtocolBufferException {
          TypeDefPb typeDefPb = TypeDefPb.parseFrom(ddlBlob);
          TypeDef typeDef = TypeDef.parseProto(typeDefPb);
          long version = graphDef.getSchemaVersion();
          String label = typeDef.getLabel();
  
          if (!label.matches(NAME_REGEX)) {
              throw new DdlException("illegal label name [" + label + "]");
          }
  
          if (graphDef.hasLabel(label)) {
              throw new DdlException(
                      "label [" + label + "] already exists, schema version [" + version + "]");
          }
  
          if (typeDef.getTypeEnum() == TypeEnum.VERTEX) {
              if (this instanceof CreateEdgeTypeExecutor) {
                  throw new DdlException("Expect edge type but got vertex type");
              }
              if (typeDef.getPkIdxs().size() == 0) {
                  throw new DdlException(
                          "Vertex type must define primary key. label [" + label + "]");
              }
          } else if (this instanceof CreateVertexTypeExecutor) {
              throw new DdlException("Expect vertex type but got edge type");
          }
          version++;
  
          GraphDef.Builder graphDefBuilder = GraphDef.newBuilder(graphDef);
          TypeDef.Builder typeDefBuilder = TypeDef.newBuilder(typeDef);
  
          int propertyIdx = graphDef.getPropertyIdx();
          Map<String, Integer> propertyNameToId = graphDef.getPropertyNameToId();
          List<PropertyDef> inputPropertiesInfo = typeDef.getProperties();
          checkDuplicatedPropertiesExists(inputPropertiesInfo);
          List<PropertyDef> propertyDefs = new ArrayList<>(inputPropertiesInfo.size());
          for (PropertyDef property : inputPropertiesInfo) {
              String propertyName = property.getName();
              if (!propertyName.matches(NAME_REGEX)) {
                  throw new DdlException("illegal property name [" + propertyName + "]");
              }
              Integer propertyId = propertyNameToId.get(propertyName);
              if (propertyId == null) {
                  propertyIdx++;
                  propertyId = propertyIdx;
                  graphDefBuilder.putPropertyNameToId(propertyName, propertyId);
                  graphDefBuilder.setPropertyIdx(propertyIdx);
              }
              propertyDefs.add(
                      PropertyDef.newBuilder(property)
                              .setId(propertyId)
                              .setInnerId(propertyId)
                              .build());
          }
          typeDefBuilder.setPropertyDefs(propertyDefs);
          int labelIdx = graphDef.getLabelIdx() + 1;
          LabelId labelId = new LabelId(labelIdx);
          typeDefBuilder.setLabelId(labelId);
          typeDefBuilder.setVersionId((int) version);
          TypeDef newTypeDef = typeDefBuilder.build();
  
          graphDefBuilder.setVersion(version);
          graphDefBuilder.addTypeDef(newTypeDef);
          graphDefBuilder.setLabelIdx(labelIdx);
  
          if (typeDef.getTypeEnum() == TypeEnum.VERTEX) {
              long tableIdx = graphDef.getTableIdx();
              tableIdx++;
              graphDefBuilder.putVertexTableId(labelId, tableIdx);
              graphDefBuilder.setTableIdx(tableIdx);
          }
          GraphDef newGraphDef = graphDefBuilder.build();
  
          List<Operation> operations = new ArrayList<>(partitionCount);
          for (int i = 0; i < partitionCount; i++) {
              Operation operation = makeOperation(i, version, newTypeDef, newGraphDef);
              operations.add(operation);
          }
          return new DdlResult(newGraphDef, operations);
      }
  
      protected abstract Operation makeOperation(
              int partition, long version, TypeDef typeDef, GraphDef graphDef);
  }
  