/* -------------------------------------------------------------------------
 *  This file is part of the oGRAC project.
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * oGRAC is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 *          http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * ddl_parser.c
 *
 *
 * IDENTIFICATION
 * src/ogsql/parser_ddl/ddl_parser.c
 *
 * -------------------------------------------------------------------------
 */
#include "ddl_parser.h"
#include "cm_config.h"
#include "cm_hash.h"
#include "cm_file.h"
#include "dml_parser.h"
#include "ddl_user_parser.h"
#include "ddl_table_parser.h"
#include "ddl_database_parser.h"
#include "ddl_space_parser.h"
#include "ddl_index_parser.h"
#include "ddl_privilege_parser.h"
#include "ddl_column_parser.h"
#include "ddl_partition_parser.h"
#include "ddl_view_parser.h"
#include "ogsql_parser.h"
#include "ddl_parser_common.h"
#include "cm_license.h"
#include "ogsql_privilege.h"
#include "cm_defs.h"
#include "pl_ddl_parser.h"
#include "srv_param_common.h"
#ifdef OG_RAC_ING
#include "shd_parser.h"
#include "shd_slowsql.h"
#include "shd_ddl_executor.h"
#include "shd_transform.h"

#endif
#ifdef __cplusplus
extern "C" {
#endif

static status_t sql_create_or_replace_lead(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;
    bool32 is_force = OG_FALSE;

    if (lex_expected_fetch_word(stmt->session->lex, "REPLACE") != OG_SUCCESS) {
        return OG_ERROR;
    }

    OG_RETURN_IFERR(lex_try_fetch(stmt->session->lex, "FORCE", &is_force));

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch (word.id) {
        case KEY_WORD_VIEW: {
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_view(stmt, OG_TRUE, is_force);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }
        case KEY_WORD_PUBLIC: {
            if (!g_instance->sql.use_bison_parser) {
                if (lex_expected_fetch_word(stmt->session->lex, "SYNONYM") != OG_SUCCESS) {
                    return OG_ERROR;
                }

                status = sql_parse_create_synonym(stmt, SYNONYM_IS_PUBLIC + SYNONYM_IS_REPLACE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }
        case KEY_WORD_PACKAGE:
        case KEY_WORD_FUNCTION:
        case KEY_WORD_PROCEDURE:
        case KEY_WORD_TYPE:
            if (!g_instance->sql.use_bison_parser) {
                status = pl_parse_create(stmt, OG_TRUE, &word);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TRIGGER:
            if (!g_instance->sql.use_bison_parser) {
                status = pl_parse_create_trigger(stmt, OG_TRUE, &word);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_SYNONYM: {
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_synonym(stmt, SYNONYM_IS_REPLACE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }

        case KEY_WORD_DIRECTORY: {
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_directory(stmt, OG_TRUE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }

        case KEY_WORD_LIBRARY: {
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_library(stmt, OG_TRUE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }

        case KEY_WORD_PROFILE: {
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_profile(stmt, OG_TRUE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        }

        default: {
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "VIEW or PUBLIC expected but %s found", W2S(&word));
            return OG_ERROR;
        }
    }

    return status;
}

static status_t sql_create_public_lead(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch (word.id) {
        case KEY_WORD_SYNONYM: {
            status = sql_parse_create_synonym(stmt, SYNONYM_IS_PUBLIC);
            break;
        }

        default: {
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "SYNONYM expected but %s found", W2S(&word));
            return OG_ERROR;
        }
    }

    return status;
}

status_t sql_parse_create_directory(sql_stmt_t *stmt, bool32 is_replace)
{
    word_t word;
    status_t status;
    knl_directory_def_t *dir_def = NULL;
    lex_t *lex = stmt->session->lex;

#ifdef OG_RAC_ING
    if (IS_COORDINATOR && IS_APP_CONN(stmt->session)) {
        OG_THROW_ERROR(ERR_COORD_NOT_SUPPORT, "Create directory");
        return OG_ERROR;
    }
#endif

    status = sql_alloc_mem(stmt->context, sizeof(knl_directory_def_t), (void **)&dir_def);
    OG_RETURN_IFERR(status);

    dir_def->is_replace = is_replace;
    status = lex_expected_fetch_variant(lex, &word);
    OG_RETURN_IFERR(status);

    status = sql_copy_object_name(stmt->context, word.type, (text_t *)&word.text, &dir_def->name);
    OG_RETURN_IFERR(status);

    status = lex_expected_fetch_word(lex, "as");
    OG_RETURN_IFERR(status);

    status = lex_expected_fetch_string(lex, &word);
    OG_RETURN_IFERR(status);

    status = sql_copy_text(stmt->context, (text_t *)&word.text, &dir_def->path);
    OG_RETURN_IFERR(status);

    if (lex_expected_end(stmt->session->lex) != OG_SUCCESS) {
        return OG_ERROR;
    }

    stmt->context->entry = (void *)dir_def;
    stmt->context->type = OGSQL_TYPE_CREATE_DIRECTORY;
    return OG_SUCCESS;
}

status_t sql_parse_create(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    status = lex_fetch(stmt->session->lex, &word);
    OG_RETURN_IFERR(status);

    switch ((uint32)word.id) {
        case KEY_WORD_DATABASE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_create_database_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case RES_WORD_USER:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_user(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            OG_RETURN_IFERR(sql_clear_origin_sql_if_error(stmt, status));
            break;
        case KEY_WORD_ROLE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_role(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            OG_RETURN_IFERR(sql_clear_origin_sql_if_error(stmt, status));
            break;
        case KEY_WORD_TENANT:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_tenant(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TABLE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_table(stmt, OG_FALSE, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_INDEX:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_index(stmt, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_INDEXCLUSTER:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_indexes(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_SEQUENCE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_sequence(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TABLESPACE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_space(stmt, OG_FALSE, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TEMPORARY:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_create_temporary_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_GLOBAL:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_create_global_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_UNIQUE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_unique_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_UNDO:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_undo_space(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_VIEW:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_view(stmt, OG_FALSE, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_PROCEDURE:
        case KEY_WORD_FUNCTION:
        case KEY_WORD_PACKAGE:
        case KEY_WORD_TYPE:
            if (!g_instance->sql.use_bison_parser) {
                status = pl_parse_create(stmt, OG_FALSE, &word);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TRIGGER:
            if (!g_instance->sql.use_bison_parser) {
                status = pl_parse_create_trigger(stmt, OG_FALSE, &word);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_OR:
            status = sql_create_or_replace_lead(stmt);
            break;
        case KEY_WORD_PUBLIC:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_create_public_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_SYNONYM:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_synonym(stmt, SYNONYM_IS_NULL);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_PROFILE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_profile(stmt, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_DIRECTORY:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_directory(stmt, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_CTRLFILE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_ctrlfiles(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_LIBRARY:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create_library(stmt, OG_FALSE);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }

    return status;
}

static status_t sql_parse_alter_trigger(sql_stmt_t *stmt)
{
    status_t status;
    word_t word;
    lex_t *lex = stmt->session->lex;
    knl_alttrig_def_t *alttrig_def = NULL;
    bool32 result = OG_FALSE;

    lex->flags |= LEX_WITH_OWNER;
    stmt->context->type = OGSQL_TYPE_ALTER_TRIGGER;

    status = sql_alloc_mem(stmt->context, sizeof(knl_alttrig_def_t), (void **)&alttrig_def);
    OG_RETURN_IFERR(status);

    stmt->context->entry = (void *)alttrig_def;

    status = lex_expected_fetch_variant(lex, &word);
    OG_RETURN_IFERR(status);

    status = sql_convert_object_name(stmt, &word, &alttrig_def->user, NULL, &alttrig_def->name);
    OG_RETURN_IFERR(status);

    status = lex_try_fetch(stmt->session->lex, "ENABLE", &result);
    OG_RETURN_IFERR(status);

    if (result) {
        alttrig_def->enable = OG_TRUE;
        return OG_SUCCESS;
    }

    status = lex_try_fetch(stmt->session->lex, "DISABLE", &result);
    OG_RETURN_IFERR(status);

    if (result) {
        alttrig_def->enable = OG_FALSE;
        return lex_expected_end(stmt->session->lex);
    }

    OG_SRC_THROW_ERROR_EX(lex->loc, ERR_SQL_SYNTAX_ERROR, "ENABLE or DISABLE expected");
    return OG_ERROR;
}

status_t og_parse_alter_trigger(sql_stmt_t *stmt, knl_alttrig_def_t **def, name_with_owner *trigger_name, bool32 enable)
{
    knl_alttrig_def_t *alttrig_def = NULL;

    stmt->context->type = OGSQL_TYPE_ALTER_TRIGGER;
    OG_RETURN_IFERR(sql_alloc_mem(stmt->context, sizeof(knl_alttrig_def_t), (void **)def));
    alttrig_def = *def;

    if (trigger_name->owner.str == NULL) {
        cm_str2text(stmt->session->curr_schema, &alttrig_def->user);
    } else {
        alttrig_def->user = trigger_name->owner;
    }
    alttrig_def->name = trigger_name->name;
    alttrig_def->enable = enable;

    stmt->context->entry = alttrig_def;
    return OG_SUCCESS;
}


static status_t sql_parse_ddl_alter(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    status = lex_fetch(stmt->session->lex, &word);
    OG_RETURN_IFERR(status);

    switch ((uint32)word.id) {
        case KEY_WORD_TABLE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_table(stmt);
                OG_BREAK_IF_ERROR(status);
                status = sql_verify_alter_table(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_TABLESPACE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_space(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_DATABASE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_database_lead(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_SEQUENCE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_sequence(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_INDEX:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_index(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case RES_WORD_USER:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_user(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            OG_RETURN_IFERR(sql_clear_origin_sql_if_error(stmt, status));
            break;

        case KEY_WORD_TENANT:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_tenant(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_PROFILE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_profile(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_FUNCTION:
            if (!g_instance->sql.use_bison_parser) {
                OG_THROW_ERROR(ERR_CAPABILITY_NOT_SUPPORT, "alter function");
                status = OG_ERROR;
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        case KEY_WORD_TRIGGER:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_alter_trigger(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;

        /* fall-through */
        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }

    return status;
}

static status_t sql_drop_public_lead(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch (word.id) {
        case KEY_WORD_SYNONYM: {
            status = sql_parse_drop_synonym(stmt, SYNONYM_IS_PUBLIC);
            break;
        }
        default: {
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "SYNONYM expected but %s found", W2S(&word));
            return OG_ERROR;
        }
    }

    return status;
}

static status_t sql_parse_analyze(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;
    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch ((key_wid_t)word.id) {
        case KEY_WORD_TABLE:
            status = sql_parse_analyze_table(stmt);
            break;

        case KEY_WORD_INDEX:
            status = sql_parse_analyze_index(stmt);
            break;

        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            return OG_ERROR;
    }

    return status;
}

static status_t sql_parse_drop_directory(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;
    knl_drop_def_t *def = NULL;
    lex_t *lex = stmt->session->lex;

#ifdef OG_RAC_ING
    if (IS_COORDINATOR && IS_APP_CONN(stmt->session)) {
        OG_THROW_ERROR(ERR_COORD_NOT_SUPPORT, "Create directory");
        return OG_ERROR;
    }
#endif

    status = sql_alloc_mem(stmt->context, sizeof(knl_drop_def_t), (void **)&def);
    OG_RETURN_IFERR(status);

    status = lex_expected_fetch_variant(lex, &word);
    OG_RETURN_IFERR(status);

    status = sql_copy_object_name(stmt->context, word.type, (text_t *)&word.text, &def->name);
    OG_RETURN_IFERR(status);

    if (lex_expected_end(stmt->session->lex) != OG_SUCCESS) {
        return OG_ERROR;
    }

    stmt->context->entry = (void *)def;
    stmt->context->type = OGSQL_TYPE_DROP_DIRECTORY;
    return lex_expected_end(lex);
}

status_t sql_parse_drop_library(sql_stmt_t *stmt)
{
    knl_drop_def_t *def = NULL;
    lex_t *lex = stmt->session->lex;
    lex->flags = LEX_WITH_OWNER;

    stmt->context->type = OGSQL_TYPE_DROP_LIBRARY;

    if (sql_alloc_mem(stmt->context, sizeof(knl_drop_def_t), (void **)&def) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (sql_parse_drop_object(stmt, def) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (lex_expected_end(lex) != OG_SUCCESS) {
        return OG_ERROR;
    }

    stmt->context->entry = def;
    return OG_SUCCESS;
}

status_t sql_parse_drop(sql_stmt_t *sql_stmt)
{
    word_t word;
    status_t status;

    status = lex_fetch(sql_stmt->session->lex, &word);
    OG_RETURN_IFERR(status);

    switch ((uint32)word.id) {
        case KEY_WORD_TABLE:
            status = sql_parse_drop_table(sql_stmt, OG_FALSE);
            break;
        case KEY_WORD_INDEX:
            status = sql_parse_drop_index(sql_stmt);
            break;
        case KEY_WORD_SEQUENCE:
            status = sql_parse_drop_sequence(sql_stmt);
            break;
        case KEY_WORD_TABLESPACE:
            status = sql_parse_drop_tablespace(sql_stmt);
            break;
        case KEY_WORD_TEMPORARY:
            status = sql_parse_drop_temporary_lead(sql_stmt);
            break;
        case KEY_WORD_VIEW:
            status = sql_parse_drop_view(sql_stmt);
            break;
        case RES_WORD_USER:
            status = sql_parse_drop_user(sql_stmt);
            break;
        case KEY_WORD_TENANT:
            status = sql_parse_drop_tenant(sql_stmt);
            break;
        case KEY_WORD_PUBLIC:
            status = sql_drop_public_lead(sql_stmt);
            break;
        case KEY_WORD_ROLE:
            status = sql_parse_drop_role(sql_stmt);
            break;
        case KEY_WORD_PROFILE:
            status = sql_parse_drop_profile(sql_stmt);
            break;
        case KEY_WORD_DIRECTORY:
            status = sql_parse_drop_directory(sql_stmt);
            break;
        case KEY_WORD_PROCEDURE:
        case KEY_WORD_FUNCTION:
            status = pl_parse_drop_procedure(sql_stmt, &word);
            break;
        case KEY_WORD_TRIGGER:
            status = pl_parse_drop_trigger(sql_stmt, &word);
            break;
        case KEY_WORD_PACKAGE:
            status = pl_parse_drop_package(sql_stmt, &word);
            break;
        case KEY_WORD_TYPE:
            status = pl_parse_drop_type(sql_stmt, &word);
            break;
        case KEY_WORD_SYNONYM:
            status = sql_parse_drop_synonym(sql_stmt, SYNONYM_IS_NULL);
            break;
        case KEY_WORD_DATABASE:
            status = sql_drop_database_lead(sql_stmt);
            break;
        case KEY_WORD_LIBRARY:
            status = sql_parse_drop_library(sql_stmt);
            break;

        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }

    return status;
}

static status_t sql_parse_purge(sql_stmt_t *stmt)
{
    word_t word;
    lex_t *lex = stmt->session->lex;
    knl_purge_def_t *purge_def = NULL;
    status_t status;

    stmt->context->type = OGSQL_TYPE_PURGE;

    if (sql_alloc_mem(stmt->context, sizeof(knl_purge_def_t), (void **)&purge_def) != OG_SUCCESS) {
        return OG_ERROR;
    }

    purge_def->part_name.len = 0;
    purge_def->part_name.str = NULL;

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch ((key_wid_t)word.id) {
        case KEY_WORD_TABLE:
            status = sql_parse_purge_table(stmt, purge_def);
            break;
        case KEY_WORD_INDEX:
            status = sql_parse_purge_index(stmt, purge_def);
            break;
        case KEY_WORD_PARTITION:
            status = sql_parse_purge_partition(stmt, purge_def);
            break;
        case KEY_WORD_TABLESPACE:
            status = sql_parse_purge_tablespace(stmt, purge_def);
            break;
        case KEY_WORD_RECYCLEBIN:
            purge_def->type = PURGE_RECYCLEBIN;
            stmt->context->entry = purge_def;
            status = lex_expected_end(lex);
            break;
        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }

    return status;
}

status_t sql_parse_truncate(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch ((key_wid_t)word.id) {
        case KEY_WORD_TABLE:
            status = sql_parse_truncate_table(stmt);
            break;
        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }
    return status;
}

status_t sql_parse_flashback(sql_stmt_t *stmt)
{
    word_t word;
    status_t status;

    if (lex_fetch(stmt->session->lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch ((key_wid_t)word.id) {
        case KEY_WORD_TABLE:
            status = sql_parse_flashback_table(stmt);
            break;
        default:
            OG_SRC_THROW_ERROR_EX(word.text.loc, ERR_SQL_SYNTAX_ERROR, "object type expected but %s found", W2S(&word));
            status = OG_ERROR;
            break;
    }
    return status;
}
static void sql_init_grant_def(sql_stmt_t *stmt, knl_grant_def_t *grant_def)
{
    cm_galist_init(&grant_def->privs, stmt->context, sql_alloc_mem);
    cm_galist_init(&grant_def->columns, stmt->context, sql_alloc_mem);
    cm_galist_init(&grant_def->grantees, stmt->context, sql_alloc_mem);
    cm_galist_init(&grant_def->privs_list, stmt->context, sql_alloc_mem);
    cm_galist_init(&grant_def->grantee_list, stmt->context, sql_alloc_mem);
    grant_def->grant_uid = stmt->session->curr_schema_id;
    return;
}

static status_t sql_parse_grant(sql_stmt_t *stmt)
{
    lex_t *lex = stmt->session->lex;
    knl_session_t *se = &stmt->session->knl_session;
    status_t status;
    knl_grant_def_t *def = NULL;
    bool32 dire_priv = OG_FALSE;
    stmt->session->sql_audit.audit_type = SQL_AUDIT_DCL;
    stmt->context->type = OGSQL_TYPE_GRANT;

    if (knl_ddl_enabled(se, OG_TRUE) != OG_SUCCESS) {
        return OG_ERROR;
    }
    status = sql_alloc_mem(stmt->context, sizeof(knl_grant_def_t), (void **)&def);
    OG_RETURN_IFERR(status);

    sql_init_grant_def(stmt, def);
    status = sql_parse_grant_privs(stmt, def);
    OG_RETURN_IFERR(status);

    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(sql_check_dir_priv(&def->privs, &dire_priv));
        if (dire_priv) {
            def->objtype = OBJ_TYPE_DIRECTORY;
        }
        OG_BIT_SET(lex->flags, LEX_WITH_OWNER);
        OG_RETURN_IFERR(sql_parse_grant_objprivs_def(stmt, lex, def));
        OG_BIT_RESET(lex->flags, LEX_WITH_OWNER);
    }

    if (def->priv_type == PRIV_TYPE_USER_PRIV) {
        OG_RETURN_IFERR(sql_check_user_privileges(&def->privs));
        def->objtype = OBJ_TYPE_USER;
        OG_BIT_SET(lex->flags, LEX_WITH_OWNER);
        OG_RETURN_IFERR(sql_parse_grant_objprivs_def(stmt, lex, def));
        OG_BIT_RESET(lex->flags, LEX_WITH_OWNER);
    }

    OG_RETURN_IFERR(sql_parse_grantee_def(stmt, lex, def));
    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        // User who has GRANT ANY OBJECT PRIVILEGE  can't grant privilege to himself except his owner object
        if (sql_check_obj_owner(lex, &stmt->session->curr_user, &def->grantees) != OG_SUCCESS) {
            if (cm_compare_text(&stmt->session->curr_user, &def->schema) != 0) {
                return OG_ERROR;
            } else {
                // sql_check_obj_owner may set error code, if object owner is current user we must clear error code.
                cm_reset_error();
            }
        }
    }
    /* check privilege's type */
    status = sql_check_privs_type(stmt, &def->privs, def->priv_type, def->objtype, &def->type_name);
    OG_RETURN_IFERR(status);

    stmt->context->entry = (void *)def;
    return OG_SUCCESS;
}

static status_t sql_parse_revoke(sql_stmt_t *stmt)
{
    lex_t *lex = stmt->session->lex;
    knl_session_t *se = &stmt->session->knl_session;
    status_t status;
    knl_revoke_def_t *def = NULL;
    bool32 dire_priv = OG_FALSE;
    stmt->session->sql_audit.audit_type = SQL_AUDIT_DCL;
    stmt->context->type = OGSQL_TYPE_REVOKE;

    if (knl_ddl_enabled(se, OG_TRUE) != OG_SUCCESS) {
        return OG_ERROR;
    }
    status = sql_alloc_mem(stmt->context, sizeof(knl_revoke_def_t), (void **)&def);
    OG_RETURN_IFERR(status);

    cm_galist_init(&def->privs, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->revokees, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->privs_list, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->revokee_list, stmt->context, sql_alloc_mem);

    status = sql_parse_revoke_privs(stmt, def);
    OG_RETURN_IFERR(status);

    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(sql_check_dir_priv(&def->privs, &dire_priv));
        if (dire_priv) {
            def->objtype = OBJ_TYPE_DIRECTORY;
        }
        OG_BIT_SET(lex->flags, LEX_WITH_OWNER);
        OG_RETURN_IFERR(sql_parse_revoke_objprivs_def(stmt, lex, def));
        OG_BIT_RESET(lex->flags, LEX_WITH_OWNER);
    }

    if (def->priv_type == PRIV_TYPE_USER_PRIV) {
        OG_RETURN_IFERR(sql_check_user_privileges(&def->privs));
        def->objtype = OBJ_TYPE_USER;
        OG_BIT_SET(lex->flags, LEX_WITH_OWNER);
        OG_RETURN_IFERR(sql_parse_revoke_objprivs_def(stmt, lex, def));
        OG_BIT_RESET(lex->flags, LEX_WITH_OWNER);
    }

    OG_RETURN_IFERR(sql_parse_revokee_def(stmt, lex, def));

    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(sql_check_obj_owner(lex, &stmt->session->curr_user, &def->revokees));
        OG_RETURN_IFERR(sql_check_obj_schema(lex, &def->schema, &def->revokees));
    }

    /* check privilege's type */
    status = sql_check_privs_type(stmt, &def->privs, def->priv_type, def->objtype, &def->type_name);
    OG_RETURN_IFERR(status);

    stmt->context->entry = (void *)def;
    return OG_SUCCESS;
}

static status_t sql_parse_comment(sql_stmt_t *stmt)
{
    word_t word;
    lex_t *lex = stmt->session->lex;
    status_t status;

    stmt->context->type = OGSQL_TYPE_COMMENT;

    if (lex_expected_fetch_word(lex, "ON") != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (lex_fetch(lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    switch (word.id) {
        case KEY_WORD_TABLE:
        case KEY_WORD_COLUMN:
            status = sql_parse_comment_table(stmt, (key_wid_t)word.id);
            break;
        default:
            OG_THROW_ERROR_EX(ERR_SQL_SYNTAX_ERROR, "object expected but %s found", W2S(&word));
            return OG_ERROR;
    }

    return status;
}

status_t sql_parse_ddl(sql_stmt_t *stmt, word_t *leader_word)
{
    status_t status;
    key_wid_t key_wid = leader_word->id;
    text_t origin_sql = sql_current_parse_text(stmt)->value;
    stmt->session->sql_audit.audit_type = SQL_AUDIT_DDL;
    OG_RETURN_IFERR(sql_alloc_context(stmt));
    OG_RETURN_IFERR(sql_create_list(stmt, &stmt->context->ref_objects));

    switch (key_wid) {
        case KEY_WORD_CREATE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_create(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_DROP:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_drop(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_TRUNCATE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_truncate(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_FLASHBACK:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_flashback(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_PURGE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_purge(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_COMMENT:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_comment(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_GRANT:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_grant(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_REVOKE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_revoke(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        case KEY_WORD_ANALYZE:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_analyze(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
        default:
            if (!g_instance->sql.use_bison_parser) {
                status = sql_parse_ddl_alter(stmt);
            } else {
                status = raw_parser(stmt, sql_current_parse_text(stmt), &stmt->context->entry);
            }
            break;
    }

    // write ddl sql into context, exclude operate pwd ddl
    if (!SQL_OPT_PWD_DDL_TYPE(stmt->context->type)) {
        OG_RETURN_IFERR(ogx_write_text(&stmt->context->ctrl, &origin_sql));
        stmt->context->ctrl.hash_value = cm_hash_text(&origin_sql, INFINITE_HASH_RANGE);
    }

    return status;
}

status_t sql_parse_scope_clause_inner(knl_alter_sys_def_t *def, lex_t *lex, bool32 force)
{
    bool32 result = OG_FALSE;
    uint32 match_id;
    status_t status;

    // if already parsed scope clause, must return
    if (def->scope >= CONFIG_SCOPE_MEMORY) {
        return OG_SUCCESS;
    }

    if (force) {
        status = lex_expected_fetch_word(lex, "scope");
        OG_RETURN_IFERR(status);
    } else {
        status = lex_try_fetch(lex, "scope", &result);
        OG_RETURN_IFERR(status);
        if (!result) {
            def->scope = CONFIG_SCOPE_BOTH;
            return OG_SUCCESS;
        }
    }

    status = lex_expected_fetch_word(lex, "=");
    OG_RETURN_IFERR(status);

    status = lex_expected_fetch_1of3(lex, "memory", "pfile", "both", &match_id);
    OG_RETURN_IFERR(status);

    if (match_id == LEX_MATCH_FIRST_WORD) {
        def->scope = CONFIG_SCOPE_MEMORY;
    } else if (match_id == LEX_MATCH_SECOND_WORD) {
        def->scope = CONFIG_SCOPE_DISK;
    } else {
        def->scope = CONFIG_SCOPE_BOTH;
    }

    return OG_SUCCESS;
}

status_t sql_parse_expected_scope_clause(knl_alter_sys_def_t *def, lex_t *lex)
{
    return sql_parse_scope_clause_inner(def, lex, OG_TRUE);
}

status_t sql_parse_scope_clause(knl_alter_sys_def_t *def, lex_t *lex)
{
    return sql_parse_scope_clause_inner(def, lex, OG_FALSE);
}

#ifndef WIN32
status_t sql_verify_lib_host(char *realfile)
{
    char file_host[OG_FILE_NAME_BUFFER_SIZE];
    if (cm_get_file_host_name(realfile, file_host) != OG_SUCCESS) {
        return OG_ERROR;
    }
    if (!cm_str_equal(file_host, cm_sys_user_name())) {
        return OG_ERROR;
    }
    return OG_SUCCESS;
}
#endif

static status_t sql_verify_library(sql_stmt_t *stmt, pl_library_def_t *def)
{
    if (sql_check_priv(stmt, &stmt->session->curr_user, &def->owner, CREATE_LIBRARY, CREATE_ANY_LIBRARY) !=
        OG_SUCCESS) {
        OG_THROW_ERROR(ERR_INSUFFICIENT_PRIV);
        return OG_ERROR;
    }

    if (def->path.len >= OG_FILE_NAME_BUFFER_SIZE) {
        OG_THROW_ERROR(ERR_FILE_PATH_TOO_LONG, OG_FILE_NAME_BUFFER_SIZE - 1);
        return OG_ERROR;
    }
    char lib_path[OG_FILE_NAME_BUFFER_SIZE];
    char realfile[OG_FILE_NAME_BUFFER_SIZE];

    OG_RETURN_IFERR(cm_text2str(&def->path, lib_path, OG_FILE_NAME_BUFFER_SIZE));
    OG_RETURN_IFERR(realpath_file((const char *)lib_path, realfile, OG_FILE_NAME_BUFFER_SIZE));
    if (strlen(realfile) == 0 || !cm_file_exist(realfile)) {
        OG_THROW_ERROR(ERR_FILE_NOT_EXIST, "library", realfile);
        return OG_ERROR;
    }

    if (cm_access_file(realfile, X_OK) != OG_SUCCESS) {
        OG_THROW_ERROR(ERR_EXECUTE_FILE, realfile, cm_get_os_error());
        return OG_ERROR;
    }
    void *lib_handle = NULL;
#ifndef WIN32
    if (cm_open_dl(&lib_handle, realfile) != OG_SUCCESS) {
        OG_THROW_ERROR(ERR_LOAD_LIBRARY, realfile, cm_get_os_error());
        return OG_ERROR;
    }
    cm_close_dl(lib_handle);
    if (cm_verify_file_host(realfile) != OG_SUCCESS) {
        OG_THROW_ERROR(ERR_FILE_EXEC_PRIV, realfile);
        return OG_ERROR;
    }
#endif

#ifdef WIN32
    char *leaf_name = strrchr(realfile, '\\');
#else
    char *leaf_name = strrchr(realfile, '/');
#endif

    if (leaf_name == NULL || strlen(leaf_name) == 1) {
        OG_THROW_ERROR(ERR_TEXT_FORMAT_ERROR, "dynamic link library");
        return OG_ERROR;
    }
    leaf_name = leaf_name + 1;
    return sql_copy_str_safe(stmt->context, leaf_name, (uint32)strlen(leaf_name), &def->leaf_name);
}

status_t sql_parse_create_library(sql_stmt_t *stmt, bool32 is_replace)
{
    word_t word;
    pl_library_def_t *def = NULL;
    lex_t *lex = stmt->session->lex;
    lex->flags = LEX_WITH_OWNER;

    stmt->context->type = OGSQL_TYPE_CREATE_LIBRARY;
    if (sql_alloc_mem(stmt->context, sizeof(pl_library_def_t), (pointer_t *)&def) != OG_SUCCESS) {
        return OG_ERROR;
    }

    def->is_replace = is_replace;

    if (lex_expected_fetch_variant(lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (sql_convert_object_name(stmt, &word, &def->owner, NULL, &def->name) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (lex_fetch(lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (word.id != KEY_WORD_AS && word.id != KEY_WORD_IS) {
        OG_SRC_THROW_ERROR_EX(LEX_LOC, ERR_SQL_SYNTAX_ERROR, "expected as or is but %s found", W2S(&word));
        return OG_ERROR;
    }

    if (lex_expected_fetch_string(lex, &word) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (sql_copy_text(stmt->context, (text_t *)&word.text, &def->path) != OG_SUCCESS) {
        return OG_ERROR;
    }

    if (sql_verify_library(stmt, def)) {
        return OG_ERROR;
    }

    stmt->context->entry = def;

    return lex_expected_end(lex);
}

status_t sql_verify_als_cpu_inf_str(void *se, void *lex, void *def)
{
    return OG_SUCCESS;
}

status_t sql_verify_als_res_recycle_ratio(void *se, void *lex, void *def)
{
    uint32 num;
    if (sql_verify_uint32(lex, def, &num) != OG_SUCCESS) {
        return OG_ERROR;
    }
    if (num < OG_MIN_RES_RECYCLE_RATIO) {
        OG_THROW_ERROR(ERR_PARAMETER_TOO_SMALL, "RES_RECYCLE_RATIO", (int64)OG_MIN_RES_RECYCLE_RATIO);
        return OG_ERROR;
    }
    if (num > OG_MAX_RES_RECYCLE_RATIO) {
        OG_THROW_ERROR(ERR_PARAMETER_TOO_LARGE, "RES_RECYCLE_RATIO", (int64)OG_MAX_RES_RECYCLE_RATIO);
        return OG_ERROR;
    }
    return OG_SUCCESS;
}

status_t sql_verify_als_create_index_parallelism(void *se, void *lex, void *def)
{
    uint32 num;
    if (sql_verify_uint32(lex, def, &num) != OG_SUCCESS) {
        return OG_ERROR;
    }
    if (num < OG_MIN_CREATE_INDEX_PARALLELISM) {
        OG_THROW_ERROR(ERR_PARAMETER_TOO_SMALL, "CREATE_INDEX_PARALLELISM", (int64)OG_MIN_CREATE_INDEX_PARALLELISM);
        return OG_ERROR;
    }
    if (num > OG_MAX_CREATE_INDEX_PARALLELISM) {
        OG_THROW_ERROR(ERR_PARAMETER_TOO_LARGE, "CREATE_INDEX_PARALLELISM", (int64)OG_MAX_CREATE_INDEX_PARALLELISM);
        return OG_ERROR;
    }
    return OG_SUCCESS;
}

status_t og_parse_create_directory(sql_stmt_t *stmt, knl_directory_def_t **def, char *dir_name, char *path,
    bool32 is_replace)
{
    status_t status;
    knl_directory_def_t *dir_def = NULL;
    text_t path_text;

#ifdef OG_RAC_ING
    if (IS_COORDINATOR && IS_APP_CONN(stmt->session)) {
        OG_THROW_ERROR(ERR_COORD_NOT_SUPPORT, "Create directory");
        return OG_ERROR;
    }
#endif

    stmt->context->type = OGSQL_TYPE_CREATE_DIRECTORY;

    status = sql_alloc_mem(stmt->context, sizeof(knl_directory_def_t), (void **)def);
    OG_RETURN_IFERR(status);
    dir_def = *def;

    dir_def->is_replace = is_replace;

    dir_def->name.str = dir_name;
    dir_def->name.len = strlen(dir_name);

    cm_str2text(path, &path_text);
    status = sql_copy_text(stmt->context, &path_text, &dir_def->path);
    OG_RETURN_IFERR(status);

    return OG_SUCCESS;
}

status_t og_parse_create_library(sql_stmt_t *stmt, pl_library_def_t **def, name_with_owner *lib_name, char *path,
    bool32 is_replace)
{
    status_t status;
    pl_library_def_t *lib_def = NULL;
    text_t path_text;

    stmt->context->type = OGSQL_TYPE_CREATE_LIBRARY;

    status = sql_alloc_mem(stmt->context, sizeof(pl_library_def_t), (void **)def);
    OG_RETURN_IFERR(status);
    lib_def = *def;

    lib_def->is_replace = is_replace;

    // Parse library name
    if (lib_name->owner.len > 0) {
        lib_def->owner = lib_name->owner;
    } else {
        cm_str2text(stmt->session->curr_schema, &lib_def->owner);
    }
    lib_def->name = lib_name->name;

    cm_str2text(path, &path_text);
    status = sql_copy_text(stmt->context, &path_text, &lib_def->path);
    OG_RETURN_IFERR(status);

    if (sql_verify_library(stmt, lib_def) != OG_SUCCESS) {
        return OG_ERROR;
    }

    return OG_SUCCESS;
}

status_t og_parse_grant(sql_stmt_t *stmt, knl_grant_def_t **grant_def, priv_type_def priv_type, galist_t *priv_list,
    object_type_t obj_type, name_with_owner *obj_name, galist_t *grantee_list, bool with_opt)
{
    knl_session_t *se = &stmt->session->knl_session;
    status_t status;
    knl_grant_def_t *def = NULL;
    bool32 dire_priv = OG_FALSE;
    stmt->session->sql_audit.audit_type = SQL_AUDIT_DCL;
    stmt->context->type = OGSQL_TYPE_GRANT;

    if (knl_ddl_enabled(se, OG_TRUE) != OG_SUCCESS) {
        return OG_ERROR;
    }

    status = sql_alloc_mem(stmt->context, sizeof(knl_grant_def_t), (void **)grant_def);
    OG_RETURN_IFERR(status);
    def = *grant_def;

    sql_init_grant_def(stmt, def);

    def->priv_type = priv_type;
    cm_galist_copy(&def->privs, priv_list);
    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(sql_check_dir_priv(&def->privs, &dire_priv));
        if (dire_priv) {
            def->objtype = OBJ_TYPE_DIRECTORY;
        }
        OG_RETURN_IFERR(og_parse_object_info(stmt, &def->schema, &def->objname, &def->objtype, &def->type_name,
            obj_type, obj_name));
    }
    if (def->priv_type == PRIV_TYPE_USER_PRIV) {
        OG_RETURN_IFERR(sql_check_user_privileges(&def->privs));
        def->objtype = OBJ_TYPE_USER;
        OG_RETURN_IFERR(og_parse_user_priv_info(stmt, &def->objname, &def->type_name, obj_name));
    }

    OG_RETURN_IFERR(og_parse_grantee_def(stmt, def, grantee_list, with_opt));
    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        // User who has GRANT ANY OBJECT PRIVILEGE  can't grant privilege to himself except his owner object
        if (og_check_obj_owner(&stmt->session->curr_user, &def->grantees) != OG_SUCCESS) {
            if (cm_compare_text(&stmt->session->curr_user, &def->schema) != 0) {
                return OG_ERROR;
            } else {
                // og_check_obj_owner may set error code, if object owner is current user we must clear error code.
                cm_reset_error();
            }
        }
    }
    /* check privilege's type */
    status = sql_check_privs_type(stmt, &def->privs, def->priv_type, def->objtype, &def->type_name);
    OG_RETURN_IFERR(status);

    return OG_SUCCESS;
}

status_t og_parse_revoke(sql_stmt_t *stmt, knl_revoke_def_t **revoke_def, priv_type_def priv_type,
    galist_t *priv_list, object_type_t obj_type, name_with_owner *obj_name, galist_t *revokee_list,
    bool cascade_opt)
{
    knl_session_t *se = &stmt->session->knl_session;
    status_t status;
    knl_revoke_def_t *def = NULL;
    bool32 dire_priv = OG_FALSE;
    stmt->session->sql_audit.audit_type = SQL_AUDIT_DCL;
    stmt->context->type = OGSQL_TYPE_REVOKE;

    if (knl_ddl_enabled(se, OG_TRUE) != OG_SUCCESS) {
        return OG_ERROR;
    }

    status = sql_alloc_mem(stmt->context, sizeof(knl_revoke_def_t), (void **)revoke_def);
    OG_RETURN_IFERR(status);
    def = *revoke_def;

    cm_galist_init(&def->privs, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->revokees, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->privs_list, stmt->context, sql_alloc_mem);
    cm_galist_init(&def->revokee_list, stmt->context, sql_alloc_mem);

    def->priv_type = priv_type;
    cm_galist_copy(&def->privs, priv_list);

    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(sql_check_dir_priv(&def->privs, &dire_priv));
        if (dire_priv) {
            def->objtype = OBJ_TYPE_DIRECTORY;
        }
        OG_RETURN_IFERR(og_parse_object_info(stmt, &def->schema, &def->objname, &def->objtype, &def->type_name,
            obj_type, obj_name));
    }

    if (def->priv_type == PRIV_TYPE_USER_PRIV) {
        OG_RETURN_IFERR(sql_check_user_privileges(&def->privs));
        def->objtype = OBJ_TYPE_USER;
        OG_RETURN_IFERR(og_parse_user_priv_info(stmt, &def->objname, &def->type_name, obj_name));
    }

    OG_RETURN_IFERR(og_parse_revokee_def(stmt, def, revokee_list, cascade_opt));

    if (def->priv_type == PRIV_TYPE_OBJ_PRIV) {
        OG_RETURN_IFERR(og_check_obj_owner(&stmt->session->curr_user, &def->revokees));
        OG_RETURN_IFERR(og_check_obj_schema(&def->schema, &def->revokees));
    }

    status = sql_check_privs_type(stmt, &def->privs, def->priv_type, def->objtype, &def->type_name);
    OG_RETURN_IFERR(status);

    return OG_SUCCESS;
}

status_t og_parse_purge(sql_stmt_t *stmt, knl_purge_def_t **purge_def, purge_type_t purge_type,
    name_with_owner *name_owner, char *single_name)
{
    knl_purge_def_t *def = NULL;

    stmt->context->type = OGSQL_TYPE_PURGE;

    if (sql_alloc_mem(stmt->context, sizeof(knl_purge_def_t), (void **)purge_def) != OG_SUCCESS) {
        return OG_ERROR;
    }
    def = *purge_def;
    def->part_name.len = 0;
    def->part_name.str = NULL;
    def->type = purge_type;

    if (purge_type == PURGE_RECYCLEBIN) {
        stmt->context->entry = def;
        return OG_SUCCESS;
    }

    switch (purge_type) {
        case PURGE_TABLE:
        case PURGE_INDEX:
            if (name_owner->owner.len > 0) {
                def->owner = name_owner->owner;
            } else {
                cm_str2text(stmt->session->curr_schema, &def->owner);
            }
            def->name = name_owner->name;
            break;
        case PURGE_PART:
            if (name_owner->owner.len > 0) {
                def->owner = name_owner->owner;
            } else {
                cm_str2text(stmt->session->curr_schema, &def->owner);
            }
            def->name = name_owner->name;
            cm_str2text(single_name, &def->part_name);
            break;
        case PURGE_TABLE_OBJECT:
        case PURGE_INDEX_OBJECT:
            cm_str2text(stmt->session->curr_schema, &def->owner);
            cm_str2text(single_name, &def->name);
            break;
        case PURGE_PART_OBJECT:
        case PURGE_TABLESPACE:
            cm_str2text(single_name, &def->name);
            break;
        default:
            break;
    }

    return OG_SUCCESS;
}

#ifdef __cplusplus
}
#endif