override def check()

in src/datasource-service/src/main/scala/org/apache/kylin/rest/ddl/ViewCheck.scala [92:160]
69 lines of code
22 McCabe index (conditional complexity)

  override def check(context: DDLCheckContext): Unit = {
    LOGGER.info("start checking DDL view name")
    var sql = context.getSql
    val project = context.getProject
    val spark = SparderEnv.getSparkSession
    val config = KylinConfig.getInstanceFromEnv
    var plan: SparkPlan = null
    try {
      sql = LogicalViewLoader.addCatalog(context.getSql, context.getProject, SparderEnv.getSparkSession)
      val logicalPlan = spark.sessionState.sqlParser.parsePlan(sql)
      plan = stripRootCommandResult(spark.sessionState.executePlan(
        logicalPlan, CommandExecutionMode.SKIP).executedPlan)
    } catch {
      case e: Exception => throwException(e.getMessage)
    }
    plan match {
      case ExecutedCommandExec(view: CreateViewCommand) =>
        if (view.viewType != null && LOGICAL_VIEW_TYPE.equalsIgnoreCase(view.viewType.toString())) {
          val viewManager = LogicalViewManager.getInstance(config)
          val originTable = viewManager.get(view.name.table)
          if (view.replace) {
            context.setCommandType(DDLConstant.REPLACE_LOGICAL_VIEW)
            if (originTable == null) {
              throwException("View name is not found.")
            }
            if (!originTable.getCreatedProject.equals(context.getProject)) {
              throwException(s"View can only modified in Project ${originTable.getCreatedProject}")
            }
          } else {
            if (originTable != null) {
              throwException(MsgPicker.getMsg.getDDLViewNameDuplicateError)
            }
            context.setCommandType(DDLConstant.CREATE_LOGICAL_VIEW)
          }
          context.setLogicalViewName(view.name.table)
        } else {
          checkHiveTableName(view.name, context)
          checkHiveDatabaseAccess(view.name, project, context)
        }
      case ExecutedCommandExec(view: ShowCreateTableCommand) =>
        checkHiveTableName(view.table, context)
        checkHiveDatabaseAccess(view.table, project, context)
      case ExecutedCommandExec(table: DropTableCommand) =>
        if (!table.isView) {
          throwException(MsgPicker.getMsg.getDDLDropError)
        }
        val tableIdentifier = table.tableName
        if (config.isDDLLogicalViewEnabled && tableIdentifier.database.isDefined
          && config.getDDLLogicalViewDB.equalsIgnoreCase(tableIdentifier.database.get)) {
          context.setCommandType(DDLConstant.DROP_LOGICAL_VIEW)
          context.setLogicalViewName(tableIdentifier.table)
          checkLogicalViewNotUsed(tableIdentifier, context.getProject)
        } else {
          checkHiveTableName(table.tableName, context)
          checkHiveDatabaseAccess(table.tableName, project, context)
        }
      case ExecutedCommandExec(table: AlterViewAsCommand) =>
        checkHiveTableName(table.name, context)
        checkHiveDatabaseAccess(table.name, project, context)
      case _ => throwException(MsgPicker.getMsg.getDDLUnSupported)
    }
    if (context.isLogicalViewCommand && !config.isDDLLogicalViewEnabled) {
      throwException("Logical View operation is not supported, please turn on config.")
    }
    if (context.isHiveCommand && !config.isDDLHiveEnabled) {
      throwException("Hive operation is not supported, please turn on config.")
    }
    checkCommandRestrict(context)
  }