private void create_table_core(final RawStore ms, final CreateTableRequest req)
      throws AlreadyExistsException, MetaException,
      InvalidObjectException, NoSuchObjectException, InvalidInputException {
    ColumnStatistics colStats = null;
    Table tbl = req.getTable();
    EnvironmentContext envContext = req.getEnvContext();
    SQLAllTableConstraints constraints = new SQLAllTableConstraints();
    constraints.setPrimaryKeys(req.getPrimaryKeys());
    constraints.setForeignKeys(req.getForeignKeys());
    constraints.setUniqueConstraints(req.getUniqueConstraints());
    constraints.setDefaultConstraints(req.getDefaultConstraints());
    constraints.setCheckConstraints(req.getCheckConstraints());
    constraints.setNotNullConstraints(req.getNotNullConstraints());
    List<String> processorCapabilities = req.getProcessorCapabilities();
    String processorId = req.getProcessorIdentifier();

    // To preserve backward compatibility throw MetaException in case of null database
    if (tbl.getDbName() == null) {
      throw new MetaException("Null database name is not allowed");
    }

    if (!MetaStoreUtils.validateName(tbl.getTableName(), conf)) {
      throw new InvalidObjectException(tbl.getTableName()
          + " is not a valid object name");
    }

    if (!MetaStoreUtils.validateTblStorage(tbl.getSd())) {
      throw new InvalidObjectException(tbl.getTableName()
              + " location must not be root path");
    }

    if (!tbl.isSetCatName()) {
      tbl.setCatName(getDefaultCatalog(conf));
    }

    Database db = get_database_core(tbl.getCatName(), tbl.getDbName());
    if (MetaStoreUtils.isDatabaseRemote(db)) {
      // HIVE-24425: Create table in REMOTE db should fail
      throw new MetaException("Create table in REMOTE database " + db.getName() + " is not allowed");
    }

    if (is_table_exists(ms, tbl.getCatName(), tbl.getDbName(), tbl.getTableName())) {
      throw new AlreadyExistsException("Table " + getCatalogQualifiedTableName(tbl)
          + " already exists");
    }

    tbl.setDbName(normalizeIdentifier(tbl.getDbName()));
    tbl.setTableName(normalizeIdentifier(tbl.getTableName()));

    if (transformer != null) {
      tbl = transformer.transformCreateTable(tbl, processorCapabilities, processorId);
    }

    Map<String, String> params = tbl.getParameters();
    if (params != null) {
      params.remove(TABLE_IS_CTAS);
      params.remove(TABLE_IS_CTLT);
      if (MetaStoreServerUtils.getBooleanEnvProp(envContext, CTAS_LEGACY_CONFIG) &&
          TableType.MANAGED_TABLE.toString().equals(tbl.getTableType())) {
        params.put("EXTERNAL", "TRUE");
        tbl.setTableType(TableType.EXTERNAL_TABLE.toString());
      }
    }

    // If the given table has column statistics, save it here. We will update it later.
    // We don't want it to be part of the Table object being created, lest the create table
    // event will also have the col stats which we don't want.
    if (tbl.isSetColStats()) {
      colStats = tbl.getColStats();
      tbl.unsetColStats();
    }

    String validate = MetaStoreServerUtils.validateTblColumns(tbl.getSd().getCols());
    if (validate != null) {
      throw new InvalidObjectException("Invalid column " + validate);
    }
    if (tbl.getPartitionKeys() != null) {
      validate = MetaStoreServerUtils.validateTblColumns(tbl.getPartitionKeys());
      if (validate != null) {
        throw new InvalidObjectException("Invalid partition column " + validate);
      }
    }
    if (tbl.isSetId()) {
      LOG.debug("Id shouldn't be set but table {}.{} has the Id set to {}. Id is ignored.", tbl.getDbName(),
          tbl.getTableName(), tbl.getId());
      tbl.unsetId();
    }
    SkewedInfo skew = tbl.getSd().getSkewedInfo();
    if (skew != null) {
      validate = MetaStoreServerUtils.validateSkewedColNames(skew.getSkewedColNames());
      if (validate != null) {
        throw new InvalidObjectException("Invalid skew column " + validate);
      }
      validate = MetaStoreServerUtils.validateSkewedColNamesSubsetCol(
          skew.getSkewedColNames(), tbl.getSd().getCols());
      if (validate != null) {
        throw new InvalidObjectException("Invalid skew column " + validate);
      }
    }

    Map<String, String> transactionalListenerResponses = Collections.emptyMap();
    Path tblPath = null;
    boolean success = false, madeDir = false;
    boolean isReplicated = false;
    try {

      ms.openTransaction();

      db = ms.getDatabase(tbl.getCatName(), tbl.getDbName());
      isReplicated = isDbReplicationTarget(db);

      firePreEvent(new PreCreateTableEvent(tbl, db, this));

      if (!TableType.VIRTUAL_VIEW.toString().equals(tbl.getTableType())) {
        if (tbl.getSd().getLocation() == null
            || tbl.getSd().getLocation().isEmpty()) {
          tblPath = wh.getDefaultTablePath(db, tbl.getTableName() + getTableSuffix(tbl),
              MetaStoreUtils.isExternalTable(tbl));
        } else {
          if (!MetaStoreUtils.isExternalTable(tbl) && !MetaStoreUtils.isNonNativeTable(tbl)) {
            LOG.warn("Location: " + tbl.getSd().getLocation()
                + " specified for non-external table:" + tbl.getTableName());
          }
          tblPath = wh.getDnsPath(new Path(tbl.getSd().getLocation()));
          // ignore suffix if it's already there (direct-write CTAS)
          if (!tblPath.getName().matches("(.*)" + SOFT_DELETE_TABLE_PATTERN)) {
            tblPath = new Path(tblPath + getTableSuffix(tbl));
          }
        }
        tbl.getSd().setLocation(tblPath.toString());
      }

      if (tblPath != null) {
        if (!wh.isDir(tblPath)) {
          if (!wh.mkdirs(tblPath)) {
            throw new MetaException(tblPath
                + " is not a directory or unable to create one");
          }
          madeDir = true;
        }
      }

      MetaStoreServerUtils.updateTableStatsForCreateTable(wh, db, tbl, envContext, conf, tblPath, madeDir);

      // set create time
      long time = System.currentTimeMillis() / 1000;
      tbl.setCreateTime((int) time);
      if (tbl.getParameters() == null ||
          tbl.getParameters().get(hive_metastoreConstants.DDL_TIME) == null) {
        tbl.putToParameters(hive_metastoreConstants.DDL_TIME, Long.toString(time));
      }

      if (CollectionUtils.isEmpty(constraints.getPrimaryKeys()) && CollectionUtils.isEmpty(constraints.getForeignKeys())
          && CollectionUtils.isEmpty(constraints.getUniqueConstraints())&& CollectionUtils.isEmpty(constraints.getNotNullConstraints())&& CollectionUtils.isEmpty(constraints.getDefaultConstraints())
          && CollectionUtils.isEmpty(constraints.getCheckConstraints())) {
        ms.createTable(tbl);
      } else {
        final String catName = tbl.getCatName();
        // Check that constraints have catalog name properly set first
        if (CollectionUtils.isNotEmpty(constraints.getPrimaryKeys()) && !constraints.getPrimaryKeys().get(0).isSetCatName()) {
          constraints.getPrimaryKeys().forEach(constraint -> constraint.setCatName(catName));
        }
        if (CollectionUtils.isNotEmpty(constraints.getForeignKeys()) && !constraints.getForeignKeys().get(0).isSetCatName()) {
          constraints.getForeignKeys().forEach(constraint -> constraint.setCatName(catName));
        }
        if (CollectionUtils.isNotEmpty(constraints.getUniqueConstraints()) && !constraints.getUniqueConstraints().get(0).isSetCatName()) {
          constraints.getUniqueConstraints().forEach(constraint -> constraint.setCatName(catName));
        }
        if (CollectionUtils.isNotEmpty(constraints.getNotNullConstraints()) && !constraints.getNotNullConstraints().get(0).isSetCatName()) {
          constraints.getNotNullConstraints().forEach(constraint -> constraint.setCatName(catName));
        }
        if (CollectionUtils.isNotEmpty(constraints.getDefaultConstraints()) && !constraints.getDefaultConstraints().get(0).isSetCatName()) {
          constraints.getDefaultConstraints().forEach(constraint -> constraint.setCatName(catName));
        }
        if (CollectionUtils.isNotEmpty(constraints.getCheckConstraints()) && !constraints.getCheckConstraints().get(0).isSetCatName()) {
          constraints.getCheckConstraints().forEach(constraint -> constraint.setCatName(catName));
        }
        // Set constraint name if null before sending to listener
        constraints = ms.createTableWithConstraints(tbl, constraints);

      }

      if (!transactionalListeners.isEmpty()) {
        transactionalListenerResponses = MetaStoreListenerNotifier.notifyEvent(transactionalListeners,
            EventType.CREATE_TABLE, new CreateTableEvent(tbl, true, this, isReplicated), envContext);
        if (CollectionUtils.isNotEmpty(constraints.getPrimaryKeys())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_PRIMARYKEY,
              new AddPrimaryKeyEvent(constraints.getPrimaryKeys(), true, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getForeignKeys())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_FOREIGNKEY,
              new AddForeignKeyEvent(constraints.getForeignKeys(), true, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getUniqueConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_UNIQUECONSTRAINT,
              new AddUniqueConstraintEvent(constraints.getUniqueConstraints(), true, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getNotNullConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_NOTNULLCONSTRAINT,
              new AddNotNullConstraintEvent(constraints.getNotNullConstraints(), true, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getCheckConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_CHECKCONSTRAINT,
              new AddCheckConstraintEvent(constraints.getCheckConstraints(), true, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getDefaultConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.ADD_DEFAULTCONSTRAINT,
              new AddDefaultConstraintEvent(constraints.getDefaultConstraints(), true, this), envContext);
        }
      }

      success = ms.commitTransaction();
    } finally {
      if (!success) {
        ms.rollbackTransaction();
        if (madeDir) {
          wh.deleteDir(tblPath, true, false, ReplChangeManager.shouldEnableCm(db, tbl));
        }
      }

      if (!listeners.isEmpty()) {
        MetaStoreListenerNotifier.notifyEvent(listeners, EventType.CREATE_TABLE,
            new CreateTableEvent(tbl, success, this, isReplicated), envContext,
            transactionalListenerResponses, ms);
        if (CollectionUtils.isNotEmpty(constraints.getPrimaryKeys())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_PRIMARYKEY,
              new AddPrimaryKeyEvent(constraints.getPrimaryKeys(), success, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getForeignKeys())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_FOREIGNKEY,
              new AddForeignKeyEvent(constraints.getForeignKeys(), success, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getUniqueConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_UNIQUECONSTRAINT,
              new AddUniqueConstraintEvent(constraints.getUniqueConstraints(), success, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getNotNullConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_NOTNULLCONSTRAINT,
              new AddNotNullConstraintEvent(constraints.getNotNullConstraints(), success, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getDefaultConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_DEFAULTCONSTRAINT,
              new AddDefaultConstraintEvent(constraints.getDefaultConstraints(), success, this), envContext);
        }
        if (CollectionUtils.isNotEmpty(constraints.getCheckConstraints())) {
          MetaStoreListenerNotifier.notifyEvent(listeners, EventType.ADD_CHECKCONSTRAINT,
              new AddCheckConstraintEvent(constraints.getCheckConstraints(), success, this), envContext);
        }
      }
    }

    // If the table has column statistics, update it into the metastore. We need a valid
    // writeId list to update column statistics for a transactional table. But during bootstrap
    // replication, where we use this feature, we do not have a valid writeId list which was
    // used to update the stats. But we know for sure that the writeId associated with the
    // stats was valid then (otherwise stats update would have failed on the source). So, craft
    // a valid transaction list with only that writeId and use it to update the stats.
    if (colStats != null) {
      long writeId = tbl.getWriteId();
      String validWriteIds = null;
      if (writeId > 0) {
        ValidWriteIdList validWriteIdList =
            new ValidReaderWriteIdList(TableName.getDbTable(tbl.getDbName(),
                tbl.getTableName()),
                new long[0], new BitSet(), writeId);
        validWriteIds = validWriteIdList.toString();
      }
      updateTableColumnStatsInternal(colStats, validWriteIds, tbl.getWriteId());
    }
  }
Checks whether the repl.last.id is being updated.
@param oldDb the old db object
@param newDb the new db object
@return true if repl.last.id is being changed.
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java