Function executeOptimizeUsingOptFlow

go/executor/pai.go:470–534

(pythonExecutor *pythonExecutor, stmt *ir.OptimizeStmt)

}

func executeOptimizeUsingOptFlow(pythonExecutor *pythonExecutor, stmt *ir.OptimizeStmt) error {
	dbName, tableName, err := createTmpTableFromSelect(stmt.Select, pythonExecutor.Session.DbConnStr)
	if err != nil {
		return err
	}

	dropTmpTableFunc := func(table string) {
		dropTmpTables([]string{table}, pythonExecutor.Session.DbConnStr)
	}

	if len(stmt.Variables) > 2 {
		joinedVarName := strings.Join(stmt.Variables, "__")
		concatColumnNames := make([]string, 0)
		for i, v := range stmt.Variables {
			if i >= 1 {
				concatColumnNames = append(concatColumnNames, `','`)
			}
			concatColumnNames = append(concatColumnNames, v)
		}
		concatColumnExpr := fmt.Sprintf("CONCAT(%s) AS %s", strings.Join(concatColumnNames, ","), joinedVarName)
		selectStmt := fmt.Sprintf("SELECT *, %s FROM %s.%s", concatColumnExpr, dbName, tableName)
		newDBName, newTableName, err := createTmpTableFromSelect(selectStmt, pythonExecutor.Session.DbConnStr)
		dropTmpTableFunc(tableName) // drop the first created table whatever
		if err != nil {
			return err
		}
		stmt.Variables = []string{joinedVarName}
		dbName = newDBName
		tableName = newTableName
	}

	defer dropTmpTableFunc(tableName)

	db, err := database.OpenAndConnectDB(pythonExecutor.Session.DbConnStr)
	if err != nil {
		return err
	}
	defer db.Close()

	splittedResultTable := strings.SplitN(stmt.ResultTable, ".", 2)
	var resultTable string
	if len(splittedResultTable) == 2 {
		if splittedResultTable[0] != dbName {
			return fmt.Errorf("database name of result table must be the same as source table")
		}
		resultTable = stmt.ResultTable
	} else {
		resultTable = fmt.Sprintf("%s.%s", dbName, stmt.ResultTable)
	}

	_, err = db.Exec(fmt.Sprintf("DROP TABLE IF EXISTS %s", resultTable))
	if err != nil {
		return err
	}

	code, err := optimize.GenerateOptimizeCode(stmt, pythonExecutor.Session, tableName, true)
	if err != nil {
		return err